Change destination datastream with Elasticsearch ingest pipeline

Elastic Stack Elasticsearch
, ,
1

We run Elastic stack in docker containers. The container logs are collected with Elastic Agent, using docker integration and datastreams.
This means that the logs of elasticsearch container itself by default end up in logs-docker.container_logs datastream. Nothing really disturbing so far.

For better stack monitoring and to properly understand the normal behaviour of elasticsearch container, I wanted to parse properly the logs from elasticsearch container with the logs-elasticsearch.server ingest pipeline. The whole pipeline would run as follows:

(original event) -> logs-docker.container_logs -> logs-docker.container_logs@custom -> logs-elasticsearch.server (with condition on the container name)

Unfortunately I was not able to get this working because elasticsearch complains about a mismatch between expected value for data_stream.dataset: elasticsearch.server and the actual value: docker.container_logs.

That would be understandable because elasticsearch.server ingest pipeline transformed the original docker.container_logs event into an event which fits the index mapping of elasticsearch.server.
Therefore I tried to change both data_stream.dataset and event.dataset of the transformed event in logs-docker.container_logs@custom pipeline (via both Set and Script processors) but this seems to be ignored (perhaps not supported?).

Anyone has succeeded to overcome this hurdle?

2

Hi @i.raisr Welcome to the community.

Have you looked at the new reroute processor ... perhaps that would work...

Hmmm... Need to think about this seems like it should be easier

3

Woah! Thanks a lot @stephenb. I must have missed somehow this Reroute processor :frowning:
I will try tomorrow and let the community know :slight_smile:
I.

4

Results of my investigation:

  • Reroute processor is not offered by Kibana UI (I run Kibana 8.8.0). Perhaps it's because it's still technical preview. However typing manually works.
  • For keycloak log events, reroute works nicely.
  • For elasticsearch.server log events, reroute processor somehow screws up data_stream.type field and the event is not rerouted (it is actually lost).

Here is the ingest pipeline hooked up on logs-docker.container_logs@custom:

PUT _ingest/pipeline/logs-docker.container_logs@custom
{
  "version": 2,
  "processors": [
    {
      "rename": {
        "field": "event.original",
        "target_field": "event.message.original",
        "ignore_missing": true,
        "if": "(ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app') || (ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch')",
        "ignore_failure": true
      }
    },
    {
      "pipeline": {
        "if": "ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app'",
        "tag": "keycloak",
        "ignore_failure": true,
        "name": "logs-keycloak.log-1.8.0-events"
      }
    },
    {
      "pipeline": {
        "name": "logs-elasticsearch.server-1.7.2",
        "if": "ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch'",
        "tag": "elasticsearch",
        "ignore_failure": true
      }
    },
    {
      "rename": {
        "field": "event.message.original",
        "target_field": "event.original",
        "ignore_missing": true,
        "if": "(ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app') && (ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch')",
        "ignore_failure": true
      }
    },
    {
      "set": {
        "if": "ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app'",
        "tag": "keycloak",
        "ignore_failure": true,
        "field": "data_stream.dataset",
        "value": "keycloak"
      }
    },
    {
      "set": {
        "field": "data_stream.dataset",
        "value": "elasticsearch.server",
        "if": "ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch'",
        "tag": "elasticsearch",
        "ignore_failure": true
      }
    },
    {
      "reroute": {
        "if": "(ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app') || (ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch')",
        "tag": "keycloak",
        "ignore_failure": true
      }
    }
  ]
}
5

And here is output last three steps of the test pipeline run (unfortunately this portal has message length limit). It can be observed that data_stream.type is changed only after the reroute process runs. It is unclear why. Recall that this happens only for log events from elasticsearch container; log events from sso-app (keycloak) container are rerouted correctly.

