# Creating an enrich policy from collapsed data

**URL:** <https://discuss.elastic.co/t/creating-an-enrich-policy-from-collapsed-data/318629>\
**Category:** Elasticsearch\
**Created:** [November 10, 2022, 8:40am UTC](https://discuss.elastic.co/t/creating-an-enrich-policy-from-collapsed-data/318629 "2022-11-10T08:40:17Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![cwe](https://avatars.discourse-cdn.com/v4/letter/c/f14d63/32.png) [@cwe](https://discuss.elastic.co/u/cwe)\
**Post date:** [November 10, 2022, 8:40am UTC](https://discuss.elastic.co/t/creating-an-enrich-policy-from-collapsed-data/318629/1 "2022-11-10T08:40:17Z")

</div>

Hi Elastic community

I have an index containing many documents per host.id describing the currently running software. This information might change over time, due to software updates.

I can use a collapse query to get the latest entry of each software info from the index, like this:

```auto
GET /filebeat-*/_search
{
  "query": {
    "term": {
      "json.logtype": {
        "value": "software"
      }
    }
  },
  "collapse": {
    "field": "host.id",
    "inner_hits": {
      "name": "most_recent",
      "size": 1,
      "sort": [{"@timestamp": "desc"}]
    }
  },
  "_source": false
}

```

so I will find the info I need inside inner\_hits:

```auto
    "hits" : [
      {
        "_index" : "filebeat-2022.09.11",
        "_type" : "_doc",
        "_id" : "IhjwLoMBm1GMkVa4CZZ-",
        "_score" : 9.112949,
        "fields" : {
          "host.id" : [
            "8f3a09ac069a4ba885e860a011f8570d"
          ]
        },
        "inner_hits" : {
          "most_recent" : {
            "hits" : {
              "total" : {
                "value" : 215,
                "relation" : "eq"
              },
              "max_score" : null,
              "hits" : [
                {
                  "_index" : "filebeat-2022.11",
                  "_type" : "_doc",
                  "_id" : "FDg4YIQBhix3UBkMF9Y-",
                  "_score" : null,
                  "_source" : {
                    "@timestamp" : "2022-11-10T06:26:27.423Z",
                    "host": {
                      "id" : "8f3a09ac069a4ba885e860a011f8570d",
                      ...
                    },
                    "json" : {
                      "logtype" : "software",
                      "software" : {
                        "build" : "2021-12-07 10:11:51",
                        "version" : "7.5.0 Build 199",
                        ...
                      }
                ...

```

Now I want to create an enrich index from exactly the data inside inner\_hits to add the software version to other incoming data from these systems. I can setup the enrich policy and generate the index like this:

```auto
PUT /_enrich/policy/enrich-filebeat-software
{
  "match": {
    "indices": [
      "filebeat-*"
    ],
    "match_field": "host.id",
    "enrich_fields": [
      "json.software"
    ],
    "query": {
      "terms": {
        "json.logtype": {
          "value": "software"
        }
      }
    }
  }
}

```

But this contains all documents for the host.id, not the latest, making the enrich index overly large and taking a long time to execute.

How could I generate an enrich index from the inner\_hits of my initial search? Do I need to create an intermediate index from which I generate the enrich index?

Any pointers are greatly appreciated.

Best regards,  
Christian

---

<div class="post-metadata">

**Author:** ![cwe](https://avatars.discourse-cdn.com/v4/letter/c/f14d63/32.png) [@cwe](https://discuss.elastic.co/u/cwe)\
**Post date:** [November 15, 2022, 8:23am UTC](https://discuss.elastic.co/t/creating-an-enrich-policy-from-collapsed-data/318629/2 "2022-11-15T08:23:34Z")

</div>

Hi community,

I figured out how to do this, I'll post my findings here so they can be reused if needed...

My watcher looks like this:

```auto
PUT _watcher/watch/watch-filebeat-software
{
  "trigger": {
    "schedule": { "interval": "24h" }
  },
  "input": {
    "search": {
      "request": {
        "indices": ["filebeat-*"],
        "body": {
          "query": {
            "term": {
              "json.logtype": {
                "value": "software"
              }
            }
          },
          "size": 10000,
          "collapse": {
            "field": "host.id",
              "inner_hits": {
                "name": "most_recent",
                "size": 1,
                "sort": [{"@timestamp": "desc"}],
                "_source": ["host.id", "json.software", "@timestamp"]
              }
            },
            "_source": false
        }
      }
    }
  },
  "actions": {
    "index_payload": {
      "transform": {
        "script": """
          def documents = ctx.payload.hits.hits.stream()
            .map(hit -> [
              "_index": "watch-filebeat-software",
              "_id": hit.inner_hits.most_recent.hits.hits.0._source.host.id,
              "host": hit.inner_hits.most_recent.hits.hits.0._source.host,
              "json": hit.inner_hits.most_recent.hits.hits.0._source.json,
              "@timestamp": hit.inner_hits.most_recent.hits.hits.0._source['@timestamp']
            ])
            .collect(Collectors.toList());
          return ["_doc": documents];
        """
      },
      "index": {}
    },
    "enrich": {
      "webhook": {
        "method": "PUT",
        "scheme": "https",
        "host": " ********.elastic-cloud.com",
        "port": 9243,
        "path": "/_enrich/policy/enrich-filebeat-software/_execute",
        "headers": {
          "Authorization": "ApiKey ***************************************************** =="
        }
      }
    }
  }
}

```

The enrich policy is configured like this:

```auto
PUT /_enrich/policy/enrich-filebeat-software
{
  "match": {
    "indices": [
      "watch-filebeat-software"
    ],
    "match_field": "host.id",
    "enrich_fields": [
      "json"
    ],
    "query": { "match_all": {} }
  }
}

```

With this I have an index `watch-filebeat-software` which is small (one entry per host.id) and is updated by the watcher once a day. At the same time the enrich policy is executed, so the enrich index is kept up to date.

Best regards,  
Christian

---

<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:** [December 13, 2022, 8:23am UTC](https://discuss.elastic.co/t/creating-an-enrich-policy-from-collapsed-data/318629/3 "2022-12-13T08:23:44Z")

</div>

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