# Sending data to existing elasticsearch index through Watcher

**URL:** <https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368>\
**Category:** Elasticsearch\
**Tags:** elastic-stack-alerting\
**Created:** [April 29, 2020, 12:44pm UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368 "2020-04-29T12:44:20Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![shubhamshd99](https://avatars.discourse-cdn.com/v4/letter/s/8edcca/32.png) [@shubhamshd99](https://discuss.elastic.co/u/shubhamshd99)\
**Post date:** [April 29, 2020, 12:44pm UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/1 "2020-04-29T12:44:20Z")

</div>

Hi Guys,

I'm trying to send the output of watcher to existing index on elasticsearch, but it throws an error with following description -\> **"java.lang.IllegalArgumentException: Field [\_source] is defined both as an object and a field"**

Basically after performing some modification on the data of one index using transform, the result is being sent to another existing index!!!

```
"transform":{
"script":
"""
def doc_count = ctx.payload.hits.total;
def unique_hits = [];
def unique_sources = ctx.payload.aggregations.unique_sources.buckets.stream().map(hit->hit.key).collect(Collectors.toList());
def hit;
for(def source : unique_sources){
  for(def i=0; i< doc_count; i++){
    hit = ctx.payload.hits.hits[i];
    
    
    if (hit.containsKey('_index'))
    {
      hit.remove('_index');
    }
    
    if(source==hit._source.source){
      hit['_index'] = "index-for-action_index-test";
      unique_hits.add(hit);
      break;
    }
  }
}

return ['_doc' : unique_hits];
"""

```

},

And it's not necessarily related to existing index, I also tried to send output to whole new index, but it gave me same error.

