Add otel-specific outputController to libbeat pipeline#50075
Add otel-specific outputController to libbeat pipeline#50075faec merged 2 commits intoelastic:mainfrom
Conversation
faec
added
enhancement
backport-skip
botelastic
bot
added
needs_team
🤖 GitHub commentsJust comment with:
|
|
Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane) |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (5)
💤 Files with no reviewable changes (1)
🚧 Files skipped from review as they are similar to previous changes (3)
📝 WalkthroughWalkthroughThe changes separate OTel receiver wiring from the standard libbeat pipeline. The 🚥 Pre-merge checks | ✅ 2✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (2)
libbeat/publisher/pipeline/pipeline.go (1)
197-230: Add doc comment and consider nil-safety for monitors.Metrics.
NewForReceiveris exported but lacks documentation per coding guidelines.Unlike
New()(lines 163, 171-173), this doesn't guard againstmonitors.Metricsbeing nil before callingnewMetricsObserver. The current call site inbeat.goalways provides a non-nil registry, but as a public API, defensive handling would be safer.Suggested changes
+// NewForReceiver creates a Pipeline for use by Beats receivers running under +// the OTel Collector runtime. Unlike New(), it wires the otelconsumer output +// directly without requiring output factory registration. func NewForReceiver( beatInfo beat.Info, monitors Monitors, userQueueConfig conf.Namespace, settings Settings, ) (*Pipeline, error) { + var obs observer = nilObserver + if monitors.Metrics != nil { + obs = newMetricsObserver(monitors.Metrics) + } p := &Pipeline{ beatInfo: beatInfo, monitors: monitors, - observer: newMetricsObserver(monitors.Metrics), + observer: obs, waitCloseTimeout: settings.WaitClose,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@libbeat/publisher/pipeline/pipeline.go` around lines 197 - 230, Public function NewForReceiver lacks a doc comment and isn't defensive against a nil monitors.Metrics; add a Go doc comment describing purpose/parameters/return for NewForReceiver and update the function to check monitors.Metrics before calling newMetricsObserver (e.g., if monitors != nil && monitors.Metrics != nil { observer = newMetricsObserver(monitors.Metrics) } else { observer = newMetricsObserver(defaultRegistryOrNil) }) to mirror the nil-safety used by New(), ensuring observer is always initialized safely even when monitors or monitors.Metrics is nil.libbeat/otel/otelconsumer/otelconsumer.go (1)
60-60: Add doc comment for exported function.Per coding guidelines, exported functions should be documented. A brief description would help callers understand when to use this vs the standard output factory pattern.
Suggested doc comment
+// MakeOtelConsumer creates an output group that bridges Beat events to an +// OpenTelemetry Collector pipeline via the beat.LogConsumer. func MakeOtelConsumer(beat beat.Info, observer outputs.Observer, cfg *config.C, beatPaths *paths.Path) (outputs.Group, error) {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@libbeat/otel/otelconsumer/otelconsumer.go` at line 60, Add a Go doc comment for the exported function MakeOtelConsumer describing its purpose and usage: state that MakeOtelConsumer constructs an OpenTelemetry outputs.Group consumer for Beats (accepting beat.Info, outputs.Observer, *config.C, *paths.Path) and indicate when to prefer this factory over the standard output factory pattern (e.g., for OTEL-specific telemetry/export behavior). Place the comment immediately above the MakeOtelConsumer declaration and keep it concise, starting with the function name per Go conventions.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@libbeat/otel/otelconsumer/otelconsumer.go`:
- Line 60: Add a Go doc comment for the exported function MakeOtelConsumer
describing its purpose and usage: state that MakeOtelConsumer constructs an
OpenTelemetry outputs.Group consumer for Beats (accepting beat.Info,
outputs.Observer, *config.C, *paths.Path) and indicate when to prefer this
factory over the standard output factory pattern (e.g., for OTEL-specific
telemetry/export behavior). Place the comment immediately above the
MakeOtelConsumer declaration and keep it concise, starting with the function
name per Go conventions.
In `@libbeat/publisher/pipeline/pipeline.go`:
- Around line 197-230: Public function NewForReceiver lacks a doc comment and
isn't defensive against a nil monitors.Metrics; add a Go doc comment describing
purpose/parameters/return for NewForReceiver and update the function to check
monitors.Metrics before calling newMetricsObserver (e.g., if monitors != nil &&
monitors.Metrics != nil { observer = newMetricsObserver(monitors.Metrics) } else
{ observer = newMetricsObserver(defaultRegistryOrNil) }) to mirror the
nil-safety used by New(), ensuring observer is always initialized safely even
when monitors or monitors.Metrics is nil.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 91e8d4c1-fe5f-4ad8-b39f-9fbb457d81ab
📒 Files selected for processing (8)
libbeat/otel/otelconsumer/otelconsumer.golibbeat/publisher/includes/includes.golibbeat/publisher/pipeline/output_otel.golibbeat/publisher/pipeline/output_otel_test.golibbeat/publisher/pipeline/output_process.golibbeat/publisher/pipeline/output_process_test.golibbeat/publisher/pipeline/pipeline.gox-pack/libbeat/cmd/instance/beat.go
💤 Files with no reviewable changes (1)
- libbeat/publisher/includes/includes.go
faec
force-pushed
the
otel-output-controller
branch
from
97206de to
999d1de
Compare
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
libbeat/publisher/pipeline/pipeline.go (1)
160-170:⚠️ Potential issue | 🟠 Major
WaitCloseModeno longer disables wait-close in the normal pipeline path.Line 164 now copies
settings.WaitCloseunconditionally, while Line 169 only controls force-closing. WithWaitCloseMode == NoWaitOnClose,Pipeline.Close()will still wait inoutputController.waitClose(...)for up toWaitClose, which is the opposite of the documented mode semantics and changes shutdown behavior for non-receiver Beats.Suggested fix
p := &Pipeline{ beatInfo: beat, monitors: monitors, observer: nilObserver, - waitCloseTimeout: settings.WaitClose, + waitCloseTimeout: 0, processors: settings.Processors, paths: settings.Paths, } - p.forceCloseQueue = settings.WaitCloseMode == WaitOnPipelineCloseThenForce + switch settings.WaitCloseMode { + case WaitOnPipelineClose, WaitOnPipelineCloseThenForce: + p.waitCloseTimeout = settings.WaitClose + } + p.forceCloseQueue = settings.WaitCloseMode == WaitOnPipelineCloseThenForce🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@libbeat/publisher/pipeline/pipeline.go` around lines 160 - 170, The Pipeline currently always sets waitCloseTimeout from settings.WaitClose while forceCloseQueue is only set from settings.WaitCloseMode, which means WaitCloseMode==NoWaitOnClose does not disable waiting; update the Pipeline constructor (the Pipeline struct initialization and assignment of waitCloseTimeout/forceCloseQueue) so that when settings.WaitCloseMode == NoWaitOnClose you set waitCloseTimeout to zero (or otherwise disable waiting) and only set waitCloseTimeout from settings.WaitClose for modes that allow waiting; adjust the logic around Pipeline.waitCloseTimeout, Pipeline.forceCloseQueue, and any use in Pipeline.Close/outputController.waitClose to reflect this conditional behavior so NoWaitOnClose truly prevents waiting.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@libbeat/publisher/pipeline/pipeline.go`:
- Around line 160-170: The Pipeline currently always sets waitCloseTimeout from
settings.WaitClose while forceCloseQueue is only set from
settings.WaitCloseMode, which means WaitCloseMode==NoWaitOnClose does not
disable waiting; update the Pipeline constructor (the Pipeline struct
initialization and assignment of waitCloseTimeout/forceCloseQueue) so that when
settings.WaitCloseMode == NoWaitOnClose you set waitCloseTimeout to zero (or
otherwise disable waiting) and only set waitCloseTimeout from settings.WaitClose
for modes that allow waiting; adjust the logic around Pipeline.waitCloseTimeout,
Pipeline.forceCloseQueue, and any use in
Pipeline.Close/outputController.waitClose to reflect this conditional behavior
so NoWaitOnClose truly prevents waiting.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: b0c0ba61-9dec-4429-82b9-364e5b36a4a2
📒 Files selected for processing (8)
libbeat/otel/otelconsumer/otelconsumer.golibbeat/publisher/includes/includes.golibbeat/publisher/pipeline/output_otel.golibbeat/publisher/pipeline/output_otel_test.golibbeat/publisher/pipeline/output_process.golibbeat/publisher/pipeline/output_process_test.golibbeat/publisher/pipeline/pipeline.gox-pack/libbeat/cmd/instance/beat.go
💤 Files with no reviewable changes (1)
- libbeat/publisher/includes/includes.go
🚧 Files skipped from review as they are similar to previous changes (1)
- libbeat/publisher/pipeline/output_process.go
cmacknz
left a comment
cmacknz
left a comment
There was a problem hiding this comment.
LGTM pending Lee agreeing his comment is resolved.
4 participants
Add an OTel-specific implementation of
pipeline.outputController, and create it via a separate pipeline initialization API for use with Beats receivers. Most of the core logic is unchanged in this pass (in particular this should cause no change in behavior), but the code paths are shorter/simpler than the ones for the process runtime since there are many complications and corner cases that aren't relevant when running as a Beats receiver (in particular Beats receivers do not need to dynamically reload their output configuration).Checklist
stresstest.shscript to run them under stress conditions and race detector to verify their stability../changelog/fragmentsusing the changelog tool.Related issues
otelconsumerfrom standard libbeat pipeline setup #49051