Include pipeline and plugin IDs to the JSON logs.

[mashhurs]

Release notes

Include pipeline and plugin IDs to the JSON logs.

What does this PR do?

Adds pipeline and plugin IDs to the JSON logs.

Why is it important/What is the impact to the user?

It crucial to know which plugin from which pipeline is generating a specific logs, especially errors. For now, when using JSON log format with Logstash, it is hard to figure out where the errors (or particular log lines) are coming from.
After this PR, logs provide more visibility for the case which plugin and pipeline are generating the logs.

Examples,

• normal info logs

// BEFORE
{"level":"INFO","loggerName":"org.logstash.beats.Server","timeMillis":1765220975362,"thread":"[main]<beats","logEvent":{"message":"Starting server on port: 5044"}}

// AFTER this change
{"level":"INFO","loggerName":"org.logstash.beats.Server","timeMillis":1765228334380,"thread":"[main]<beats","pipeline.id":"main","plugin.id":"f935c783dab1e6bdf43dd9619dfc90694990fcfa214b1d1d68a50d5497e9bd63","logEvent":{"message":"Starting server on port: 5044"}}

• error or warn logs

// BEFORE
{"level":"WARN","loggerName":"logstash.filters.mutate","timeMillis":1765220156385,"thread":"[main]>worker0","logEvent":{"message":"Exception caught while applying mutate filter","exception":"invalid value for BigDecimal(): \"123e\""}}

// AFTER this change
{"level":"INFO","loggerName":"org.logstash.beats.Server","timeMillis":1765228573066,"thread":"[main]<beats","pipeline.id":"main","plugin.id":"elastic-agent-id","logEvent":{"message":"Starting server on port: 5044"}}

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• [ ] I have made corresponding changes to the documentation
• [ ] I have made corresponding change to the default configuration files (and/or docker env variables)
• I have added tests that prove my fix is effective or that my feature works

How to test this PR locally

• pull this change and compile Logstash: ./gradlew clean bootstrap assemble installDefaultGems
• use any pipeline configs, for example:

// test.conf
input {
  elastic_agent {
    id => "elastic-agent-id"
     port => 5044
  }

  generator {
    id => "generator-id"
    count => 1
    codec => json
    lines => [
	'{"fileset":{"module":"system","name":"test", "module_id": "123e"}}'
    ]
  }
}

filter {
    if [fileset][module_id] {
        mutate { convert => { "[fileset][module_id]" => "integer" } }
    }
}


output {
  stdout { codec => rubydebug { metadata => true } }
}

• change default log format to json in logstash.yml, add log.format: json line
• Run Logstash: bin/logstash -f config/test.conf

Related issues

• Closes Improve json logging pipeline error reporting #18438

Use cases

Screenshots

Logs

» bin/logstash -f config/test.conf