Actually, at first, I faced another issue -\> **"could not execute action [index\_payload] of watch [_inlined_]. [ctx.payload.\_index] or [ctx.payload.\_doc.\_index] were set together with action [index] field. Only set one of them"** and after following [this](https://discuss.elastic.co/t/transform-single-hit-into-multiple-documents-with-index-action/145010/4) topic, I arrived at the current issue ☹

---

<div class="post-metadata">

**Author:** ![Luca\_Belluccini](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/luca_belluccini/32/33239_2.png) [@Luca\_Belluccini](https://discuss.elastic.co/u/Luca_Belluccini)\
**Post date:** [April 29, 2020, 12:57pm UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/2 "2020-04-29T12:57:11Z")

</div>

Hello @shubhamshd99

1. Would it be possible to share the full Watch body?

2. As we do not know the data, would it be possible to replace the `index` action with:

Thank you in advance.

---

<div class="post-metadata">

**Author:** ![shubhamshd99](https://avatars.discourse-cdn.com/v4/letter/s/8edcca/32.png) [@shubhamshd99](https://discuss.elastic.co/u/shubhamshd99)\
**Post date:** [April 30, 2020, 10:21am UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/3 "2020-04-30T10:21:39Z")

</div>

Hi @Luca_Belluccini  
Please find full body of the watcher and result of logging and action index as well!!!

**Watcher** :-

```
{
  "trigger": {
    "schedule": {
      "interval" : "5000m"
    }
  },
  "input": {
    "search": {
      "request": {
        "search_type": "query_then_fetch",
        "indices": [
        "cft_portal_web_access-2020.03.19"
        ],
        "rest_total_hits_as_int": true,
        "body": {
          "size" : 1000,
          "aggregations" : {
            "unique_sources" : {
              "terms" : {
                "field" : "source",
                "size" : 100
              }
            }
          }
        }
      }
  }
  },
  "condition": {
    "compare": {
      "ctx.payload.hits.total": {
        "gt": 0
      }
    }
  },
  "transform":{
    "script":
    """
    def doc_count = ctx.payload.hits.total;
    def unique_hits = [];
    def unique_sources = ctx.payload.aggregations.unique_sources.buckets.stream().map(hit->hit.key).collect(Collectors.toList());
    

    def hit;
    for(def source : unique_sources){
      for(def i=0; i< doc_count; i++){
        //hit = ctx.payload.hits.hits[i];
        
        if(source==ctx.payload.hits.hits[i]._source.source){
          //.put('_index', 'index-for-action_index-test');
          hit = ctx.payload.hits.hits[i];
          hit.remove('_index');
          unique_hits.add(hit);
          break;
        }
      }
    }
    
    
    //return unique_hits;
    return ['_doc' : unique_hits];
    """
  },
  "actions": {
    "index_payload" : {
      "index" : {
        "index" : "index-for-action_index-test"
      }
    },
    "logging": { "logging" : { "text": "{{ctx}}" } }
  }
}

```

**Index action :**

```
{
          "id" : "index_payload",
          "type" : "index",
          "status" : "failure",
          "index" : {
            "response" : [
              {
                "failed" : true,
                "message" : "java.lang.IllegalArgumentException: Field [_source] is defined both as an object and a field.",
                "id" : "mrEh8nABOjhT50Px45Xt",
                "type" : "_doc",
                "index" : "index-for-action_index-test"
              },
              {
                "failed" : true,
                "message" : "java.lang.IllegalArgumentException: Field [_source] is defined both as an object and a field.",
                "id" : "q7Eh8nABOjhT50Px45bt",
                "type" : "_doc",
                "index" : "index-for-action_index-test"
              }
           ] 
        }
}

```

**Logging action :**  
`{metadata={app=wps_bn_my, description=PI-001 : Monitor Web Access – Interal, env=non-prod}, watch_id=index-action, payload={_doc=[{_source={host=ldnpsr37510.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/now.gem.ldnpsr37510.10643.swebwpsp.web.access.ssl}, _score=1.0}, {_source={host=ldnpsr37511.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/now.gem.ldnpsr37511.10643.swebwpsp.web.access.ssl}, _score=1.0}, {_source={host=nykpsr10615.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/now.cf.nykpsr10615.10100.swebwpsp.web.access}, _score=1.0}, {_source={host=nykpsr10614.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/now.cf.nykpsr10614.10643.swebwpsp.web.access.ssl}, _score=1.0}, {_source={host=nykpsr10593.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/now.pi.nykpsr10593.10643.swebwpsp.web.access.ssl}, _score=1.0}, {_source={host=ldnpsr37587.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/nowext.gem.ldnpsr37587.10100.swebwpsp.web.access_log}, _score=1.0}, {_source={host=nykpsr10591.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/now.pi.nykpsr10591.10643.swebwpsp.web.access.ssl}, _score=1.0}, {_source={host=ldnpsr37585.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/nowext.gem.ldnpsr37585.10100.swebwpsp.web.access_log}, _score=1.0}, {_source={host=ldnpsr37510.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/nowws.barclays.com/logs/nowws.gem.ldnpsr37510.10101.swebwpsp.web.access}, _score=1.0}, {_source={host=ldnpsr37511.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/nowws.barclays.com/logs/nowws.gem.ldnpsr37511.10101.swebwpsp.web.access}, _score=1.0}, {_source={host=ldnpsr37511.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/now.gem.ldnpsr37511.10100.swebwpsp.web.access}, _score=1.0}, {_source={host=sgppsr00674.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/now.barclays.com/logs/nowext.sgp.sgppsr00674.10100.swebwpsp.web.access}, _score=1.0}, {_source={host=nykpsr10614.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/nowws.barclays.com/logs/nowws.cf.nykpsr10614.10101.swebwpsp.web.access}, _score=1.0}, {_source={host=nykpsr10615.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/nowws.barclays.com/logs/nowws.cf.nykpsr10615.10101.swebwpsp.web.access}, _score=1.0}, {_source={host=nykpsr10591.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/nowws.barclays.com/logs/nowws.pi.nykpsr10591.10101.swebwpsp.web.access}, _score=1.0}, {_source={host=nykpsr10593.intranet.barcapint.com, sourcetype=wps.portal.web.access, source=/apps/apache/nowws.barclays.com/logs/nowws.pi.nykpsr10593.10101.swebwpsp.web.access}, _score=1.0}]}, id=index-action_e25ae02a-3b9d-4738-8b49-5f04f606fe77-2020-04-30T10:07:25.342787Z, trigger={triggered_time=2020-04-30T10:07:25.342746Z, scheduled_time=2020-04-30T10:07:25.342746Z}, vars={}, execution_time=2020-04-30T10:07:25.342787Z}"`

---

<div class="post-metadata">

**Author:** ![Luca\_Belluccini](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/luca_belluccini/32/33239_2.png) [@Luca\_Belluccini](https://discuss.elastic.co/u/Luca_Belluccini)\
**Post date:** [May 1, 2020, 12:09am UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/4 "2020-05-01T00:09:31Z")

</div>

Thank you @shubhamshd99 for the details.

This should work:

```auto
{
    "trigger": {
      "schedule": {
        "interval": "5000m"
      }
    },
    "input": {
      "search": {
        "request": {
          "search_type": "query_then_fetch",
          "indices": [
            "cft_portal_web_access-2020.03.19"
          ],
          "rest_total_hits_as_int": true,
          "body": {
            "size": 1000,
            "aggregations": {
              "unique_sources": {
                "terms": {
                  "field": "source",
                  "size": 100
                }
              }
            }
          }
        }
      }
    },
    "condition": {
      "compare": {
        "ctx.payload.hits.total": {
          "gt": 0
        }
      }
    },
    "transform": {
      "script": """
def doc_count = ctx.payload.hits.total;
def unique_hits = [];
def unique_sources = ctx.payload.aggregations.unique_sources.buckets.stream().map(hit->hit.key).collect(Collectors.toList());
for(def source : unique_sources){
  for(def i = 0; i< doc_count; i++){
    if(source == ctx.payload.hits.hits[i]._source.source){
      unique_hits.add(ctx.payload.hits.hits[i]._source);
      break;
    }
  }
}
return ['_doc' : unique_hits];
    """
    },
    "actions": {
      "index_payload": {
        "index": {
          "index": "index-for-action_index-test"
        }
      },
      "logging": {
        "logging": {
          "text": "{{ctx}}"
        }
      }
    }
  }

```

Still, I would suggest to use a Transform Job if you want to "aggregate" the data, one per `source`. See [https://www.elastic.co/guide/en/elasticsearch/reference/current/transforms.html](https://www.elastic.co/guide/en/elasticsearch/reference/current/transforms.html)

---

<div class="post-metadata">

**Author:** ![shubhamshd99](https://avatars.discourse-cdn.com/v4/letter/s/8edcca/32.png) [@shubhamshd99](https://discuss.elastic.co/u/shubhamshd99)\
**Post date:** [May 3, 2020, 8:18am UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/5 "2020-05-03T08:18:12Z")

</div>

Thank you @Luca_Belluccini and also apologies for getting back little late!!!  
but the following line would just add \_source part right?? **unique\_hits.add(ctx.payload.hits.hits[i].\_source);**  
But there are other fields like id, score, etc and I want these to be included as well, thats why I tried to remove only \_index, instead of just adding \_source.  
If there is something I missed, please highlight!!!

---

<div class="post-metadata">

**Author:** ![Luca\_Belluccini](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/luca_belluccini/32/33239_2.png) [@Luca\_Belluccini](https://discuss.elastic.co/u/Luca_Belluccini)\
**Post date:** [May 3, 2020, 10:13am UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/6 "2020-05-03T10:13:25Z")

</div>

Hello @shubhamshd99

The only error you were making was the following.

The result of your painless script was:

```auto
{
  "doc": [
    { "_source": { "field...": ... }, "_id": ... }
  ]
}

```

Instead, we need:

```auto
{
  "doc": [
    { "field...": ... , "_id": ...
  ]
}

```

If you want to preserve the `_id`, I suggest doing:

```auto
    "transform": {
      "script": """
def doc_count = ctx.payload.hits.total;
def unique_hits = [];
def unique_sources = ctx.payload.aggregations.unique_sources.buckets.stream().map(hit->hit.key).collect(Collectors.toList());
for(def source : unique_sources){
  for(def i = 0; i< doc_count; i++){
    if(source == ctx.payload.hits.hits[i]._source.source){
      ctx.payload.hits.hits[i]._source._id = ctx.payload.hits.hits[i]._id;
      unique_hits.add(ctx.payload.hits.hits[i]._source);
      break;
    }
  }
}
return ['_doc' : unique_hits];
    """
    },

```

There are other ways to do it, such as:

```auto
    "transform": {
      "script": """
def doc_count = ctx.payload.hits.total;
def unique_hits = [];
def unique_sources = ctx.payload.aggregations.unique_sources.buckets.stream().map(hit->hit.key).collect(Collectors.toList());
for(def source : unique_sources){
  for(def i = 0; i< doc_count; i++){
    if(source == ctx.payload.hits.hits[i]._source.source){
      def event = ctx.payload.hits.hits[i]._source;
      event._id = ctx.payload.hits.hits[i]._id;
      unique_hits.add(event);
      break;
    }
  }
}
return ['_doc' : unique_hits];
    """
    },

```

They are exactly the same.

Please consider that if you maintain `_id`, the documents will be overwritten if the `_id` are the same.

---

<div class="post-metadata">

**Author:** ![shubhamshd99](https://avatars.discourse-cdn.com/v4/letter/s/8edcca/32.png) [@shubhamshd99](https://discuss.elastic.co/u/shubhamshd99)\
**Post date:** [May 4, 2020, 2:32am UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/7 "2020-05-04T02:32:22Z")

</div>

Thank you so much for this beautiful explanation 😇

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [June 1, 2020, 2:32am UTC](https://discuss.elastic.co/t/sending-data-to-existing-elasticsearch-index-through-watcher/230368/8 "2020-06-01T02:32:27Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.