{
  "docs": [
    {
      "processor_results": [
       {
          "processor_type": "set",
          "status": "skipped",
          "tag": "keycloak",
          "if": {
            "condition": "ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app'",
            "result": false
          }
        },
        {
          "processor_type": "set",
          "status": "success",
          "tag": "elasticsearch",
          "if": {
            "condition": "ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch'",
            "result": true
          },
          "doc": {
            "_index": ".ds-logs-docker.container_logs-sccoe-2023.06.29-000013",
            "_version": "-3",
            "_id": "B0q7BokBxFYDMds2r3yb",
            "_source": {
              "container": {
                "name": "elasticsearch",
                "image": {
                  "name": "nexus.sccoe.int/sys/elastic/elastic-elasticsearch:0.11.0-01"
                },
                "id": "85a2ffa2853d3c9b114626f3f3c9003b08a911494a7edada623d725c1f249e8a",
                "labels": {
                  "org_opencontainers_image_title": "Elasticsearch",
                  "com_docker_compose_oneoff": "False",
                  "com_docker_compose_project": "sccoe",
                  "org_opencontainers_image_created": "2023-05-23T17:16:07.179039820Z",
                  "org_opencontainers_image_version": "8.8.0",
                  "org_opencontainers_image_vendor": "Elastic",
                  "org_label-schema_vcs-url": "https://github.com/elastic/elasticsearch",
                  "org_label-schema_vcs-ref": "c01029875a091076ed42cdb3a41c10b1a9a5a20f",
                  "org_label-schema_vendor": "Elastic",
                  "com_docker_compose_service": "elasticsearch-esacr",
                  "com_docker_compose_image": "sha256:05d5aea6674aa8cfea3000260533db457b3b21f59c7af7db050f156ffaae4bfc",
                  "org_opencontainers_image_licenses": "Elastic-License-2.0",
                  "org_label-schema_license": "Elastic-License-2.0",
                  "org_label-schema_build-date": "2023-05-23T17:16:07.179039820Z",
                  "com_docker_compose_config-hash": "5150130d0468dc39b6ee19a3cc31a22c671b7a616f586b039f5f7d7bcd5c03bb",
                  "org_label-schema_version": "8.8.0",
                  "com_docker_compose_project_config_files": "/home/env-admin/elastic/docker-compose.yml",
                  "com_docker_compose_depends_on": "elasticsearch-init-esacr:service_started:false",
                  "org_opencontainers_image_source": "https://github.com/elastic/elasticsearch",
                  "org_label-schema_name": "Elasticsearch",
                  "org_opencontainers_image_revision": "c01029875a091076ed42cdb3a41c10b1a9a5a20f",
                  "org_label-schema_schema-version": "1.0",
                  "org_opencontainers_image_documentation": "https://www.elastic.co/guide/en/elasticsearch/reference/index.html",
                  "org_opencontainers_image_url": "https://www.elastic.co/products/elasticsearch",
                  "com_docker_compose_container-number": "1",
                  "org_label-schema_usage": "https://www.elastic.co/guide/en/elasticsearch/reference/index.html",
                  "com_docker_compose_version": "2.17.3",
                  "org_opencontainers_image_ref_name": "ubuntu",
                  "com_docker_compose_project_working_dir": "/home/env-admin/elastic",
                  "org_label-schema_url": "https://www.elastic.co/products/elasticsearch"
                }
              },
              "agent": {
                "name": "elastic",
                "id": "beeaa337-bee6-4614-a7ef-196ba0e11121",
                "type": "filebeat",
                "ephemeral_id": "2946f2a4-2394-4178-b2ea-bccbc1c72361",
                "version": "8.8.0"
              },
              "process": {
                "thread": {
                  "name": "elasticsearch[elastic][masterService#updateTask][T#167]"
                }
              },
              "log": {
                "file": {
                  "path": "/var/lib/docker/containers/85a2ffa2853d3c9b114626f3f3c9003b08a911494a7edada623d725c1f249e8a/85a2ffa2853d3c9b114626f3f3c9003b08a911494a7edada623d725c1f249e8a-json.log"
                },
                "offset": 1599400,
                "level": "INFO",
                "logger": "org.elasticsearch.cluster.metadata.MetadataMappingService"
              },
              "elastic_agent": {
                "id": "beeaa337-bee6-4614-a7ef-196ba0e11121",
                "version": "8.8.0",
                "snapshot": false
              },
              "sccoe": {
                "component": {
                  "name": "System Support",
                  "id": "SYS"
                }
              },
              "message": "update_mapping [_doc]",
              "input": {
                "type": "filestream"
              },
              "@timestamp": "2023-06-29T10:38:21.642Z",
              "ecs": {
                "version": "1.2.0"
              },
              "elasticsearch": {
                "node": {
                  "name": "elastic",
                  "id": "3QW8_ZQRRAOb8tDuwzGX4w"
                },
                "cluster": {
                  "name": "SCCoE",
                  "uuid": "uXtSsZ7KShykEYcdwZGqnQ"
                },
                "server": {
                  "tags": [
                    "elastic-agent",
                    "beats_input_codec_plain_applied"
                  ]
                },
                "index": {
                  "name": ".ds-metrics-prometheus.collector-sccoe-2023.06.29-000008",
                  "id": "nBWAS9AwRsO_eBlZaDJ_KA"
                }
              },
              "stream": "stdout",
              "data_stream": {
                "namespace": "sccoe",
                "type": "logs",
                "dataset": "elasticsearch.server"
              },
              "service": {
                "name": "ES_ECS",
                "type": "elasticsearch"
              },
              "@version": "1",
              "host": {
                "hostname": "elastic",
                "os": {
                  "kernel": "4.18.0-193.28.1.el8_2.x86_64",
                  "codename": "Ootpa",
                  "name": "Red Hat Enterprise Linux",
                  "type": "linux",
                  "family": "redhat",
                  "version": "8.2 (Ootpa)",
                  "platform": "rhel"
                },
                "containerized": false,
                "ip": [
                  "172.16.108.165",
                  "fe80::28cf:f4ea:b97b:f1a4",
                  "172.17.0.1",
                  "fe80::42:73ff:fecb:e96a",
                  "10.200.0.1",
                  "fe80::42:c5ff:fe37:9d14",
                  "fe80::3c64:4cff:fe76:218",
                  "fe80::3846:24ff:fee6:780e",
                  "fe80::cab:d5ff:fed9:da27",
                  "fe80::c46e:a0ff:fe3c:a8e2",
                  "fe80::b483:42ff:febe:728f",
                  "fe80::38ea:5aff:fed8:1ea9",
                  "fe80::9874:62ff:fe7a:c2e0",
                  "fe80::6c60:e0ff:fe51:2abd"
                ],
                "name": "elastic",
                "id": "3QW8_ZQRRAOb8tDuwzGX4w",
                "mac": [
                  "00-50-56-BE-03-DB",
                  "02-42-73-CB-E9-6A",
                  "02-42-C5-37-9D-14",
                  "0E-AB-D5-D9-DA-27",
                  "3A-46-24-E6-78-0E",
                  "3A-EA-5A-D8-1E-A9",
                  "3E-64-4C-76-02-18",
                  "6E-60-E0-51-2A-BD",
                  "9A-74-62-7A-C2-E0",
                  "B6-83-42-BE-72-8F",
                  "C6-6E-A0-3C-A8-E2"
                ],
                "architecture": "x86_64"
              },
              "event": {
                "agent_id_status": "auth_metadata_missing",
                "ingested": "2023-06-29T10:48:46.809808694Z",
                "created": "2023-06-29T10:38:21.642Z",
                "kind": "event",
                "message": {
                  "original": "{\"@timestamp\":\"2023-06-29T10:38:21.642Z\", \"log.level\": \"INFO\", \"message\":\"[.ds-metrics-prometheus.collector-sccoe-2023.06.29-000008/nBWAS9AwRsO_eBlZaDJ_KA] update_mapping [_doc]\", \"ecs.version\": \"1.2.0\",\"service.name\":\"ES_ECS\",\"event.dataset\":\"elasticsearch.server\",\"process.thread.name\":\"elasticsearch[elastic][masterService#updateTask][T#167]\",\"log.logger\":\"org.elasticsearch.cluster.metadata.MetadataMappingService\",\"elasticsearch.cluster.uuid\":\"uXtSsZ7KShykEYcdwZGqnQ\",\"elasticsearch.node.id\":\"3QW8_ZQRRAOb8tDuwzGX4w\",\"elasticsearch.node.name\":\"elastic\",\"elasticsearch.cluster.name\":\"SCCoE\"}\n"
                },
                "category": [
                  "database"
                ],
                "type": [
                  "info"
                ],
                "dataset": "elasticsearch.server",
                "hash": "c78c03eca0d05a9c74e97305cbf79bc49f904f8a71d35ca6dd4df2d81246192f"
              }
            },
            "_ingest": {
              "pipeline": "_simulate_pipeline",
              "timestamp": "2023-06-29T10:48:46.809808694Z"
            }
          }
        },
        {
          "processor_type": "reroute",
          "status": "success",
          "tag": "keycloak",
          "if": {
            "condition": "(ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app') || (ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch')",
            "result": true
          },
          "doc": {
            "_index": ".ds-elasticsearch.server-sccoe",
            "_version": "-3",
            "_id": "B0q7BokBxFYDMds2r3yb",
            "_source": {
              "container": {
                "name": "elasticsearch",
                "image": {
                  "name": "nexus.sccoe.int/sys/elastic/elastic-elasticsearch:0.11.0-01"
                },
                "id": "85a2ffa2853d3c9b114626f3f3c9003b08a911494a7edada623d725c1f249e8a",
                "labels": {
                  "org_opencontainers_image_title": "Elasticsearch",
                  "com_docker_compose_oneoff": "False",
                  "com_docker_compose_project": "sccoe",
                  "org_opencontainers_image_created": "2023-05-23T17:16:07.179039820Z",
                  "org_opencontainers_image_version": "8.8.0",
                  "org_opencontainers_image_vendor": "Elastic",
                  "org_label-schema_vcs-url": "https://github.com/elastic/elasticsearch",
                  "org_label-schema_vcs-ref": "c01029875a091076ed42cdb3a41c10b1a9a5a20f",
                  "org_label-schema_vendor": "Elastic",
                  "com_docker_compose_service": "elasticsearch-esacr",
                  "com_docker_compose_image": "sha256:05d5aea6674aa8cfea3000260533db457b3b21f59c7af7db050f156ffaae4bfc",
                  "org_opencontainers_image_licenses": "Elastic-License-2.0",
                  "org_label-schema_license": "Elastic-License-2.0",
                  "org_label-schema_build-date": "2023-05-23T17:16:07.179039820Z",
                  "com_docker_compose_config-hash": "5150130d0468dc39b6ee19a3cc31a22c671b7a616f586b039f5f7d7bcd5c03bb",
                  "org_label-schema_version": "8.8.0",
                  "com_docker_compose_project_config_files": "/home/env-admin/elastic/docker-compose.yml",
                  "com_docker_compose_depends_on": "elasticsearch-init-esacr:service_started:false",
                  "org_opencontainers_image_source": "https://github.com/elastic/elasticsearch",
                  "org_label-schema_name": "Elasticsearch",
                  "org_opencontainers_image_revision": "c01029875a091076ed42cdb3a41c10b1a9a5a20f",
                  "org_label-schema_schema-version": "1.0",
                  "org_opencontainers_image_documentation": "https://www.elastic.co/guide/en/elasticsearch/reference/index.html",
                  "org_opencontainers_image_url": "https://www.elastic.co/products/elasticsearch",
                  "com_docker_compose_container-number": "1",
                  "org_label-schema_usage": "https://www.elastic.co/guide/en/elasticsearch/reference/index.html",
                  "com_docker_compose_version": "2.17.3",
                  "org_opencontainers_image_ref_name": "ubuntu",
                  "com_docker_compose_project_working_dir": "/home/env-admin/elastic",
                  "org_label-schema_url": "https://www.elastic.co/products/elasticsearch"
                }
              },
              "agent": {
                "name": "elastic",
                "id": "beeaa337-bee6-4614-a7ef-196ba0e11121",
                "type": "filebeat",
                "ephemeral_id": "2946f2a4-2394-4178-b2ea-bccbc1c72361",
                "version": "8.8.0"
              },
              "process": {
                "thread": {
                  "name": "elasticsearch[elastic][masterService#updateTask][T#167]"
                }
              },
              "log": {
                "file": {
                  "path": "/var/lib/docker/containers/85a2ffa2853d3c9b114626f3f3c9003b08a911494a7edada623d725c1f249e8a/85a2ffa2853d3c9b114626f3f3c9003b08a911494a7edada623d725c1f249e8a-json.log"
                },
                "offset": 1599400,
                "level": "INFO",
                "logger": "org.elasticsearch.cluster.metadata.MetadataMappingService"
              },
              "elastic_agent": {
                "id": "beeaa337-bee6-4614-a7ef-196ba0e11121",
                "version": "8.8.0",
                "snapshot": false
              },
              "sccoe": {
                "component": {
                  "name": "System Support",
                  "id": "SYS"
                }
              },
              "message": "update_mapping [_doc]",
              "input": {
                "type": "filestream"
              },
              "@timestamp": "2023-06-29T10:38:21.642Z",
              "ecs": {
                "version": "1.2.0"
              },
              "elasticsearch": {
                "node": {
                  "name": "elastic",
                  "id": "3QW8_ZQRRAOb8tDuwzGX4w"
                },
                "cluster": {
                  "name": "SCCoE",
                  "uuid": "uXtSsZ7KShykEYcdwZGqnQ"
                },
                "server": {
                  "tags": [
                    "elastic-agent",
                    "beats_input_codec_plain_applied"
                  ]
                },
                "index": {
                  "name": ".ds-metrics-prometheus.collector-sccoe-2023.06.29-000008",
                  "id": "nBWAS9AwRsO_eBlZaDJ_KA"
                }
              },
              "stream": "stdout",
              "data_stream": {
                "namespace": "sccoe",
                "type": ".ds",
                "dataset": "elasticsearch.server"
              },
              "service": {
                "name": "ES_ECS",
                "type": "elasticsearch"
              },
              "@version": "1",
              "host": {
                "hostname": "elastic",
                "os": {
                  "kernel": "4.18.0-193.28.1.el8_2.x86_64",
                  "codename": "Ootpa",
                  "name": "Red Hat Enterprise Linux",
                  "type": "linux",
                  "family": "redhat",
                  "version": "8.2 (Ootpa)",
                  "platform": "rhel"
                },
                "containerized": false,
                "ip": [
                  "172.16.108.165",
                  "fe80::28cf:f4ea:b97b:f1a4",
                  "172.17.0.1",
                  "fe80::42:73ff:fecb:e96a",
                  "10.200.0.1",
                  "fe80::42:c5ff:fe37:9d14",
                  "fe80::3c64:4cff:fe76:218",
                  "fe80::3846:24ff:fee6:780e",
                  "fe80::cab:d5ff:fed9:da27",
                  "fe80::c46e:a0ff:fe3c:a8e2",
                  "fe80::b483:42ff:febe:728f",
                  "fe80::38ea:5aff:fed8:1ea9",
                  "fe80::9874:62ff:fe7a:c2e0",
                  "fe80::6c60:e0ff:fe51:2abd"
                ],
                "name": "elastic",
                "id": "3QW8_ZQRRAOb8tDuwzGX4w",
                "mac": [
                  "00-50-56-BE-03-DB",
                  "02-42-73-CB-E9-6A",
                  "02-42-C5-37-9D-14",
                  "0E-AB-D5-D9-DA-27",
                  "3A-46-24-E6-78-0E",
                  "3A-EA-5A-D8-1E-A9",
                  "3E-64-4C-76-02-18",
                  "6E-60-E0-51-2A-BD",
                  "9A-74-62-7A-C2-E0",
                  "B6-83-42-BE-72-8F",
                  "C6-6E-A0-3C-A8-E2"
                ],
                "architecture": "x86_64"
              },
              "event": {
                "agent_id_status": "auth_metadata_missing",
                "ingested": "2023-06-29T10:48:46.809808694Z",
                "created": "2023-06-29T10:38:21.642Z",
                "kind": "event",
                "message": {
                  "original": "{\"@timestamp\":\"2023-06-29T10:38:21.642Z\", \"log.level\": \"INFO\", \"message\":\"[.ds-metrics-prometheus.collector-sccoe-2023.06.29-000008/nBWAS9AwRsO_eBlZaDJ_KA] update_mapping [_doc]\", \"ecs.version\": \"1.2.0\",\"service.name\":\"ES_ECS\",\"event.dataset\":\"elasticsearch.server\",\"process.thread.name\":\"elasticsearch[elastic][masterService#updateTask][T#167]\",\"log.logger\":\"org.elasticsearch.cluster.metadata.MetadataMappingService\",\"elasticsearch.cluster.uuid\":\"uXtSsZ7KShykEYcdwZGqnQ\",\"elasticsearch.node.id\":\"3QW8_ZQRRAOb8tDuwzGX4w\",\"elasticsearch.node.name\":\"elastic\",\"elasticsearch.cluster.name\":\"SCCoE\"}\n"
                },
                "category": [
                  "database"
                ],
                "type": [
                  "info"
                ],
                "dataset": "elasticsearch.server",
              }
            },
            "_ingest": {
              "pipeline": "_simulate_pipeline",
              "timestamp": "2023-06-29T10:48:46.809808694Z"
            }
          }
        }
      ]
    }
  ]
}
6

