opensearch-project / opensearch-project/data-prepper
[BUG] aggregate put_all not working with nested keys
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 374
- Forks
- 355
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 8
Description
Describe the bug
It seems that the put_all aggregate function only works with "root" keys, and do not add new nested keys.
For example, I have these 2 postfix events that I need to aggregate, based on /host/id and /email/local_id :
{
"@timestamp": "2024-01-19T13:25:35.503436Z",
"postfix": {
"nrcpt": "1",
"size": "17603"
},
"application": "postfix",
"host": {
"id": "95e0888c1df24d79a4bc827f9636cca7"
},
"email": {
"from": {
"address": "EMAIL_FROM"
},
"local_id": "7A20930009B"
},
"message": "queue active"
}
{
"@timestamp": "2024-01-19T13:25:35.527242Z",
"postfix": {
"relay": "localhost[::1]:24",
"delays": "0/0/0/0.02",
"dsn": "2.1.5",
"delay": "0.03",
"status": "sent"
},
"application": "postfix",
"host": {
"id": "95e0888c1df24d79a4bc827f9636cca7"
},
"email": {
"local_id": "7A20930009B",
"to": {
"address": "EMAIL_TO"
}
},
"test": "OK"
}
The put_all function only add keys that are at the '/' level of the JSON, and drop the other keys :
{
"@timestamp": "2024-01-19T13:25:35.503436Z",
"postfix": {
"nrcpt": "1",
"size": "17603"
},
"application": "postfix",
"host": {
"id": "95e0888c1df24d79a4bc827f9636cca7"
},
"email": {
"from": {
"address": "EMAIL_FROM"
},
"local_id": "7A20930009B"
},
"message": "queue active",
"test": "ok"
}
To Reproduce
Steps to reproduce the behavior:
pipeline:
processor:
- aggregate:
identification_keys:
- "/host/id"
- "/email/local_id"
action:
put_all:
group_duration: "10s"
aggregate_when: '/email/local_id != null'
Expected behavior
It needs to create the missing entries in the nested keys :
{
"@timestamp": "2024-01-19T13:25:35.503436Z",
"postfix": {
"nrcpt": "1",
"size": "17603",
"relay": "localhost[::1]:24",
"delays": "0/0/0/0.02",
"dsn": "2.1.5",
"delay": "0.03",
"status": "sent"
},
"application": "postfix",
"host": {
"id": "95e0888c1df24d79a4bc827f9636cca7"
},
"email": {
"from": {
"address": "EMAIL_FROM"
},
"to": {
"address": "EMAIL_TO"
}
"local_id": "7A20930009B"
},
"message": "queue active",
"test": "ok"
}
Environment (please complete the following information):
- latest dataprepper docker image (2.6.1)
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by reproducing the aggregate pipeline from the issue with the two sample events and nested identification keys. Read the aggregate processor's put_all action and its existing tests, if present, to trace how nested paths are merged. Done means missing nested fields from the second event are retained in the aggregated output without breaking root-level fields.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100