Add suppress_health_degradation stream config to prevent optional streams from degrading agent health#49511
Add suppress_health_degradation stream config to prevent optional streams from degrading agent health#49511Oddly wants to merge 3 commits into
Conversation
botelastic
Bot
added
the
needs_team
🤖 GitHub commentsJust comment with:
|
|
This pull request does not have a backport label.
To fixup this pull request, you need to add the backport labels for the needed
|
|
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:
📝 WalkthroughWalkthroughAdds a per-stream Suggested labelsTeam:Elastic-Agent-Data-Plane 🚥 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.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@x-pack/libbeat/management/unit.go`:
- Around line 369-375: The update() path currently only mutates
u.streamStates[*].suppressHealthDegradation and never recomputes the unit
aggregate health, leaving stale Failed/Degraded unit state; after you change
existing.suppressHealthDegradation in update(), force a recompute and publish of
the unit health so suppression flips take effect immediately—either invoke
updateStateForStream(...) for that stream in a way that bypasses the "ignore
same-state" check (add a force parameter or call a new helper), or implement a
small helper that scans u.streamStates to recalculate the aggregate health and
calls the existing publish/update routine to emit the new unit state.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: abb20d59-2b6e-4e8a-a10e-63d15e50284c
📒 Files selected for processing (2)
x-pack/libbeat/management/unit.gox-pack/libbeat/management/unit_test.go
Oddly
force-pushed
the
suppress-health-degradation
branch
from
dac7eef to
f2ff4f3
Compare
There was a problem hiding this comment.
♻️ Duplicate comments (1)
x-pack/libbeat/management/unit.go (1)
369-375:⚠️ Potential issue | 🔴 CriticalRecompute and publish unit health when suppression flips.
update()mutatesexisting.suppressHealthDegradationbut never recalculates/emits aggregate unit status. If a stream is alreadyDegraded/Failed, unit health can stay stale indefinitely because same-state stream updates are ignored later (Line 323).Proposed fix (minimal)
func (u *agentUnit) update(cu *client.Unit) { u.mtx.Lock() defer u.mtx.Unlock() u.softDeleted = false u.clientUnit = cu + suppressionChanged := false inputStatus := getStatus(cu.Expected().State) if u.inputLevelState.state != inputStatus { u.inputLevelState = unitState{ state: inputStatus, } } newStreamStates, newStreamIDs := getStreamStates(cu.Expected()) for key, state := range newStreamStates { if existing, exists := u.streamStates[key]; exists { - // Preserve current health state but update the suppressHealthDegradation flag - // in case the stream config changed. - existing.suppressHealthDegradation = state.suppressHealthDegradation + if existing.suppressHealthDegradation != state.suppressHealthDegradation { + suppressionChanged = true + } + existing.suppressHealthDegradation = state.suppressHealthDegradation u.streamStates[key] = existing continue } u.streamStates[key] = state } @@ switch { @@ } + + if suppressionChanged { + state, msg := u.calcState() + streamsPayload := make(map[string]interface{}, len(u.streamStates)) + for id, streamState := range u.streamStates { + streamsPayload[id] = map[string]interface{}{ + "status": getUnitState(streamState.state).String(), + "error": streamState.msg, + } + } + if err := u.clientUnit.UpdateState(getUnitState(state), msg, map[string]interface{}{"streams": streamsPayload}); err != nil { + u.logger.Warnf("failed to update state for input %s: %v", u.ID(), err) + } + } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@x-pack/libbeat/management/unit.go` around lines 369 - 375, In update(), when you mutate existing.suppressHealthDegradation for entries in u.streamStates, detect that the suppression flag flipped and then trigger the unit-level health recomputation/publish path so the aggregate unit health is recalculated and emitted; specifically, inside the loop that updates existing.suppressHealthDegradation (in update()), after assigning existing.suppressHealthDegradation = state.suppressHealthDegradation, call the same routine you use for stream-state transitions to recompute and publish aggregate unit health (i.e., the unit health recompute/publish codepath) so unit health doesn't remain stale when suppression toggles.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@x-pack/libbeat/management/unit.go`:
- Around line 369-375: In update(), when you mutate
existing.suppressHealthDegradation for entries in u.streamStates, detect that
the suppression flag flipped and then trigger the unit-level health
recomputation/publish path so the aggregate unit health is recalculated and
emitted; specifically, inside the loop that updates
existing.suppressHealthDegradation (in update()), after assigning
existing.suppressHealthDegradation = state.suppressHealthDegradation, call the
same routine you use for stream-state transitions to recompute and publish
aggregate unit health (i.e., the unit health recompute/publish codepath) so unit
health doesn't remain stale when suppression toggles.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: da31b39f-bea6-4a10-b13e-ec60181529fd
📒 Files selected for processing (2)
x-pack/libbeat/management/unit.gox-pack/libbeat/management/unit_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- x-pack/libbeat/management/unit_test.go
…eams from degrading agent health Adds a `suppress_health_degradation` config flag at the stream level in the management framework. When set, fetch failures are still logged and tracked but do not mark the agent as Degraded. This lets operators build broad Fleet policies with integrations that may not be running on every enrolled host, without those absent services dragging the entire agent into a degraded state. Unlike a per-input approach, this change lives in the health aggregation layer (`calcState()` in `unit.go`) so every input type gets the flag for free: CEL, httpjson, metricbeat modules, filebeat inputs, and any future inputs that report per-stream health. The flag is read from the stream's existing `Source` protobuf field — no proto changes or per-input modifications needed. Behaviour when `suppress_health_degradation: true`: - Input still retries every `period` - Errors are logged at ERROR level - Per-stream status still reports Degraded/Failed in the streams payload - The stream is excluded from the unit's aggregate health calculation - On recovery, stream status resets to Running as usual Tested on a live Elastic Agent 9.3.1 cluster across three input types (CEL, redis/metrics, nginx/metrics) with 9 permutations — all passed. Supersedes elastic#49492 Fixes: elastic/elastic-agent#12885
Oddly
force-pushed
the
suppress-health-degradation
branch
from
f2ff4f3 to
74e2856
Compare
Live cluster validationTested on a patched Elastic Agent 9.3.1 built from this branch via Fresh install (policy created from scratch each time):
In-place update (same policy updated via PUT without agent restart, exercises the
Cross-input types (three integrations on one agent policy):
|
pierrehilbert
added
the
Team:Elastic-Agent-Data-Plane
|
Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane) |
botelastic
Bot
removed
the
needs_team
|
/test |
|
You need to update/format the files. |
|
/test |
| // but their degraded/failed states do not drag the overall unit health down. | ||
| suppressHealth := false | ||
| if src := stream.GetSource(); src != nil { | ||
| if v, ok := src.GetFields()["suppress_health_degradation"]; ok { |
There was a problem hiding this comment.
We try to avoid using individual booleans like this, to make it easier to add more configuration later.
The pattern we are introducing for outputs puts configuration under a status_reporting object, which came from elastic/elastic-agent#10890. This would allow us to keep adding functionality, like the ability to mute specific types of errors in the future. We hopefully will introduce something like the what the OTel hostmetrics receiver has done for process metrics to our system/metrics receiver for example.
It would be great to follow a similar convention, with a status_reporting configuration object here. I also think we should increase the granularity and allow muting the degraded and failed states separately.
status_reporting:
report_degraded: false
report_failed: falseIt is also possible to have inputs without streams, where this configuration would apply at the input level. This could be an a follow up enhancement we do if this gets too complicated to handle in this PR. You would want the input level setting to be inherited into each stream when there are streams present, so that you can control the whole input easily without duplication in the case where there are many streams.
…gregation Rework the stream health suppression from a single flat boolean into a status_reporting config block with report_degraded and report_failed as separate fields. This makes it possible to choose independently whether degraded or failed streams affect the unit's aggregate health. Setting status_reporting on the input applies to all its streams by default. Individual streams can still override if they need different behaviour.
Oddly
force-pushed
the
suppress-health-degradation
branch
from
831a240 to
95b667f
Compare
|
I think the change is as you requested; The possibility to set this per input and/or stream(stream takes priority) for failure or degradation separately. However, when testing version 9.3.2 I found out that agentbeat was replaced by elastic-otel-collector in version 9.3.0, invalidating this approach beyond 9.3.0. Tested with an agent on a 8.19.10 cluster, in the following scenarios: DetailsStream-level config:
Live policy changes:
Input-level inheritance:
However, in my testing of version 9.3.2 I found out that agentbeat was replaced by elastic-otel-collector in version 9.3.0, invalidating this approach beyond 9.3.0. |
So yes we are in the middle of an architecture change, with two ways to do things here as we transition everything to running into the OpenTelemetry collector. The receiver variant used in 9.3+ is here: beats/x-pack/otel/status/reporter.go Lines 35 to 42 in f558378 The main problem is how to provide the input level configuration to it, as it doesn't have access to it the way the previous implementation did. It can be passed the beat configuration when it is created though. The BeatReceiver struct has access to the instance.Beat that has the configuration in it: beats/x-pack/libbeat/cmd/instance/receiver.go Lines 118 to 123 in f558378 |
|
It is possible to implement this instead in Elastic Agent, since it receives the status events for both the Beat sub-processes and the OpenTelemetry collector variants of this. This would allow this feature to work even for things that are not Beats, like Elastic Defend. The main disadvantage would be that it wouldn't work for a user using the EDOT collector without the Elastic Agent supervisor, but I think that is likely acceptable for now. |
|
A small correction, this already works in the 9.3+ OTEL just like we implemented in agentbeat. Earlier I tested the patched version, but that didn't work because it had a OTEL backend and actually gave a skewed result. So, in conclusion, I think we can just merge this PR for anything lower than 9.3 and have the same functionality in both backends. |
|
I haven't had a chance to double check that this works as described but I haven't forgotten about it. |
|
Hi @cmacknz, could you take a look at this change? |
|
In the upcoming 9.5.0 release we'll be executing most beats in the OpenTelemetry collector, so this change needs to be made in elastic-agent instead of beats so it'll still work and it'll work for things that are not Beats as mentioned in #49511 (comment). I still like this feature but we haven't had a chance to look at implementation at all on our end. |
The Coordinator health rollup (generateReportableState/hasState) now reads a per-unit status_reporting config from the unit's expected config. Setting report_degraded: false and/or report_failed: false excludes that unit from the agent's aggregate health, so broad policies with inputs that are not running on every enrolled host (or that fail) no longer drag the whole agent into a degraded state. The unit's own state is still reported per component; only the aggregate rollup ignores it. Works for both process (Beat) and OpenTelemetry-translated components since both flow through the Coordinator. Ports the approach from elastic/beats#49511 into elastic-agent per maintainer guidance so it also covers non-Beats components and the OTel collector runtime.
The Coordinator health rollup (generateReportableState/hasState) now reads a per-unit status_reporting config from the unit's expected config. Setting report_degraded: false and/or report_failed: false excludes that unit from the agent's aggregate health, so broad policies with inputs that are not running on every enrolled host (or that fail) no longer drag the whole agent into a degraded state. The unit's own state is still reported per component; only the aggregate rollup ignores it. Works for both process (Beat) and OpenTelemetry-translated components since both flow through the Coordinator. Ports the approach from elastic/beats#49511 into elastic-agent per maintainer guidance so it also covers non-Beats components and the OTel collector runtime.
The Coordinator health rollup (generateReportableState/hasState) now reads a per-unit status_reporting config from the unit's expected config. Setting report_degraded: false and/or report_failed: false excludes that unit from the agent's aggregate health, so broad policies with inputs that are not running on every enrolled host (or that fail) no longer drag the whole agent into a degraded state. The unit's own state is still reported per component; only the aggregate rollup ignores it. Works for both process (Beat) and OpenTelemetry-translated components since both flow through the Coordinator. Ports the approach from elastic/beats#49511 into elastic-agent per maintainer guidance so it also covers non-Beats components and the OTel collector runtime.
The Coordinator health rollup (generateReportableState/hasState) now reads a per-unit status_reporting config from the unit's expected config. Setting report_degraded: false and/or report_failed: false excludes that unit from the agent's aggregate health, so broad policies with inputs that are not running on every enrolled host (or that fail) no longer drag the whole agent into a degraded state. The unit's own state is still reported per component; only the aggregate rollup ignores it. Works for both process (Beat) and OpenTelemetry-translated components since both flow through the Coordinator. Ports the approach from elastic/beats#49511 into elastic-agent per maintainer guidance so it also covers non-Beats components and the OTel collector runtime.
|
@cmacknz Continuing work in elastic/elastic-agent#15192. |
The Coordinator health rollup (generateReportableState/hasState) now reads a per-unit status_reporting config from the unit's expected config. Setting report_degraded: false and/or report_failed: false excludes that unit from the agent's aggregate health, so broad policies with inputs that are not running on every enrolled host (or that fail) no longer drag the whole agent into a degraded state. The unit's own state is still reported per component; only the aggregate rollup ignores it. Works for both process (Beat) and OpenTelemetry-translated components since both flow through the Coordinator. Ports the approach from elastic/beats#49511 into elastic-agent per maintainer guidance so it also covers non-Beats components and the OTel collector runtime.
7 participants
Proposed commit message
Adds a
suppress_health_degradationconfig flag at the stream level in themanagement framework. When set, fetch failures from all inputs are still logged and tracked but do not mark the agent as Degraded. This lets operators build broad Fleet
policies with integrations that may not be running on every enrolled host,
without those absent services dragging the entire agent into a degraded state.
Unlike a per-input approach (which we tried first), this change lives in the health aggregation layer
(
calcState()inunit.go) so every input type gets the flag for free: CEL,httpjson, metricbeat modules, filebeat inputs, and any future inputs that
report per-stream health.
The flag is read from the stream's existing
Sourceprotobuf field.Behaviour when
suppress_health_degradation: true:periodUsage in Fleet integrations:
Users can also inject
suppress_health_degradation: truevia Fleet's AdvancedSettings YAML on any existing integration without package changes.
Tested on a live Elastic Agent 9.3.1 cluster across three input types (CEL,
redis/metrics, nginx/metrics) with 9 permutations:
Replaces #49492
Fixes: elastic/elastic-agent#12885