Using system java: /usr/bin/java
Sending Logstash logs to /dev/logstash/logs which is now configured via log4j2.properties
{"level":"INFO","loggerName":"logstash.runner","timeMillis":1765228731532,"thread":"main","logEvent":{"message":"Log4j configuration path used is: /dev/logstash/config/log4j2.properties"}}
{"level":"WARN","loggerName":"deprecation.logstash.runner","timeMillis":1765228731548,"thread":"main","logEvent":{"message":"The use of JAVA_HOME has been deprecated. Logstash 8.0 and later ignores JAVA_HOME and uses the bundled JDK. Running Logstash with the bundled JDK is recommended. The bundled JDK has been verified to work with each specific version of Logstash, and generally provides best performance and reliability. If you have compelling reasons for using your own JDK (organizational-specific compliance requirements, for example), you can configure LS_JAVA_HOME to use that version instead."}}
{"level":"INFO","loggerName":"logstash.runner","timeMillis":1765228731549,"thread":"main","logEvent":{"message":"Starting Logstash","logstash.version":"9.3.0","jruby.version":"jruby 9.4.13.0 (3.1.4) 2025-06-10 9938a3461f OpenJDK 64-Bit Server VM 21.0.8+9-LTS on 21.0.8+9-LTS +indy +jit [arm64-darwin]"}}
{"level":"INFO","loggerName":"logstash.runner","timeMillis":1765228731549,"thread":"main","logEvent":{"message":"JVM bootstrap flags: [-Xms1g, -Xmx1g, -Djava.awt.headless=true, -Dfile.encoding=UTF-8, -XX:+HeapDumpOnOutOfMemoryError, -Djava.security.egd=file:/dev/urandom, -Djruby.regexp.interruptible=true, -Djruby.compile.invokedynamic=true, -Djdk.io.File.enableADS=true, -Dlog4j2.isThreadContextMapInheritable=true, --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED, --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED, --add-opens=java.base/java.security=ALL-UNNAMED, --add-opens=java.base/java.io=ALL-UNNAMED, --add-opens=java.base/java.nio.channels=ALL-UNNAMED, --add-opens=java.base/sun.nio.ch=ALL-UNNAMED, --add-opens=java.management/sun.management=ALL-UNNAMED, -Dio.netty.allocator.maxOrder=11]"}}
{"level":"INFO","loggerName":"org.logstash.jackson.StreamReadConstraintsUtil","timeMillis":1765228731562,"thread":"main","logEvent":{"message":"Jackson default value override `logstash.jackson.stream-read-constraints.max-string-length` configured to `200000000` (logstash default)"}}
{"level":"INFO","loggerName":"org.logstash.jackson.StreamReadConstraintsUtil","timeMillis":1765228731562,"thread":"main","logEvent":{"message":"Jackson default value override `logstash.jackson.stream-read-constraints.max-number-length` configured to `10000` (logstash default)"}}
{"level":"INFO","loggerName":"org.logstash.jackson.StreamReadConstraintsUtil","timeMillis":1765228731562,"thread":"main","logEvent":{"message":"Jackson default value override `logstash.jackson.stream-read-constraints.max-nesting-depth` configured to `1000` (logstash default)"}}
{"level":"WARN","loggerName":"logstash.config.source.multilocal","timeMillis":1765228731574,"thread":"LogStash::Runner","logEvent":{"message":"Ignoring the 'pipelines.yml' file because command line options are specified"}}
{"level":"INFO","loggerName":"logstash.agent","timeMillis":1765228731781,"thread":"Api Webserver","logEvent":{"message":"Successfully started Logstash API endpoint","port":9600,"ssl_enabled":false}}
{"level":"INFO","loggerName":"org.reflections.Reflections","timeMillis":1765228731865,"thread":"Converge PipelineAction::Create<main>","logEvent":{"message":"Reflections took 40 ms to scan 1 urls, producing 163 keys and 559 values"}}
{"level":"INFO","loggerName":"logstash.codecs.json","timeMillis":1765228731953,"thread":"Converge PipelineAction::Create<main>","logEvent":{"message":"ECS compatibility is enabled but `target` option was not specified. This may cause fields to be set at the top-level of the event where they are likely to clash with the Elastic Common Schema. It is recommended to set the `target` option to avoid potential schema conflicts (if your data is ECS compliant or non-conflicting, feel free to ignore this message)"}}
{"level":"INFO","loggerName":"logstash.javapipeline","timeMillis":1765228732006,"thread":"Converge PipelineAction::Create<main>","pipeline.id":"main","logEvent":{"message":"Pipeline `main` is configured with `pipeline.ecs_compatibility: v8` setting. All plugins in this pipeline will default to `ecs_compatibility => v8` unless explicitly configured otherwise."}}
{"level":"INFO","loggerName":"logstash.javapipeline","timeMillis":1765228732019,"thread":"[main]-pipeline-manager","pipeline.id":"main","logEvent":{"message":"Starting pipeline","pipeline_id":"main","pipeline.workers":12,"pipeline.batch.size":125,"pipeline.batch.delay":50,"pipeline.max_inflight":1500,"batch_metric_sampling":"minimal","pipeline.sources":["/dev/logstash/config/tests/temp.conf"],"thread":"#<Thread:0x4a190e85 /dev/logstash/logstash-core/lib/logstash/java_pipeline.rb:147 run>"}}
{"level":"INFO","loggerName":"logstash.javapipeline","timeMillis":1765228732274,"thread":"[main]-pipeline-manager","pipeline.id":"main","logEvent":{"message":"Pipeline Java execution initialization time","seconds":0.25}}
{"level":"INFO","loggerName":"logstash.inputs.beats","timeMillis":1765228732281,"thread":"[main]-pipeline-manager","pipeline.id":"main","logEvent":{"message":"Starting input listener","address":"0.0.0.0:5044"}}
{"level":"INFO","loggerName":"logstash.javapipeline","timeMillis":1765228732285,"thread":"[main]-pipeline-manager","pipeline.id":"main","logEvent":{"message":"Pipeline started","pipeline.id":"main"}}
{"level":"INFO","loggerName":"logstash.agent","timeMillis":1765228732290,"thread":"Agent thread","logEvent":{"message":"Pipelines running","count":1,"running_pipelines":["main"],"non_running_pipelines":[]}}
{"level":"INFO","loggerName":"org.logstash.beats.Server","timeMillis":1765228732313,"thread":"[main]<beats","pipeline.id":"main","plugin.id":"elastic-agent-id","logEvent":{"message":"Starting server on port: 5044"}}
{"level":"WARN","loggerName":"logstash.filters.mutate","timeMillis":1765228732409,"thread":"[main]>worker0","pipeline.id":"main","plugin.id":"b477c8323249b70d31bc9b34629db6aada967f4a32bd4d976e1ba35275c2a05f","logEvent":{"message":"Exception caught while applying mutate filter","exception":"invalid value for BigDecimal(): \"123e\""}}
{
    "@timestamp" => 2025-12-08T21:18:52.296795Z,
      "@version" => "1",
          "host" => {
        "name" => "Miks-M4"
    },
         "event" => {
        "sequence" => 0,
        "original" => "{\"fileset\":{\"module\":\"system\",\"name\":\"test\", \"module_id\": \"123e\"}}"
    },
       "fileset" => {
        "module_id" => "123e",
           "module" => "system",
             "name" => "test"
    },
          "tags" => [
        [0] "_mutate_error"
    ]
}

[github-actions]

🤖 GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)
• /run exhaustive tests : Run the exhaustive tests Buildkite pipeline.

[mergify]

This pull request does not have a backport label. Could you fix it @mashhurs? 🙏
To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit.
• If no backport is necessary, please add the backport-skip label

[mashhurs]

Review note: for the safety I have placed a condition but it exists as we ThreadContext.put("pipeline.id", pipeline_id) and I have confirmed the places we accessing from org.apache.logging.log4j.ThreadContext, example java_pipeline.rb#start

Util::with_logging_thread_context("pipeline.id" => pipeline_id) do
        collect_stats
        collect_dlq_stats
        initialize_flow_metrics
      end

Note that, plugin IDs will be placed in the thread context in worker logics.

[elasticmachine]

💛 Build succeeded, but was flaky

• Buildkite Build
• Commit: 39c1bf9

Failed CI Steps

• 🥼 IT Persistent Queues / part 1-of-6

History

• 💚 Build #3924 succeeded e11b77a

cc @mashhurs

[andsel]

LGTM

[mashhurs]

@Mergifyio backport 9.2

[mergify]

backport 9.2

✅ Backports have been created

• #18474 [9.2] (backport #18470) Include pipeline and plugin IDs to the JSON logs. has been created for branch 9.2