[Elasticsearch]: Ingest pipeline created to process Elasticserver logs truncates log messages

[kvalliyurnatt]

Integration Name

Elasticsearch [elasticsearch]

Dataset Name

No response

Integration Version

1.16.0

Agent Version

8.17.0

Agent Output Type

elasticsearch

Elasticsearch Version

8.17.0

OS Version and Architecture

cloud

Software/API Version

No response

Error Message

No response

Event Original

{"@timestamp":"2025-01-22T15:50:04.517Z", "log.level": "INFO", "message":"[co.elastic.elasticsearch.metering.sampling.SampledStorageMetricsProvider] is not ready for collect yet", "ecs.version": "1.2.0","service.name":"ES_ECS","event.dataset":"elasticsearch.server","process.thread.name":"elasticsearch[es-es-search-c886b6975-9tsbz][metering_reporter][T#1]","log.logger":"co.elastic.elasticsearch.metering.usagereports.UsageReportCollector","elasticsearch.cluster.uuid":"vaYFhXk-Q0WFOdSjyaylnA","elasticsearch.node.id":"fnqA_AXpTFm6uZdtXDNIPA","elasticsearch.node.name":"es-es-search-c886b6975-9tsbz","elasticsearch.cluster.name":"es"}

What did you do?

inputs:
  # Logs: Collect audit, deprecation, gc, server and slowlog logs from Elasticsearch instances
  - id: elasticsearch-logfile
    type: logfile
    streams:
      # Audit logs: Collect Elasticsearch audit logs using log input
      - id: logfile-elasticsearch.audit
        data_stream:
          dataset: elasticsearch.audit
          type: logs
        paths:
          - /var/log/elasticsearch/*_audit.json
        exclude_files:
          - .gz$
        processors:
          - add_locale: null
          - add_fields:
              target: ''
              fields:
                ecs.version: 1.10.0
          - decode_json_fields:
              fields:
                - message
              target: _json
          - rename:
              fields:
                - from: _json.request.body
                  to: _request
              ignore_missing: true
          - drop_fields:
              fields:
                - _json
          - detect_mime_type:
              field: _request
              target: http.request.mime_type
          - drop_fields:
              fields:
                - _request
              ignore_missing: true
        # condition: <CONDITION> # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Deprecation logs: Collect Elasticsearch deprecation logs using log input
      - id: logfile-elasticsearch.deprecation
        data_stream:
          dataset: elasticsearch.deprecation
          type: logs
        paths:
          - /var/log/elasticsearch/*_deprecation.json
        exclude_files:
          - .gz$
          - _slowlog.log$
          - _access.log$
        # condition: <CONDITION> # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Garbage Collection logs: Collect Elasticsearch gc logs using log input
      - id: logfile-elasticsearch.gc
        data_stream:
          dataset: elasticsearch.gc
          type: logs
        paths:
          - '/var/log/elasticsearch/gc.log.[0-9]*'
          - /var/log/elasticsearch/gc.log
        exclude_files:
          - .gz$
        exclude_lines:
          - '^(OpenJDK|Java HotSpot).* Server VM '
          - '^CommandLine flags: '
          - '^Memory: '
          - '^{'
        multiline:
          pattern: '^(\[?[0-9]{4}-[0-9]{2}-[0-9]{2}|{)'
          negate: true
          match: after
        processors:
          - add_fields:
              target: ''
              fields:
                ecs.version: 1.10.0
        # condition: <CONDITION> # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Server logs: Collect server logs using log input
      - id: logfile-elasticsearch.server
        data_stream:
          dataset: elasticsearch.server
          type: logs
        paths:
          - /var/log/elasticsearch/*_server.json
        exclude_files:
          - .gz$
          - _slowlog.log$
          - _access.log$
          - _deprecation.log$
        # condition: <CONDITION> # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Slowlog logs: Collect Elasticsearch slowlog logs using log input
      - id: logfile-elasticsearch.slowlog
        data_stream:
          dataset: elasticsearch.slowlog
          type: logs
        paths:
          - /var/log/elasticsearch/*_index_search_slowlog.json
          - /var/log/elasticsearch/*_index_indexing_slowlog.json
        exclude_files:
          - .gz$
        # condition: <CONDITION> # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
  # Metrics (Stack Monitoring): Collect index, CCR, cluster stats, machine learning and node metrics from Elasticsearch instances to power the Stack Monitoring application in Kibana
  - id: elasticsearch-elasticsearch/metrics
    type: elasticsearch/metrics
    streams:
      # CCR metrics: Collect Elasticsearch Cross Cluster Replication metrics
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.ccr
        data_stream:
          dataset: elasticsearch.stack_monitoring.ccr
          type: metrics
        metricsets:
          - ccr
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Cluster stats: Collect Elasticsearch cluster wide metrics
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.cluster_stats
        data_stream:
          dataset: elasticsearch.stack_monitoring.cluster_stats
          type: metrics
        metricsets:
          - cluster_stats
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Enrich policies metrics: Collect Elasticsearch enrich policies stats
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.enrich
        data_stream:
          dataset: elasticsearch.stack_monitoring.enrich
          type: metrics
        metricsets:
          - enrich
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Index metrics: Collect Elasticsearch index metrics
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.index
        data_stream:
          dataset: elasticsearch.stack_monitoring.index
          type: metrics
        metricsets:
          - index
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Index recovery metrics: Returns information about ongoing and completed shard recoveries for one or more indices.
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.index_recovery
        data_stream:
          dataset: elasticsearch.stack_monitoring.index_recovery
          type: metrics
        metricsets:
          - index_recovery
        hosts:
          - 'http://localhost:9200'
        index_recovery.active_only: true
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Index summary metrics: Collect summaries of Elasticsearch index metrics
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.index_summary
        data_stream:
          dataset: elasticsearch.stack_monitoring.index_summary
          type: metrics
        metricsets:
          - index_summary
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Ingest Pipeline metrics: Collect metrics on Ingest Pipelines
      - id: elasticsearch/metrics-elasticsearch.ingest_pipeline
        data_stream:
          dataset: elasticsearch.ingest_pipeline
          type: metrics
        metricsets:
          - ingest_pipeline
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: null
        timeout: 10s
        ssl: null
        ingest_pipeline.processor_sample_rate: 0.25
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Anomaly detection machine learning job metrics: Collect usage information for anomaly detection Machine Learning jobs
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.ml_job
        data_stream:
          dataset: elasticsearch.stack_monitoring.ml_job
          type: metrics
        metricsets:
          - ml_job
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Node metrics: Collect node metrics from Elasticsearch
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.node
        data_stream:
          dataset: elasticsearch.stack_monitoring.node
          type: metrics
        metricsets:
          - node
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Cluster nodes statistics: Collect node stats from Elasticsearch
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.node_stats
        data_stream:
          dataset: elasticsearch.stack_monitoring.node_stats
          type: metrics
        metricsets:
          - node_stats
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Pending tasks metrics: Collect cluster-level changes that have not yet been executed.
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.pending_tasks
        data_stream:
          dataset: elasticsearch.stack_monitoring.pending_tasks
          type: metrics
        metricsets:
          - pending_tasks
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input
        # tags:
        #  - <TAGS> # Tags
      # Shards metrics: Collect Elasticsearch shard metrics
      - id: elasticsearch/metrics-elasticsearch.stack_monitoring.shard
        data_stream:
          dataset: elasticsearch.stack_monitoring.shard
          type: metrics
        metricsets:
          - shard
        hosts:
          - 'http://localhost:9200'
        scope: node
        period: 10s
        timeout: 10s
        ssl: null
        # username: <USERNAME> # Username: Use when connecting to elasticsearch
        # password: <PASSWORD> # Password: Use when connecting to elasticsearch
        # api_key: <API_KEY> # API Key: Elasticsearch API Key in Beats format. Use when connecting to elasticsearch in place of username/password.
        # condition: '${kubernetes_leaderelection.leader} == true and <CONDITION>' # Condition: Condition to filter when to collect this input

What did you see?

We are seeing the message field get truncated if it begins with a []. The text in the [] is being moved to a field called elasticsearch.index.name

This seems to be because of the grok pattern defined here: https://github.com/elastic/integrations/blob/main/packages/elasticsearch/data_stream/server/elasticsearch/ingest_pipeline/pipeline-json.yml

That moves things inside the [] as an index name

What did you expect to see?

We expected messages that begin with [] to be retained within the message field.

Anything else?

No response

[consulthys]

I was able to trace the ingest pipeline back to the addition of the parsing of index names to this commit... in 2018!!

The pipeline evolved over the years from handling raw text logs to structured JSON logs and was ported from the elasticsearch module in Filebeat to the elasticsearch Elastic Agent integration, but the grok pattern didn't change much since then. The ES logs must have changed over the years to include more non-index related data between [] brackets, which is perfectly understandable and would make total sense.

Whether this qualifies as a bug or not is up for debate. Obviously, we're now parsing content in square brackets as index names, which they are not always.

We're currently not sure how to handle this without breaking all the known (and unknown, read "customer") downstream processes that might build upon the elasticsearch.index.name field. We'll discuss this internally and get back here with options, but feel free to chime in if you have ideas.

[rjernst]

This seems like a bug to me. Whether or not users are "relying" on the bracket extraction, it's most likely wrong, so the values they are getting are not index names. For example, consider the following message:

Failed to [create/update] built-in role [machine_learning_admin]

Will the first or second set of brackets be used? In either case, this is not an index name, so anything a user is doing with it is wrong, or not what they really want. Additioanlly, if the second set of brackets is used, the message will be effectively truncated to an empty string. That isn't useful to users in using our logs for debugging Elasticsearch, nor to ES engineers in trying to understand what is going on.

[consulthys]

Thanks @rjernst
I totally agree and I was trying to understand what context in 2018 led to think that whatever is in brackets would be an index name. I'm not even sure this was the case back then. We'll try to figure out how to change this with the least impact.

[rjernst]

In non-json logs it is a thing. Even today, when we log something specific to an index, the index name gets placed in brackets (similar to how we have the node name in brackets). But it is context dependent, and I'm not sure it is done consistently (it requires getting a logger specific to that index). It also is not output, afaict, in json logs (it's not part of the message, so unless a dev explicitly puts the index name in the message, it wont' be seen there). Does ECS have a field for the index name?

[consulthys]

In non-json logs it is a thing.

Indeed, that grok pattern was created back then when JSON logs didn't exist, so that might explain.

it requires getting a logger specific to that index

Yes, through a PrefixLogger but it's hard, if at all possible, to know in which context and for which type of "marker" the logger was created.

Does ECS have a field for the index name?

In standard ECS fields, there isn't as far as I can tell. The elasticsearch Filebeat and Metricbeat modules have a specific elasticsearch section where the index name is usually stored in the elasticsearch.index.name sub-field.

Are you trying to see if for JSON logs ECSJsonLayout.additionalFields() could maybe provide such a field since there are similar elasticsearch.cluster.name/id and elasticsearch.node.name/id fields?

[kvalliyurnatt]

Just want to clarify one thing, the specific issue we are currently seeing is that it only replaces the things in the [] only if the message field begins with it. so for example if the message is the following

[create/update] built-in role [machine_learning_admin]

then create is being moved into elasticsearch.index.name and update to elasticsearch.index.id the message field has built-in role [machine_learning_admin]

For the message that @rjernst provided

Failed to [create/update] built-in role [machine_learning_admin]

there will be no truncation and the message will be displayed as it is. This is the behavior I saw in my testing.

This is still an issue that we want to be fixed.

[consulthys]

So ideally, we could just remove the [] parsing pattern in the JSON pipeline and leave the one for non-JSON logs