Hi @i.raisr

What version are you on?

Having trouble reproducing ... as I can not see the actual command you ran, nor the input docs etc.

Can you create a gist or pastebin with a complete example, input, commands outputs etc. not just parts?

7

Hi @stephenb,

Thank you for reply.
This is Elastic stack 8.8.0, on-premise.

Please find the information you requested at the following pastebin links:

The pipeline processors involved in this flow are depicted here:

image
image1629×501 46.5 KB

Let me know if there is anything else I can provide.
Kind regards,
Ivo

8

Hi @i.raisr

I am looking at this ... you said it worked for another document can you provide that document as well... I think I see the issue I just want to confirm.. there may be a bug...

I can reproduce at this point... I want to see the "working" case

I did get a short fix / workaround to work like this manually setting the destination
Seems the bug / issues is that it is reading the actuall index name for the type not the dataset values

{
      "reroute": {
        "if": "(ctx?.host?.name == 'sso' && ctx?.container?.name == 'sso-app') || (ctx?.host?.name == 'elastic' && ctx?.container?.name == 'elasticsearch')",
        "ignore_failure": false,
        "tag": "keycloak",
        "destination" : "logs-elasticsearch.server-sccoe"
      }
    }

I am chatting with internals... looking perhaps a bug...

I would set the full destination as a workaround for the time being.

