Foreach Ingest processor + conditional append processor

Elastic Stack Elasticsearch
2

Try a script processor...

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "script": {
          "lang": "painless",
          "source": """
ctx.error_values = ctx.logs.stream().map(log -> 
  log.fields.stream()
    .filter(field -> field.key == "error")
    .map(field -> field.value)
    .collect(Collectors.toList())
  )
  .flatMap(l -> l.stream())
  .collect(Collectors.toList())"""
        }
      }
    ]
  },
  "docs": [
    {
      "_source": {
        "something": "some value",
        "logs": [
          {
            "fields": [
              {
                "key": "error",
                "value": "This is an error message."
              },
              {
                "key": "info",
                "value": "Arbitrary info"
              },
              {
                "key": "error",
                "value": "This is a second error message."
              }
            ]
          }
        ]
      }
    },
    {
      "_source": {
        "something": "some value",
        "logs": [
          {
            "fields": [
              {
                "key": "error",
                "value": "This is a third error message."
              },
              {
                "key": "info",
                "value": "Arbitrary info"
              }
            ]
          }
        ]
      }
    }
    
  ]
}

note, that the triple double ticks syntax is from kibana and might ne to be replaced when used with curl.