add pipeline id to thread context during init/start

[yaauie]

Release notes

• improved pipeline-aware logging during pipeline startup

What does this PR do?

Executes pipeline-initialization and pipeline-startup code with the pipeline's id in the thread logging context

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

When logging is configured to show the pipeline id from the thread context (as it is by default), the messages related to pipeline startup include the pipeline's id.

Additionally, if per-pipeline logging is enabled, these messages will now be routed to the pipeline's isolated logs.

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

Single logs:

bin/logstash --log.level=trace -Squeue.type=persisted --config.string="input { generator { count => 1 } }"

Pipeline-separated logs:

bin/logstash --log.level=trace -Squeue.type=persisted -Spipeline.separate_logs=true --config.string="input { generator { count => 1 } }"

Observe that logging includes the pipeline id.

Related issues

Use cases

Screenshots

Logs

Baseline

[2025-08-08T18:13:09,514][DEBUG][org.logstash.ackedqueue.QueueUpgrade] PQ version file with correct version information (v2) found.
[2025-08-08T18:13:09,515][DEBUG][org.logstash.ackedqueue.Queue] opening head page: 0, in: /Users/rye/src/elastic/logstash@main/data/queue/main, with checkpoint: pageNum=0, firstUnackedPageNum=0, firstUnackedSeqNum=2, minSeqNum=1, elementCount=1, isFullyAcked=yes
[2025-08-08T18:13:09,522][DEBUG][org.logstash.ackedqueue.io.MmapPageIOV2] PageIO recovery for '/Users/rye/src/elastic/logstash@main/data/queue/main/page.0' element index:1, readNextElement exception: Element seqNum 0 is expected to be 2
[2025-08-08T18:13:09,525][DEBUG][org.logstash.ackedqueue.io.MmapPageIOV2] PageIO deleting '/Users/rye/src/elastic/logstash@main/data/queue/main/page.0'
[2025-08-08T18:13:09,526][DEBUG][org.logstash.ackedqueue.io.FileCheckpointIO] CheckpointIO deleting '/Users/rye/src/elastic/logstash@main/data/queue/main/checkpoint.0'
[2025-08-08T18:13:09,527][DEBUG][org.logstash.ackedqueue.Queue] created new head page: MmapPageIOV2{file=/Users/rye/src/elastic/logstash@main/data/queue/main/page.1, capacity=67108864, minSeqNum=0, elementCount=0, head=1}
[2025-08-08T18:13:09,531][INFO ][logstash.javapipeline    ] 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.
[2025-08-08T18:13:09,533][TRACE][org.logstash.instrument.metrics.BaseFlowMetric] FlowMetric(input_throughput) baseline -> FlowCapture{nanoTimestamp=60991327904791 numerator=0.0 denominator=0.000123959}
[2025-08-08T18:13:09,534][DEBUG][org.logstash.execution.AbstractPipelineExt] Flow metric registered: `input_throughput` in namespace `[:stats, :pipelines, :main, :flow]`

With this change:

[2025-08-08T18:14:19,153][DEBUG][org.logstash.ackedqueue.QueueUpgrade][main] PQ version file with correct version information (v2) found.
[2025-08-08T18:14:19,154][DEBUG][org.logstash.ackedqueue.Queue][main] opening head page: 1, in: /Users/rye/src/elastic/logstash@main/data/queue/main, with checkpoint: pageNum=1, firstUnackedPageNum=1, firstUnackedSeqNum=3, minSeqNum=2, elementCount=1, isFullyAcked=yes
[2025-08-08T18:14:19,168][DEBUG][org.logstash.ackedqueue.io.MmapPageIOV2][main] PageIO recovery for '/Users/rye/src/elastic/logstash@main/data/queue/main/page.1' element index:1, readNextElement exception: Element seqNum 0 is expected to be 3
[2025-08-08T18:14:19,171][DEBUG][org.logstash.ackedqueue.io.MmapPageIOV2][main] PageIO deleting '/Users/rye/src/elastic/logstash@main/data/queue/main/page.1'
[2025-08-08T18:14:19,173][DEBUG][org.logstash.ackedqueue.io.FileCheckpointIO][main] CheckpointIO deleting '/Users/rye/src/elastic/logstash@main/data/queue/main/checkpoint.1'
[2025-08-08T18:14:19,174][DEBUG][org.logstash.ackedqueue.Queue][main] created new head page: MmapPageIOV2{file=/Users/rye/src/elastic/logstash@main/data/queue/main/page.2, capacity=67108864, minSeqNum=0, elementCount=0, head=1}
[2025-08-08T18:14:19,179][INFO ][logstash.javapipeline    ][main] 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.
[2025-08-08T18:14:19,181][TRACE][org.logstash.instrument.metrics.BaseFlowMetric][main] FlowMetric(input_throughput) baseline -> FlowCapture{nanoTimestamp=61060975164416 numerator=0.0 denominator=0.000139083}
[2025-08-08T18:14:19,182][DEBUG][org.logstash.execution.AbstractPipelineExt][main] Flow metric registered: `input_throughput` in namespace `[:stats, :pipelines, :main, :flow]`

[github-actions]

🤖 GitHub comments

Expand to view the GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)

[mergify]

This pull request does not have a backport label. Could you fix it @yaauie? 🙏
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

[yaauie]

review note: the pipeline_id method always returns a non-nil ruby-string pipeline id after the super-method in AbstractPipelineExt#initialize has run.

[elastic-sonarqube]

Quality Gate passed

Issues
0 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
No data about Coverage
No data about Duplication

See analysis details on SonarQube

[elasticmachine]

💚 Build Succeeded

• Buildkite Build
• Commit: 0966475

[donoghuc]

Verified logging context is appended as expected. The new --silent CLI flag makes this very nice testing locally :)