9

Hi @stephenb,

Thank you for looking into this matter.
Surprisingly, I can no longer reproduce the "working" case for keycloak events.
I tried with various documents but all of them end up with incorrect data_stream.type equal to .ds (instead of logs).

If that is indeed a bug, I will await once it is fixed in an Elastic release.
Kind regards,
Ivo

10

Hi @i.raisr

So talking to the team ... The bug is in _simulate but does work when posting to the actual/real data stream

So the _simulate has a bug that is probably why you can not reproduce the working version..

So the little simulate / test window in the Kibana UI will not work, but when it runs it should work assuming there are no other issues.

See if you can follow this... this is run and posting to an actual data stream and the reroute works.

Also you can always set the entire destination as I showed above that seems to work for both _simulate and actually indexing into a data stream

DELETE _data_stream/logs-elasticsearch.server-sccoe

PUT _ingest/pipeline/reroute-test
{
  "processors": [
    {
        "set": {
          "field": "data_stream.dataset",
          "value": "elasticsearch.server",
          "ignore_failure": false
        }
      },
    {
      "reroute": {
        "ignore_failure": false
      }
    }
  ]
}

# Test normal direct message 
POST logs-elasticsearch.server-sccoe/_doc
{
  "message": "message 1 - Direct ",
  "input": {
    "type": "filestream"
  },
  "@timestamp": "2023-06-30T06:33:03.798Z",
  "ecs": {
    "version": "8.0.0"
  },
  "stream": "stdout",
  "data_stream": {
    "namespace": "sccoe",
    "type": "logs",
    "dataset": "elasticsearch.server"
  }
}

