PUT _ingest/pipeline/logs-ecs-json
{
  "processors": [
    {
      "rename": {
        "field": "message",
        "target_field": "_ecs_json_message",
        "ignore_missing": true,
        "if": "ctx.message.startsWith('{') && ctx.message.endsWith('}') && ctx.message.contains('\"@timestamp\"') && ctx.message.contains('\"ecs') && ctx.message.contains('version\"')"
      }
    },
    {
      "json": {
        "field": "_ecs_json_message",
        "add_to_root": true,
        "add_to_root_conflict_strategy": "merge",
        "allow_duplicate_keys": true,
        "if": "ctx.containsKey('_ecs_json_message')",
        "on_failure": [
          {
            "rename": {
              "field": "_ecs_json_message",
              "target_field": "message",
              "ignore_missing": true
            }
          },
          {
            "set": {
              "field": "error.message",
              "value": "Error while parsing JSON",
              "override": false
            }
          }
        ]
      }
    },
    {
      "remove": {
        "field": "_ecs_json_message",
        "ignore_missing": true
      }
    },
    {
      "dot_expander": {
        "field": "*",
        "override": true
      }
    },
    {
      "set": {
        "field": "data_stream.dataset",
        "copy_from": "event.dataset",
        "override": false
      }
    },
    {
      "script": {
        "source": "ctx.data_stream.dataset = /[\\/*?\"<>|, #:-]/.matcher(ctx.data_stream.dataset).replaceAll('_')",
        "if": "ctx.data_stream?.dataset != null"
      }
    },
    {
      "script": {
        "source": "ctx.data_stream.namespace = /[\\/*?\"<>|, #:]/.matcher(ctx.data_stream.namespace).replaceAll('_')",
        "if": "ctx.data_stream?.namespace != null"
      }
    },
    {
      "set": {
        "field": "data_stream.type",
        "value": "logs",
        "override": false
      }
    },
    {
      "set": {
        "field": "data_stream.dataset",
        "value": "generic",
        "override": false
      }
    },
    {
      "set": {
        "field": "data_stream.namespace",
        "value": "default",
        "override": false
      }
    },
    {
      "set": {
        "field": "event.dataset",
        "copy_from": "data_stream.dataset",
        "override": true
      }
    },
    {
      "set": {
        "field": "_index",
        "value": "logs-{{{data_stream.dataset}}}-{{{data_stream.namespace}}}"
      }
    }
  ]
}