# Using transform with enrich pipeline to enrich ip with latest vulnerabilities
**URL:** https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553
**Category:** Elasticsearch
**Created:** [December 6, 2022, 9:31am UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553 "2022-12-06T09:31:41Z")
**Posts on this page:** 8
**Page:** 1
### Author:  [@siginigin](https://discuss.elastic.co/u/siginigin)
#### Post date: [December 6, 2022, 9:31am UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/1 "2022-12-06T09:31:41Z")
Hi,
I want to enrich incoming event that contains local source.ip/destination.ip with vulnerability field with this logic:
match source.ip with server.ip, get highest vulnerability.severity\_num from latest vulnerability.report\_date.
I created index with vulnerabilies, each document contains server.ip (IP), vulnerability.report\_date (date), vulnerability.severity\_num (number), vulnerability.scan\_date(date). I have a script that pulls out report data from vulnerability scanner, sends it to logstash to parse it and store in vulnerabilities index, on daily basis. Field vulnerability.scan\_date as event date (@timestamp).
Now, I tried to enrich incoming events with logstash. It works, but it's slow - for 1k eps it has to do 2k queries to elasticsearch and events in logstash become lagging.
So I decided to move this logic to elasticsearch's enrich pipeline. Problem with this that enrich processor cannot perform aggregated query, only direct match. The cure for this to create another index with server.ip as key with latest highest vulnerability.
I found transforms, which could transform my vulnerabilities index to latest\_vulnerabilities index, but here I'm stucked. I tried pivot transform with this logic:
1. group by: server.ip
aggregations: vulnerabilities.report\_date, vulnerabilities.severity\_num
- this gives me one document per IP, latest report is correct, but highest severity is taken from all reports, not only from latest report
1. group by: server.ip, vulnerabilities.report\_date
aggregations: vulnerabilities.severity\_num
- this gives me one document per report\_date with correct highest severity, but there are multiple documents with same server.ip and I'm not sure which doc will be taken by enrich processor.
1. sub-aggregation:
```auto
{
"source": {
"index": "vulnerabilities",
"query": {
"bool": {
"should": [
{
"exists": {
"field": "vulnerability.severity_num"
}
}
],
"minimum_should_match": 1
}
}
},
"pivot": {
"group_by": {
"server_ip": {
"terms": {
"field": "server.ip",
"missing_bucket": true
}
}
},
"aggregations": {
"max_report": {
"max": {
"field": "vulnerability.report_date"
},
"aggs": {
"max_vuln": {
"max": {
"field": "vulnerability.severity_num"
}
}
}
}
}
},
"description": "Latest vulnerability per ip",
"dest": {
"index": "vulnerabilities_latest"
},
"frequency": "5m",
"sync": {
"time": {
"field": "vulnerability.scan_date",
"delay": "60s"
}
},
"retention_policy": {
"time": {
"field": "vulnerability.scan_date",
"max_age": "30d"
}
}
}
```
- but this does not work
`"reason" : "Aggregator [max_report] of type [max] cannot accept sub-aggregations"`
How can I achive my goal?
Thank you for help.
---
### Author:  [@Hendrik\_Muhs](https://discuss.elastic.co/u/Hendrik_Muhs)
#### Post date: [December 7, 2022, 6:58am UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/2 "2022-12-07T06:58:36Z")
> [@siginigin](#):
>
> 1. group by: server.ip
> aggregations: vulnerabilities.report\_date, vulnerabilities.severity\_num
>
> - this gives me one document per IP, latest report is correct, but highest severity is taken from all reports, not only from latest report
To get the severity from the latest report have a look at [top\_metrics](https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-metrics-top-metrics.html). You get the _latest_ by specifying `sort` on a timestamp field with order `desc`.
Hope that helps.
---
### Author:  [@siginigin](https://discuss.elastic.co/u/siginigin)
#### Post date: [December 12, 2022, 9:53am UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/3 "2022-12-12T09:53:59Z")
Hi Hendrik,
thank you for your answer. Unfortunately I can not use "latest" because I'm not looking for latest vulnerability of some server.ip, but highest vulnerability from latest report of some server.ip. I can't select numeric field such as severity\_num when using "latest".
I tried to use pivot, with top\_metrics with this dataset:
|server.ip|vulnerability.report\_date|vulnerability.severity\_num|
|10.1.2.3|Dec 5, 2022 @ 16:00:14.000|5|
|10.1.2.3|Dec 5, 2022 @ 16:00:14.000|5.3|
|10.1.2.3|Dec 5, 2022 @ 16:00:14.000|7.5|
|10.1.2.3|Dec 5, 2022 @ 16:00:14.000|2.4|
|10.1.2.3|Dec 5, 2022 @ 16:00:14.000|1.3|
and this transform:
```auto
{
"group_by": {
"server.ip": {
"terms": {
"field": "server.ip"
}
}
},
"aggregations": {
"top_metrics": {
"top_metrics": {
"metrics": [
{
"field": "vulnerability.severity_num"
}
],
"sort": {
"vulnerability.report_date": "desc"
}
}
}
}
}
```
but it returns 5 instead of 7.5. I also tried to add sort for vulnerability.severity\_num, but result was the same:
```auto
{
"group_by": {
"server.ip": {
"terms": {
"field": "server.ip"
}
}
},
"aggregations": {
"top_metrics": {
"top_metrics": {
"metrics": [
{
"field": "vulnerability.severity_num"
}
],
"sort": {
"vulnerability.report_date": "desc",
"vulnerability.severity_num": "desc"
}
}
}
}
}
```
Why sort doesn't take another field (vulnerability.severity\_num) into account?
Thank you for help.
Regards, Sigi
---
### Author:  [@Hendrik\_Muhs](https://discuss.elastic.co/u/Hendrik_Muhs)
#### Post date: [December 12, 2022, 10:45am UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/4 "2022-12-12T10:45:11Z")
> [@siginigin](#):
>
> Why sort doesn't take another field (vulnerability.severity\_num) into account?
Top metrics doesn't support [more than 1 sort criteria](https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-metrics-top-metrics.html#_sort). I wonder why it even lets you specify 2. I will check and report an issue in case.
I see 2 options, a `scripted_metric` that implements your own version of top metric but with secondary sort or a runtime field that combines the report date and the severity.
However, 1st I like to understand:
> [@siginigin](#):
>
> |server.ip|vulnerability.report\_date|vulnerability.severity\_num|
> |10.1.2.3|Dec 5, 2022 @ 16:00:14.000|5|
> |10.1.2.3|Dec 5, 2022 @ 16:00:14.000|5.3|
> |10.1.2.3|Dec 5, 2022 @ 16:00:14.000|7.5|
> |10.1.2.3|Dec 5, 2022 @ 16:00:14.000|2.4|
> |10.1.2.3|Dec 5, 2022 @ 16:00:14.000|1.3|
Are the reports coming in regularly and all with the same date? Are those daily reports?
---
### Author:  [@siginigin](https://discuss.elastic.co/u/siginigin)
#### Post date: [December 12, 2022, 12:10pm UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/5 "2022-12-12T12:10:22Z")
Hi Hendrik,
> I see 2 options, a `scripted_metric` that implements your own version of top metric but with secondary sort or a runtime field that combines the report date and the severity.
Ok, I'm going take deep dive on that.
> Are the reports coming in regularly and all with the same date? Are those daily reports?
We can assume daily reports. In dataset there is only one report of one IP - this is more general dataset:
```auto
server.ip vulnerability.report_date vulnerability.severity_num
10.1.2.3 Dec 5, 2022 @ 16:00:14.000 5
10.1.2.3 Dec 5, 2022 @ 16:00:14.000 5.3
10.1.2.3 Dec 5, 2022 @ 16:00:14.000 7.5
10.1.2.3 Dec 5, 2022 @ 16:00:14.000 2.4
10.1.2.3 Dec 5, 2022 @ 16:00:14.000 1.3
10.2.5.2 Dec 5, 2022 @ 16:00:14.000 2.5
10.2.5.2 Dec 5, 2022 @ 16:00:14.000 6.3
10.2.5.2 Dec 5, 2022 @ 16:00:14.000 4.2
10.2.3.2 Dec 5, 2022 @ 16:00:14.000 4.2
172.30.10.2 Dec 5, 2022 @ 16:00:14.000 10
172.30.10.2 Dec 5, 2022 @ 16:00:14.000 7.4
172.30.10.2 Dec 5, 2022 @ 16:00:14.000 1.4
172.30.10.2 Dec 5, 2022 @ 16:00:14.000 7.5
172.30.10.2 Dec 5, 2022 @ 16:00:14.000 6.5
10.1.2.3 Dec 4, 2022 @ 16:00:14.000 5
10.1.2.3 Dec 4, 2022 @ 16:00:14.000 5.3
10.1.2.3 Dec 4, 2022 @ 16:00:14.000 7.5
10.1.2.3 Dec 4, 2022 @ 16:00:14.000 2.4
10.1.2.3 Dec 4, 2022 @ 16:00:14.000 1.3
10.2.5.2 Dec 4, 2022 @ 16:00:14.000 2.5
10.2.5.2 Dec 4, 2022 @ 16:00:14.000 6.3
10.2.5.2 Dec 4, 2022 @ 16:00:14.000 4.2
10.2.3.2 Dec 4, 2022 @ 16:00:14.000 4.2
172.30.10.2 Dec 4, 2022 @ 16:00:14.000 10
172.30.10.2 Dec 4, 2022 @ 16:00:14.000 7.4
172.30.10.2 Dec 4, 2022 @ 16:00:14.000 1.4
172.30.10.2 Dec 4, 2022 @ 16:00:14.000 7.5
172.30.10.2 Dec 4, 2022 @ 16:00:14.000 6.5
```
There 2 reports (4.12 and 5.12), each report consist of multiple IP's.
Transform of this dataset should produce following:
```auto
server.ip vulnerability.report_date vulnerability.severity_num
10.1.2.3 Dec 5, 2022 @ 16:00:14.000 7.5
10.2.5.2 Dec 5, 2022 @ 16:00:14.000 6.3
10.2.3.2 Dec 5, 2022 @ 16:00:14.000 4.2
172.30.10.2 Dec 5, 2022 @ 16:00:14.000 10
```
Thank you.
---
### Author:  [@Hendrik\_Muhs](https://discuss.elastic.co/u/Hendrik_Muhs)
#### Post date: [December 12, 2022, 12:47pm UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/6 "2022-12-12T12:47:04Z")
> [@siginigin](#):
>
> Ok, I'm going take deep dive on that.
The transform docs contain [painless examples](https://www.elastic.co/guide/en/elasticsearch/reference/8.4/transform-painless-examples.html#painless-top-hits) that should help you to get started.
> [@siginigin](#):
>
> We can assume daily reports. In dataset there is only one report of one IP - this is more general dataset:
I suggest to consider a `date_histogram` in `group_by` with a calendar interval of 1 day. The `top_metrics` can than sort on severity only. That way your transform destination isn't the _latest_ severity anymore, but daily ones. Maybe that's even useful for you.
To reduce storage you could age out old data by using a `retention_policy`. A `retention_policy` deletes documents when they are older than the specified time range.
In terms of performance a transform with a `date_histogram` performs much better, because it doesn't have to look back and re-load all historic data points. The current one in contrast re-iterates through all data points of a server ip on every run. With `date_histogram` + `terms` the transform will only aggregate on data from max 2 days in the past.
---
### Author:  [@siginigin](https://discuss.elastic.co/u/siginigin)
#### Post date: [December 12, 2022, 2:51pm UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/7 "2022-12-12T14:51:05Z")
Hi Hendrik,
I found easier way - I created 2 transforms:
1. Pivot transform from vulnerabilities index group by server.ip, vulnerability.report\_date, aggregate max of vulnerability.severity\_num, output to new index max\_ip\_vulnerabilities\_by\_latest\_report\_date.
2. Latest transform from max\_ip\_vulnerabilities\_by\_latest\_report\_date index, unique keys server.ip, sort by vulnerability.report\_date -\> output to new index latest\_vulnerabilities -\> this index contains last highest vulnerability for each ip and each ip is in index only once so it can be used as key.
Now I can create enrich with latest\_vulnerabilities index.
Thank you for help.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [January 9, 2023, 2:51pm UTC](https://discuss.elastic.co/t/using-transform-with-enrich-pipeline-to-enrich-ip-with-latest-vulnerabilities/320553/8 "2023-01-09T14:51:16Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.