# Post rerouted doc
POST logs-docker.container_logs-sccoe/_doc?pipeline=reroute-test
{
  "message": "message re-route1",
  "input": {
    "type": "filestream"
  },
  "@timestamp": "2023-06-30T06:33:03.798Z",
  "ecs": {
    "version": "8.0.0"
  },
  "stream": "stdout",
  "data_stream": {
    "namespace": "sccoe",
    "type": "logs",
    "dataset": "docker.container_logs"
  }
}

# Note this result looks good
{
  "_index": ".ds-logs-elasticsearch.server-sccoe-2023.06.30-000001",
  "_id": "86haDYkBI8KwcWx0ojtu",
  "_version": 1,
  "result": "created",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "_seq_no": 1,
  "_primary_term": 1
}

# Then Run Search and They Are There!!
GET logs-elasticsearch.server-sccoe/_search

# Result both are there

{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 2,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": ".ds-logs-elasticsearch.server-sccoe-2023.06.30-000001",
        "_id": "8qhaDYkBI8KwcWx0lTvG",
        "_score": 1,
        "_source": {
          "message": "message 1 - Direct ",
          "input": {
            "type": "filestream"
          },
          "@timestamp": "2023-06-30T06:33:03.798Z",
          "ecs": {
            "version": "8.0.0"
          },
          "stream": "stdout",
          "data_stream": {
            "namespace": "sccoe",
            "type": "logs",
            "dataset": "elasticsearch.server"
          }
        }
      },
      {
        "_index": ".ds-logs-elasticsearch.server-sccoe-2023.06.30-000001",
        "_id": "86haDYkBI8KwcWx0ojtu",
        "_score": 1,
        "_source": {
          "input": {
            "type": "filestream"
          },
          "@timestamp": "2023-06-30T06:33:03.798Z",
          "ecs": {
            "version": "8.0.0"
          },
          "message": "message re-route1",
          "stream": "stdout",
          "data_stream": {
            "namespace": "sccoe",
            "type": "logs",
            "dataset": "elasticsearch.server"
          }
        }
      }
    ]
  }
}

And note with full "destination" set it works for both _simulate and real indexing into the data stream

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "processors": [
      {
        "set": {
          "field": "data_stream.dataset",
          "value": "elasticsearch.server",
          "ignore_failure": false
        }
      },
      {
        "reroute": {
          "destination" : "logs-elasticsearch.server-sccoe",
          "ignore_failure": false
        }
      }
      ]
  }, 
  "docs": [
    {
      "_id": "-NYBC4kBqzI6pGDGasKv",
      "_index": "logs-docker.container_logs-sccoe-2023.06.29-000013",
      "_source": {
        "message": "{\"@timestamp\":\"2023-06-30T06:33:03.795Z\", \"log.level\": \"INFO\", \"message\":\"[.ds-logs-endpoint.events.process-sccoe-2023.06.29-000022/LCINFinVQ2OIt329g2Q-Wg] deleting index\", \"ecs.version\": \"1.2.0\",\"service.name\":\"ES_ECS\",\"event.dataset\":\"elasticsearch.server\",\"process.thread.name\":\"elasticsearch[elastic][masterService#updateTask][T#107]\",\"log.logger\":\"org.elasticsearch.cluster.metadata.MetadataDeleteIndexService\",\"trace.id\":\"a5beabd55bce7f0361141b569a7721a7\",\"elasticsearch.cluster.uuid\":\"uXtSsZ7KShykEYcdwZGqnQ\",\"elasticsearch.node.id\":\"3QW8_ZQRRAOb8tDuwzGX4w\",\"elasticsearch.node.name\":\"elastic\",\"elasticsearch.cluster.name\":\"SCCoE\"}\n",
        "input": {
          "type": "filestream"
        },
        "@timestamp": "2023-06-30T06:33:03.798Z",
        "ecs": {
          "version": "8.0.0"
        },
        "stream": "stdout",
        "data_stream": {
          "namespace": "sccoe",
          "type": "logs",
          "dataset": "docker.container_logs"
        }
      }
    }
    ]
}

# Result 

{
  "docs": [
    {
      "doc": {
        "_index": "logs-elasticsearch.server-sccoe",
        "_id": "-NYBC4kBqzI6pGDGasKv",
        "_version": "-3",
        "_source": {
          "input": {
            "type": "filestream"
          },
          "@timestamp": "2023-06-30T06:33:03.798Z",
          "ecs": {
            "version": "8.0.0"
          },
          "message": """{"@timestamp":"2023-06-30T06:33:03.795Z", "log.level": "INFO", "message":"[.ds-logs-endpoint.events.process-sccoe-2023.06.29-000022/LCINFinVQ2OIt329g2Q-Wg] deleting index", "ecs.version": "1.2.0","service.name":"ES_ECS","event.dataset":"elasticsearch.server","process.thread.name":"elasticsearch[elastic][masterService#updateTask][T#107]","log.logger":"org.elasticsearch.cluster.metadata.MetadataDeleteIndexService","trace.id":"a5beabd55bce7f0361141b569a7721a7","elasticsearch.cluster.uuid":"uXtSsZ7KShykEYcdwZGqnQ","elasticsearch.node.id":"3QW8_ZQRRAOb8tDuwzGX4w","elasticsearch.node.name":"elastic","elasticsearch.cluster.name":"SCCoE"}
""",
          "stream": "stdout",
          "data_stream": {
            "namespace": "sccoe",
            "type": "logs",
            "dataset": "elasticsearch.server"
          }
        },
        "_ingest": {
          "timestamp": "2023-06-30T17:40:52.109740546Z"
        }
      }
    }
  ]
}
11

Thanks @i.raisr for finding this!
Issue

12

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