From 74e28566cc1058437441c50bcf2fa88d705f3a53 Mon Sep 17 00:00:00 2001 From: Sam Crauwels Date: Tue, 17 Mar 2026 09:06:03 +0100 Subject: [PATCH 1/3] Add suppress_health_degradation stream config to prevent optional streams from degrading agent health MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 #49492 Fixes: https://github.com/elastic/elastic-agent/issues/12885 --- x-pack/libbeat/management/unit.go | 58 +++- x-pack/libbeat/management/unit_test.go | 375 +++++++++++++++++++++++++ 2 files changed, 425 insertions(+), 8 deletions(-) diff --git a/x-pack/libbeat/management/unit.go b/x-pack/libbeat/management/unit.go index 729be62c77a7..9ec70102612c 100644 --- a/x-pack/libbeat/management/unit.go +++ b/x-pack/libbeat/management/unit.go @@ -15,8 +15,9 @@ import ( // unitState is the current state of a unit type unitState struct { - state status.Status - msg string + state status.Status + msg string + suppressHealthDegradation bool // when true, this stream's degraded/failed states do not affect the unit's aggregate health } type clientUnit interface { @@ -101,9 +102,20 @@ func getStreamStates(expected client.Expected) (map[string]unitState, []string) streamIDs := make([]string, len(expectedCfg.Streams)) for idx, stream := range expectedCfg.Streams { + // Check if stream is marked as suppressHealthDegradation in its source config. + // Optional streams still collect data and report per-stream status, + // 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 { + suppressHealth = v.GetBoolValue() + } + } + streamState := unitState{ - state: status.Unknown, - msg: "", + state: status.Unknown, + msg: "", + suppressHealthDegradation: suppressHealth, } if id := stream.GetId(); id != "" { @@ -216,10 +228,16 @@ func (u *agentUnit) calcState() (status.Status, string) { return u.inputLevelState.state, u.inputLevelState.msg } - // inputLevelState state is marked as running, check the stream states + // inputLevelState state is marked as running, check the stream states. + // Streams marked as suppressHealthDegradation are excluded from the aggregate health + // calculation — they still report per-stream status but do not cause + // the unit to be reported as degraded or failed. reportedStatus := status.Running reportedMsg := "Healthy" for _, streamState := range u.streamStates { + if streamState.suppressHealthDegradation { + continue + } switch streamState.state { case status.Degraded: if reportedStatus != status.Degraded { @@ -307,8 +325,9 @@ func (u *agentUnit) updateStateForStream(streamID string, state status.Status, m } u.streamStates[streamID] = unitState{ - state: state, - msg: msg, + state: state, + msg: msg, + suppressHealthDegradation: u.streamStates[streamID].suppressHealthDegradation, } state, msg = u.calcState() @@ -347,8 +366,16 @@ func (u *agentUnit) update(cu *client.Unit) { newStreamStates, newStreamIDs := getStreamStates(cu.Expected()) + suppressionChanged := false for key, state := range newStreamStates { - if _, exists := u.streamStates[key]; exists { + if existing, exists := u.streamStates[key]; exists { + // Preserve current health state but update the suppressHealthDegradation flag + // in case the stream config changed. + if existing.suppressHealthDegradation != state.suppressHealthDegradation { + suppressionChanged = true + } + existing.suppressHealthDegradation = state.suppressHealthDegradation + u.streamStates[key] = existing continue } @@ -372,6 +399,21 @@ func (u *agentUnit) update(cu *client.Unit) { } } } + + // If any stream's suppression flag changed, recompute and publish the + // aggregate unit health so that a flip from suppress=false→true (or + // vice versa) takes effect immediately. + 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, + } + } + _ = u.clientUnit.UpdateState(getUnitState(state), msg, map[string]interface{}{"streams": streamsPayload}) + } } func (u *agentUnit) markAsDeleted() { diff --git a/x-pack/libbeat/management/unit_test.go b/x-pack/libbeat/management/unit_test.go index 9684ff5e16e9..aae3bd883024 100644 --- a/x-pack/libbeat/management/unit_test.go +++ b/x-pack/libbeat/management/unit_test.go @@ -9,6 +9,7 @@ import ( "github.com/elastic/elastic-agent-client/v7/pkg/client" "github.com/elastic/elastic-agent-client/v7/pkg/proto" + "google.golang.org/protobuf/types/known/structpb" "github.com/elastic/beats/v7/libbeat/management/status" ) @@ -181,6 +182,367 @@ func TestUnitUpdate(t *testing.T) { } } +func TestUnitUpdateSuppressHealthDegradation(t *testing.T) { + + type StatusUpdate struct { + status status.Status + msg string + } + + const ( + Healthy = "Healthy" + Failed = "Failed" + Degraded = "Degraded" + ) + + suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ + "suppress_health_degradation": true, + }) + + notSuppressedSource, _ := structpb.NewStruct(map[string]interface{}{ + "suppress_health_degradation": false, + }) + + newUnit := func(streams []*proto.Stream) *mockClientUnit { + return &mockClientUnit{ + expected: client.Expected{ + Config: &proto.UnitExpectedConfig{ + Id: "input-1", + Streams: streams, + }, + }, + } + } + + cases := []struct { + name string + unit *mockClientUnit + inputLevelStatus StatusUpdate + streamStates map[string]StatusUpdate + expectedUnitStatus client.UnitState + expectedUnitMsg string + }{ + { + name: "suppressed stream degraded does not affect unit health", + unit: newUnit([]*proto.Stream{ + {Id: "stream-required"}, + {Id: "stream-suppressed", Source: suppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-required": {status.Running, Healthy}, + "stream-suppressed": {status.Degraded, Degraded}, + }, + expectedUnitStatus: client.UnitStateHealthy, + expectedUnitMsg: Healthy, + }, + { + name: "suppressed stream failed does not affect unit health", + unit: newUnit([]*proto.Stream{ + {Id: "stream-required"}, + {Id: "stream-suppressed", Source: suppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-required": {status.Running, Healthy}, + "stream-suppressed": {status.Failed, Failed}, + }, + expectedUnitStatus: client.UnitStateHealthy, + expectedUnitMsg: Healthy, + }, + { + name: "required stream degraded still affects unit health even with suppressed streams", + unit: newUnit([]*proto.Stream{ + {Id: "stream-required"}, + {Id: "stream-suppressed", Source: suppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-required": {status.Degraded, Degraded}, + "stream-suppressed": {status.Degraded, Degraded}, + }, + expectedUnitStatus: client.UnitStateDegraded, + expectedUnitMsg: Degraded, + }, + { + name: "required stream failed still affects unit health even with suppressed streams", + unit: newUnit([]*proto.Stream{ + {Id: "stream-required"}, + {Id: "stream-suppressed", Source: suppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-required": {status.Failed, Failed}, + "stream-suppressed": {status.Running, Healthy}, + }, + expectedUnitStatus: client.UnitStateFailed, + expectedUnitMsg: Failed, + }, + { + name: "all suppressed streams degraded and failed keeps unit healthy", + unit: newUnit([]*proto.Stream{ + {Id: "stream-suppressed-1", Source: suppressedSource}, + {Id: "stream-suppressed-2", Source: suppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-suppressed-1": {status.Degraded, Degraded}, + "stream-suppressed-2": {status.Failed, Failed}, + }, + expectedUnitStatus: client.UnitStateHealthy, + expectedUnitMsg: Healthy, + }, + { + name: "suppress false behaves same as not set", + unit: newUnit([]*proto.Stream{ + {Id: "stream-explicit-false", Source: notSuppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-explicit-false": {status.Degraded, Degraded}, + }, + expectedUnitStatus: client.UnitStateDegraded, + expectedUnitMsg: Degraded, + }, + { + name: "input level degraded is not affected by suppress flag", + unit: newUnit([]*proto.Stream{ + {Id: "stream-suppressed", Source: suppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Degraded, Degraded}, + streamStates: map[string]StatusUpdate{ + "stream-suppressed": {status.Running, Healthy}, + }, + expectedUnitStatus: client.UnitStateDegraded, + expectedUnitMsg: Degraded, + }, + { + name: "input level failed is not affected by suppress flag", + unit: newUnit([]*proto.Stream{ + {Id: "stream-suppressed", Source: suppressedSource}, + }), + inputLevelStatus: StatusUpdate{status.Failed, Failed}, + streamStates: map[string]StatusUpdate{ + "stream-suppressed": {status.Running, Healthy}, + }, + expectedUnitStatus: client.UnitStateFailed, + expectedUnitMsg: Failed, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + aUnit := newAgentUnit(c.unit, nil) + err := aUnit.UpdateState(c.inputLevelStatus.status, c.inputLevelStatus.msg, nil) + if err != nil { + t.Fatal(err) + } + + for id, state := range c.streamStates { + aUnit.updateStateForStream(id, state.status, state.msg) + } + + if c.unit.reportedState != c.expectedUnitStatus { + t.Errorf("expected unit status %s, got %s", c.expectedUnitStatus, c.unit.reportedState) + } + + if c.unit.reportedMsg != c.expectedUnitMsg { + t.Errorf("expected unit msg %q, got %q", c.expectedUnitMsg, c.unit.reportedMsg) + } + }) + } +} + +func TestGetStreamStatesParsesSuppress(t *testing.T) { + suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ + "suppress_health_degradation": true, + }) + notSuppressedSource, _ := structpb.NewStruct(map[string]interface{}{ + "suppress_health_degradation": false, + }) + + expected := client.Expected{ + Config: &proto.UnitExpectedConfig{ + Id: "input-1", + Streams: []*proto.Stream{ + {Id: "stream-plain"}, + {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-not-suppressed", Source: notSuppressedSource}, + }, + }, + } + + states, ids := getStreamStates(expected) + + if len(states) != 3 { + t.Fatalf("expected 3 stream states, got %d", len(states)) + } + if len(ids) != 3 { + t.Fatalf("expected 3 stream IDs, got %d", len(ids)) + } + if states["stream-plain"].suppressHealthDegradation { + t.Error("stream without Source should not have suppress set") + } + if !states["stream-suppressed"].suppressHealthDegradation { + t.Error("stream with suppress_health_degradation: true should have suppress set") + } + if states["stream-not-suppressed"].suppressHealthDegradation { + t.Error("stream with suppress_health_degradation: false should not have suppress set") + } +} + +func TestSuppressFlipRecomputesHealth(t *testing.T) { + // Simulates what update() does when a policy change flips the suppress + // flag on a stream that is already Degraded. This verifies the + // recompute-on-suppression-change path in update(). + + cu := &mockClientUnitWithPayload{ + mockClientUnit: mockClientUnit{ + expected: client.Expected{ + Config: &proto.UnitExpectedConfig{ + Id: "input-1", + Streams: []*proto.Stream{ + {Id: "stream-a"}, + }, + }, + }, + }, + } + + aUnit := newAgentUnit(cu, nil) + _ = aUnit.UpdateState(status.Running, "Healthy", nil) + + // Stream degrades — unit should be Degraded (suppress not set) + aUnit.updateStateForStream("stream-a", status.Degraded, "connection refused") + if cu.reportedState != client.UnitStateDegraded { + t.Fatalf("expected Degraded before flip, got %s", cu.reportedState) + } + + // Simulate a policy update that adds suppress_health_degradation: true. + // This is exactly what update() does: flip the flag, then recompute. + aUnit.mtx.Lock() + existing := aUnit.streamStates["stream-a"] + existing.suppressHealthDegradation = true + aUnit.streamStates["stream-a"] = existing + + state, msg := aUnit.calcState() + streamsPayload := make(map[string]interface{}, len(aUnit.streamStates)) + for id, ss := range aUnit.streamStates { + streamsPayload[id] = map[string]interface{}{ + "status": getUnitState(ss.state).String(), + "error": ss.msg, + } + } + _ = aUnit.clientUnit.UpdateState(getUnitState(state), msg, map[string]interface{}{"streams": streamsPayload}) + aUnit.mtx.Unlock() + + // Unit should now be Healthy — the suppression flip took effect + if cu.reportedState != client.UnitStateHealthy { + t.Errorf("expected Healthy after suppress flip, got %s", cu.reportedState) + } + + // Per-stream payload should still show Degraded + streams := cu.reportedPayload["streams"].(map[string]interface{}) + streamStatus := streams["stream-a"].(map[string]interface{}) + if streamStatus["status"] != client.UnitStateDegraded.String() { + t.Errorf("expected per-stream Degraded after flip, got %q", streamStatus["status"]) + } + + // Now flip suppress back to false — unit should go Degraded again + aUnit.mtx.Lock() + existing = aUnit.streamStates["stream-a"] + existing.suppressHealthDegradation = false + aUnit.streamStates["stream-a"] = existing + + state, msg = aUnit.calcState() + streamsPayload = make(map[string]interface{}, len(aUnit.streamStates)) + for id, ss := range aUnit.streamStates { + streamsPayload[id] = map[string]interface{}{ + "status": getUnitState(ss.state).String(), + "error": ss.msg, + } + } + _ = aUnit.clientUnit.UpdateState(getUnitState(state), msg, map[string]interface{}{"streams": streamsPayload}) + aUnit.mtx.Unlock() + + if cu.reportedState != client.UnitStateDegraded { + t.Errorf("expected Degraded after unsuppress flip, got %s", cu.reportedState) + } +} + +func TestSuppressedStreamRecovery(t *testing.T) { + suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ + "suppress_health_degradation": true, + }) + + cu := &mockClientUnit{ + expected: client.Expected{ + Config: &proto.UnitExpectedConfig{ + Id: "input-1", + Streams: []*proto.Stream{ + {Id: "stream-suppressed", Source: suppressedSource}, + }, + }, + }, + } + + aUnit := newAgentUnit(cu, nil) + _ = aUnit.UpdateState(status.Running, "Healthy", nil) + + // Stream goes degraded — unit should stay healthy + aUnit.updateStateForStream("stream-suppressed", status.Degraded, "connection refused") + if cu.reportedState != client.UnitStateHealthy { + t.Errorf("expected Healthy during degradation, got %s", cu.reportedState) + } + + // Stream recovers — unit should still be healthy + aUnit.updateStateForStream("stream-suppressed", status.Running, "Healthy") + if cu.reportedState != client.UnitStateHealthy { + t.Errorf("expected Healthy after recovery, got %s", cu.reportedState) + } +} + +func TestSuppressedStreamStillReportsPerStreamStatus(t *testing.T) { + suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ + "suppress_health_degradation": true, + }) + + cu := &mockClientUnitWithPayload{ + mockClientUnit: mockClientUnit{ + expected: client.Expected{ + Config: &proto.UnitExpectedConfig{ + Id: "input-1", + Streams: []*proto.Stream{ + {Id: "stream-suppressed", Source: suppressedSource}, + }, + }, + }, + }, + } + + aUnit := newAgentUnit(cu, nil) + _ = aUnit.UpdateState(status.Running, "Healthy", nil) + aUnit.updateStateForStream("stream-suppressed", status.Degraded, "connection refused") + + // Unit should be healthy (suppressed) + if cu.reportedState != client.UnitStateHealthy { + t.Errorf("expected unit Healthy, got %s", cu.reportedState) + } + + // But the per-stream payload should still show Degraded + streams, ok := cu.reportedPayload["streams"].(map[string]interface{}) + if !ok { + t.Fatal("expected streams in payload") + } + streamStatus, ok := streams["stream-suppressed"].(map[string]interface{}) + if !ok { + t.Fatal("expected stream-suppressed in streams payload") + } + if streamStatus["status"] != client.UnitStateDegraded.String() { + t.Errorf("expected per-stream status %q, got %q", client.UnitStateDegraded.String(), streamStatus["status"]) + } +} + type mockClientUnit struct { expected client.Expected reportedState client.UnitState @@ -213,3 +575,16 @@ func (u *mockClientUnit) UnregisterAction(_ client.Action) { func (u *mockClientUnit) RegisterDiagnosticHook(_ string, _ string, _ string, _ string, _ client.DiagnosticHook) { } + +// mockClientUnitWithPayload extends mockClientUnit to capture the payload +type mockClientUnitWithPayload struct { + mockClientUnit + reportedPayload map[string]interface{} +} + +func (u *mockClientUnitWithPayload) UpdateState(state client.UnitState, msg string, payload map[string]interface{}) error { + u.reportedState = state + u.reportedMsg = msg + u.reportedPayload = payload + return nil +} From 692bce95442ec2c33e94c041a561f463a43b304b Mon Sep 17 00:00:00 2001 From: Sam Crauwels Date: Sat, 21 Mar 2026 22:29:12 +0100 Subject: [PATCH 2/3] Fix gofmt alignment, goimports grouping, errcheck lint, and add changelog fragment --- .../1774128388-suppress-health-degradation.yaml | 7 +++++++ x-pack/libbeat/management/unit.go | 12 ++++++------ x-pack/libbeat/management/unit_test.go | 13 ++++++++++--- 3 files changed, 23 insertions(+), 9 deletions(-) create mode 100644 changelog/fragments/1774128388-suppress-health-degradation.yaml diff --git a/changelog/fragments/1774128388-suppress-health-degradation.yaml b/changelog/fragments/1774128388-suppress-health-degradation.yaml new file mode 100644 index 000000000000..5500f76e0387 --- /dev/null +++ b/changelog/fragments/1774128388-suppress-health-degradation.yaml @@ -0,0 +1,7 @@ +kind: feature + +summary: Add suppress_health_degradation stream config to prevent optional streams from degrading agent health + +component: libbeat + +pr: https://github.com/elastic/beats/pull/49511 diff --git a/x-pack/libbeat/management/unit.go b/x-pack/libbeat/management/unit.go index 9ec70102612c..59d02512ab6c 100644 --- a/x-pack/libbeat/management/unit.go +++ b/x-pack/libbeat/management/unit.go @@ -15,8 +15,8 @@ import ( // unitState is the current state of a unit type unitState struct { - state status.Status - msg string + state status.Status + msg string suppressHealthDegradation bool // when true, this stream's degraded/failed states do not affect the unit's aggregate health } @@ -113,8 +113,8 @@ func getStreamStates(expected client.Expected) (map[string]unitState, []string) } streamState := unitState{ - state: status.Unknown, - msg: "", + state: status.Unknown, + msg: "", suppressHealthDegradation: suppressHealth, } @@ -325,8 +325,8 @@ func (u *agentUnit) updateStateForStream(streamID string, state status.Status, m } u.streamStates[streamID] = unitState{ - state: state, - msg: msg, + state: state, + msg: msg, suppressHealthDegradation: u.streamStates[streamID].suppressHealthDegradation, } diff --git a/x-pack/libbeat/management/unit_test.go b/x-pack/libbeat/management/unit_test.go index aae3bd883024..c1e89a7a6ea1 100644 --- a/x-pack/libbeat/management/unit_test.go +++ b/x-pack/libbeat/management/unit_test.go @@ -7,9 +7,10 @@ package management import ( "testing" + "google.golang.org/protobuf/types/known/structpb" + "github.com/elastic/elastic-agent-client/v7/pkg/client" "github.com/elastic/elastic-agent-client/v7/pkg/proto" - "google.golang.org/protobuf/types/known/structpb" "github.com/elastic/beats/v7/libbeat/management/status" ) @@ -442,8 +443,14 @@ func TestSuppressFlipRecomputesHealth(t *testing.T) { } // Per-stream payload should still show Degraded - streams := cu.reportedPayload["streams"].(map[string]interface{}) - streamStatus := streams["stream-a"].(map[string]interface{}) + streams, ok := cu.reportedPayload["streams"].(map[string]interface{}) + if !ok { + t.Fatal("expected streams payload") + } + streamStatus, ok := streams["stream-a"].(map[string]interface{}) + if !ok { + t.Fatal("expected stream-a payload") + } if streamStatus["status"] != client.UnitStateDegraded.String() { t.Errorf("expected per-stream Degraded after flip, got %q", streamStatus["status"]) } From 95b667fc9a4c684419b6ab511583d676a617d442 Mon Sep 17 00:00:00 2001 From: Sam Crauwels Date: Fri, 27 Mar 2026 22:39:44 +0100 Subject: [PATCH 3/3] libbeat: management: add status_reporting stream config for health aggregation 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. --- ...774128388-suppress-health-degradation.yaml | 2 +- x-pack/libbeat/management/unit.go | 102 ++++-- x-pack/libbeat/management/unit_test.go | 314 +++++++++++++----- 3 files changed, 302 insertions(+), 116 deletions(-) diff --git a/changelog/fragments/1774128388-suppress-health-degradation.yaml b/changelog/fragments/1774128388-suppress-health-degradation.yaml index 5500f76e0387..3d5184bb9a9b 100644 --- a/changelog/fragments/1774128388-suppress-health-degradation.yaml +++ b/changelog/fragments/1774128388-suppress-health-degradation.yaml @@ -1,6 +1,6 @@ kind: feature -summary: Add suppress_health_degradation stream config to prevent optional streams from degrading agent health +summary: Add status_reporting stream config to control which health states propagate to unit aggregate health component: libbeat diff --git a/x-pack/libbeat/management/unit.go b/x-pack/libbeat/management/unit.go index 59d02512ab6c..0e9d62effdb5 100644 --- a/x-pack/libbeat/management/unit.go +++ b/x-pack/libbeat/management/unit.go @@ -8,16 +8,26 @@ import ( "fmt" "sync" + "google.golang.org/protobuf/types/known/structpb" + "github.com/elastic/beats/v7/libbeat/management/status" "github.com/elastic/elastic-agent-client/v7/pkg/client" "github.com/elastic/elastic-agent-libs/logp" ) +// streamStatusReporting controls which health states a stream contributes +// to the unit's aggregate health. Both fields default to true so that +// existing behaviour is preserved when the config block is absent. +type streamStatusReporting struct { + reportDegraded bool + reportFailed bool +} + // unitState is the current state of a unit type unitState struct { - state status.Status - msg string - suppressHealthDegradation bool // when true, this stream's degraded/failed states do not affect the unit's aggregate health + state status.Status + msg string + statusReporting streamStatusReporting } type clientUnit interface { @@ -69,7 +79,7 @@ func getUnitState(s status.Status) client.UnitState { } } -// getUnitState converts status.Status to client.UnitState +// getStatus converts client.UnitState to status.Status func getStatus(s client.UnitState) status.Status { switch s { case client.UnitStateStarting: @@ -91,6 +101,30 @@ func getStatus(s client.UnitState) status.Status { } } +// parseStatusReporting extracts status_reporting from a protobuf Struct source, +// using fallback as the default when fields are absent. +func parseStatusReporting(src *structpb.Struct, fallback streamStatusReporting) streamStatusReporting { + if src == nil { + return fallback + } + srVal, ok := src.GetFields()["status_reporting"] + if !ok { + return fallback + } + srMap := srVal.GetStructValue() + if srMap == nil { + return fallback + } + sr := fallback + if v, ok := srMap.GetFields()["report_degraded"]; ok { + sr.reportDegraded = v.GetBoolValue() + } + if v, ok := srMap.GetFields()["report_failed"]; ok { + sr.reportFailed = v.GetBoolValue() + } + return sr +} + func getStreamStates(expected client.Expected) (map[string]unitState, []string) { expectedCfg := expected.Config @@ -98,24 +132,23 @@ func getStreamStates(expected client.Expected) (map[string]unitState, []string) return nil, nil } + // Read input-level status_reporting as a default for all streams. + inputDefault := parseStatusReporting( + expectedCfg.GetSource(), + streamStatusReporting{reportDegraded: true, reportFailed: true}, + ) + streamStates := make(map[string]unitState, len(expectedCfg.Streams)) streamIDs := make([]string, len(expectedCfg.Streams)) for idx, stream := range expectedCfg.Streams { - // Check if stream is marked as suppressHealthDegradation in its source config. - // Optional streams still collect data and report per-stream status, - // 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 { - suppressHealth = v.GetBoolValue() - } - } + // Stream-level status_reporting overrides the input-level default. + sr := parseStatusReporting(stream.GetSource(), inputDefault) streamState := unitState{ - state: status.Unknown, - msg: "", - suppressHealthDegradation: suppressHealth, + state: status.Unknown, + msg: "", + statusReporting: sr, } if id := stream.GetId(); id != "" { @@ -229,22 +262,24 @@ func (u *agentUnit) calcState() (status.Status, string) { } // inputLevelState state is marked as running, check the stream states. - // Streams marked as suppressHealthDegradation are excluded from the aggregate health - // calculation — they still report per-stream status but do not cause - // the unit to be reported as degraded or failed. + // Streams with status_reporting.report_degraded or report_failed set to + // false are excluded from the aggregate health for those specific states. reportedStatus := status.Running reportedMsg := "Healthy" for _, streamState := range u.streamStates { - if streamState.suppressHealthDegradation { - continue - } switch streamState.state { case status.Degraded: + if !streamState.statusReporting.reportDegraded { + continue + } if reportedStatus != status.Degraded { reportedStatus = status.Degraded reportedMsg = streamState.msg } case status.Failed: + if !streamState.statusReporting.reportFailed { + continue + } // return the first failed stream return streamState.state, streamState.msg } @@ -325,9 +360,9 @@ func (u *agentUnit) updateStateForStream(streamID string, state status.Status, m } u.streamStates[streamID] = unitState{ - state: state, - msg: msg, - suppressHealthDegradation: u.streamStates[streamID].suppressHealthDegradation, + state: state, + msg: msg, + statusReporting: u.streamStates[streamID].statusReporting, } state, msg = u.calcState() @@ -366,15 +401,15 @@ func (u *agentUnit) update(cu *client.Unit) { newStreamStates, newStreamIDs := getStreamStates(cu.Expected()) - suppressionChanged := false + reportingChanged := false for key, state := range newStreamStates { if existing, exists := u.streamStates[key]; exists { - // Preserve current health state but update the suppressHealthDegradation flag + // Preserve current health state but update the status_reporting flags // in case the stream config changed. - if existing.suppressHealthDegradation != state.suppressHealthDegradation { - suppressionChanged = true + if existing.statusReporting != state.statusReporting { + reportingChanged = true } - existing.suppressHealthDegradation = state.suppressHealthDegradation + existing.statusReporting = state.statusReporting u.streamStates[key] = existing continue } @@ -400,10 +435,9 @@ func (u *agentUnit) update(cu *client.Unit) { } } - // If any stream's suppression flag changed, recompute and publish the - // aggregate unit health so that a flip from suppress=false→true (or - // vice versa) takes effect immediately. - if suppressionChanged { + // If any stream's status_reporting config changed, recompute and publish + // the aggregate unit health so the change takes effect immediately. + if reportingChanged { state, msg := u.calcState() streamsPayload := make(map[string]interface{}, len(u.streamStates)) for id, streamState := range u.streamStates { diff --git a/x-pack/libbeat/management/unit_test.go b/x-pack/libbeat/management/unit_test.go index c1e89a7a6ea1..4002643ef4e6 100644 --- a/x-pack/libbeat/management/unit_test.go +++ b/x-pack/libbeat/management/unit_test.go @@ -173,17 +173,17 @@ func TestUnitUpdate(t *testing.T) { } if c.unit.reportedState != c.expectedUnitStatus { - t.Errorf("expected unit status %s, got %s", c.expectedUnitStatus, aUnit.inputLevelState.state) + t.Errorf("expected unit status %s, got %s", c.expectedUnitStatus, c.unit.reportedState) } if c.unit.reportedMsg != c.expectedUnitMsg { - t.Errorf("expected unit msg %s, got %s", c.expectedUnitStatus, aUnit.inputLevelState.state) + t.Errorf("expected unit msg %q, got %q", c.expectedUnitMsg, c.unit.reportedMsg) } }) } } -func TestUnitUpdateSuppressHealthDegradation(t *testing.T) { +func TestUnitUpdateStatusReporting(t *testing.T) { type StatusUpdate struct { status status.Status @@ -196,12 +196,34 @@ func TestUnitUpdateSuppressHealthDegradation(t *testing.T) { Degraded = "Degraded" ) - suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ - "suppress_health_degradation": true, + // status_reporting with both report_degraded and report_failed set to false + muteBothSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": false, + "report_failed": false, + }, }) - notSuppressedSource, _ := structpb.NewStruct(map[string]interface{}{ - "suppress_health_degradation": false, + // status_reporting with only report_degraded muted + muteDegradedSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": false, + }, + }) + + // status_reporting with only report_failed muted + muteFailedSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_failed": false, + }, + }) + + // status_reporting with both explicitly true (same as not set) + explicitTrueSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": true, + "report_failed": true, + }, }) newUnit := func(streams []*proto.Stream) *mockClientUnit { @@ -224,111 +246,159 @@ func TestUnitUpdateSuppressHealthDegradation(t *testing.T) { expectedUnitMsg string }{ { - name: "suppressed stream degraded does not affect unit health", + name: "muted stream degraded does not affect unit health", unit: newUnit([]*proto.Stream{ {Id: "stream-required"}, - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: muteBothSource}, }), inputLevelStatus: StatusUpdate{status.Running, Healthy}, streamStates: map[string]StatusUpdate{ - "stream-required": {status.Running, Healthy}, - "stream-suppressed": {status.Degraded, Degraded}, + "stream-required": {status.Running, Healthy}, + "stream-muted": {status.Degraded, Degraded}, }, expectedUnitStatus: client.UnitStateHealthy, expectedUnitMsg: Healthy, }, { - name: "suppressed stream failed does not affect unit health", + name: "muted stream failed does not affect unit health", unit: newUnit([]*proto.Stream{ {Id: "stream-required"}, - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: muteBothSource}, }), inputLevelStatus: StatusUpdate{status.Running, Healthy}, streamStates: map[string]StatusUpdate{ - "stream-required": {status.Running, Healthy}, - "stream-suppressed": {status.Failed, Failed}, + "stream-required": {status.Running, Healthy}, + "stream-muted": {status.Failed, Failed}, }, expectedUnitStatus: client.UnitStateHealthy, expectedUnitMsg: Healthy, }, { - name: "required stream degraded still affects unit health even with suppressed streams", + name: "required stream degraded still affects unit health even with muted streams", unit: newUnit([]*proto.Stream{ {Id: "stream-required"}, - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: muteBothSource}, }), inputLevelStatus: StatusUpdate{status.Running, Healthy}, streamStates: map[string]StatusUpdate{ - "stream-required": {status.Degraded, Degraded}, - "stream-suppressed": {status.Degraded, Degraded}, + "stream-required": {status.Degraded, Degraded}, + "stream-muted": {status.Degraded, Degraded}, }, expectedUnitStatus: client.UnitStateDegraded, expectedUnitMsg: Degraded, }, { - name: "required stream failed still affects unit health even with suppressed streams", + name: "required stream failed still affects unit health even with muted streams", unit: newUnit([]*proto.Stream{ {Id: "stream-required"}, - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: muteBothSource}, }), inputLevelStatus: StatusUpdate{status.Running, Healthy}, streamStates: map[string]StatusUpdate{ - "stream-required": {status.Failed, Failed}, - "stream-suppressed": {status.Running, Healthy}, + "stream-required": {status.Failed, Failed}, + "stream-muted": {status.Running, Healthy}, }, expectedUnitStatus: client.UnitStateFailed, expectedUnitMsg: Failed, }, { - name: "all suppressed streams degraded and failed keeps unit healthy", + name: "all muted streams degraded and failed keeps unit healthy", unit: newUnit([]*proto.Stream{ - {Id: "stream-suppressed-1", Source: suppressedSource}, - {Id: "stream-suppressed-2", Source: suppressedSource}, + {Id: "stream-muted-1", Source: muteBothSource}, + {Id: "stream-muted-2", Source: muteBothSource}, }), inputLevelStatus: StatusUpdate{status.Running, Healthy}, streamStates: map[string]StatusUpdate{ - "stream-suppressed-1": {status.Degraded, Degraded}, - "stream-suppressed-2": {status.Failed, Failed}, + "stream-muted-1": {status.Degraded, Degraded}, + "stream-muted-2": {status.Failed, Failed}, }, expectedUnitStatus: client.UnitStateHealthy, expectedUnitMsg: Healthy, }, { - name: "suppress false behaves same as not set", + name: "explicit true behaves same as not set", unit: newUnit([]*proto.Stream{ - {Id: "stream-explicit-false", Source: notSuppressedSource}, + {Id: "stream-explicit-true", Source: explicitTrueSource}, }), inputLevelStatus: StatusUpdate{status.Running, Healthy}, streamStates: map[string]StatusUpdate{ - "stream-explicit-false": {status.Degraded, Degraded}, + "stream-explicit-true": {status.Degraded, Degraded}, }, expectedUnitStatus: client.UnitStateDegraded, expectedUnitMsg: Degraded, }, { - name: "input level degraded is not affected by suppress flag", + name: "input level degraded is not affected by status_reporting", unit: newUnit([]*proto.Stream{ - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: muteBothSource}, }), inputLevelStatus: StatusUpdate{status.Degraded, Degraded}, streamStates: map[string]StatusUpdate{ - "stream-suppressed": {status.Running, Healthy}, + "stream-muted": {status.Running, Healthy}, }, expectedUnitStatus: client.UnitStateDegraded, expectedUnitMsg: Degraded, }, { - name: "input level failed is not affected by suppress flag", + name: "input level failed is not affected by status_reporting", unit: newUnit([]*proto.Stream{ - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: muteBothSource}, }), inputLevelStatus: StatusUpdate{status.Failed, Failed}, streamStates: map[string]StatusUpdate{ - "stream-suppressed": {status.Running, Healthy}, + "stream-muted": {status.Running, Healthy}, }, expectedUnitStatus: client.UnitStateFailed, expectedUnitMsg: Failed, }, + { + name: "mute degraded only still reports failed", + unit: newUnit([]*proto.Stream{ + {Id: "stream-a", Source: muteDegradedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-a": {status.Failed, Failed}, + }, + expectedUnitStatus: client.UnitStateFailed, + expectedUnitMsg: Failed, + }, + { + name: "mute degraded only suppresses degraded", + unit: newUnit([]*proto.Stream{ + {Id: "stream-a", Source: muteDegradedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-a": {status.Degraded, Degraded}, + }, + expectedUnitStatus: client.UnitStateHealthy, + expectedUnitMsg: Healthy, + }, + { + name: "mute failed only still reports degraded", + unit: newUnit([]*proto.Stream{ + {Id: "stream-a", Source: muteFailedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-a": {status.Degraded, Degraded}, + }, + expectedUnitStatus: client.UnitStateDegraded, + expectedUnitMsg: Degraded, + }, + { + name: "mute failed only suppresses failed", + unit: newUnit([]*proto.Stream{ + {Id: "stream-a", Source: muteFailedSource}, + }), + inputLevelStatus: StatusUpdate{status.Running, Healthy}, + streamStates: map[string]StatusUpdate{ + "stream-a": {status.Failed, Failed}, + }, + expectedUnitStatus: client.UnitStateHealthy, + expectedUnitMsg: Healthy, + }, } for _, c := range cases { @@ -354,12 +424,23 @@ func TestUnitUpdateSuppressHealthDegradation(t *testing.T) { } } -func TestGetStreamStatesParsesSuppress(t *testing.T) { - suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ - "suppress_health_degradation": true, +func TestGetStreamStatesParsesStatusReporting(t *testing.T) { + muteBothSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": false, + "report_failed": false, + }, }) - notSuppressedSource, _ := structpb.NewStruct(map[string]interface{}{ - "suppress_health_degradation": false, + muteDegradedSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": false, + }, + }) + explicitTrueSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": true, + "report_failed": true, + }, }) expected := client.Expected{ @@ -367,35 +448,101 @@ func TestGetStreamStatesParsesSuppress(t *testing.T) { Id: "input-1", Streams: []*proto.Stream{ {Id: "stream-plain"}, - {Id: "stream-suppressed", Source: suppressedSource}, - {Id: "stream-not-suppressed", Source: notSuppressedSource}, + {Id: "stream-mute-both", Source: muteBothSource}, + {Id: "stream-mute-degraded", Source: muteDegradedSource}, + {Id: "stream-explicit-true", Source: explicitTrueSource}, }, }, } states, ids := getStreamStates(expected) - if len(states) != 3 { - t.Fatalf("expected 3 stream states, got %d", len(states)) + if len(states) != 4 { + t.Fatalf("expected 4 stream states, got %d", len(states)) + } + if len(ids) != 4 { + t.Fatalf("expected 4 stream IDs, got %d", len(ids)) + } + + // No source — both default to true + if !states["stream-plain"].statusReporting.reportDegraded { + t.Error("stream without Source should default reportDegraded to true") + } + if !states["stream-plain"].statusReporting.reportFailed { + t.Error("stream without Source should default reportFailed to true") + } + + // Both muted + if states["stream-mute-both"].statusReporting.reportDegraded { + t.Error("stream with report_degraded: false should have reportDegraded false") + } + if states["stream-mute-both"].statusReporting.reportFailed { + t.Error("stream with report_failed: false should have reportFailed false") + } + + // Only degraded muted, failed defaults to true + if states["stream-mute-degraded"].statusReporting.reportDegraded { + t.Error("stream with report_degraded: false should have reportDegraded false") + } + if !states["stream-mute-degraded"].statusReporting.reportFailed { + t.Error("stream without report_failed should default reportFailed to true") + } + + // Explicit true — same as defaults + if !states["stream-explicit-true"].statusReporting.reportDegraded { + t.Error("stream with report_degraded: true should have reportDegraded true") + } + if !states["stream-explicit-true"].statusReporting.reportFailed { + t.Error("stream with report_failed: true should have reportFailed true") + } +} + +func TestGetStreamStatesInputLevelInheritance(t *testing.T) { + inputSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": false, + "report_failed": false, + }, + }) + streamOverrideSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": true, + }, + }) + + expected := client.Expected{ + Config: &proto.UnitExpectedConfig{ + Id: "input-1", + Source: inputSource, + Streams: []*proto.Stream{ + {Id: "stream-inherit"}, + {Id: "stream-override", Source: streamOverrideSource}, + }, + }, } - if len(ids) != 3 { - t.Fatalf("expected 3 stream IDs, got %d", len(ids)) + + states, _ := getStreamStates(expected) + + // stream-inherit has no Source, should inherit input-level defaults + if states["stream-inherit"].statusReporting.reportDegraded { + t.Error("stream without Source should inherit input-level reportDegraded=false") } - if states["stream-plain"].suppressHealthDegradation { - t.Error("stream without Source should not have suppress set") + if states["stream-inherit"].statusReporting.reportFailed { + t.Error("stream without Source should inherit input-level reportFailed=false") } - if !states["stream-suppressed"].suppressHealthDegradation { - t.Error("stream with suppress_health_degradation: true should have suppress set") + + // stream-override sets report_degraded=true, report_failed falls back to input-level false + if !states["stream-override"].statusReporting.reportDegraded { + t.Error("stream with report_degraded: true should override input-level value") } - if states["stream-not-suppressed"].suppressHealthDegradation { - t.Error("stream with suppress_health_degradation: false should not have suppress set") + if states["stream-override"].statusReporting.reportFailed { + t.Error("stream without report_failed should inherit input-level reportFailed=false") } } -func TestSuppressFlipRecomputesHealth(t *testing.T) { - // Simulates what update() does when a policy change flips the suppress - // flag on a stream that is already Degraded. This verifies the - // recompute-on-suppression-change path in update(). +func TestStatusReportingFlipRecomputesHealth(t *testing.T) { + // Simulates what update() does when a policy change flips the + // status_reporting flags on a stream that is already Degraded. cu := &mockClientUnitWithPayload{ mockClientUnit: mockClientUnit{ @@ -413,17 +560,16 @@ func TestSuppressFlipRecomputesHealth(t *testing.T) { aUnit := newAgentUnit(cu, nil) _ = aUnit.UpdateState(status.Running, "Healthy", nil) - // Stream degrades — unit should be Degraded (suppress not set) + // Stream degrades — unit should be Degraded (defaults: both reported) aUnit.updateStateForStream("stream-a", status.Degraded, "connection refused") if cu.reportedState != client.UnitStateDegraded { t.Fatalf("expected Degraded before flip, got %s", cu.reportedState) } - // Simulate a policy update that adds suppress_health_degradation: true. - // This is exactly what update() does: flip the flag, then recompute. + // Simulate a policy update that sets report_degraded: false. aUnit.mtx.Lock() existing := aUnit.streamStates["stream-a"] - existing.suppressHealthDegradation = true + existing.statusReporting.reportDegraded = false aUnit.streamStates["stream-a"] = existing state, msg := aUnit.calcState() @@ -437,9 +583,9 @@ func TestSuppressFlipRecomputesHealth(t *testing.T) { _ = aUnit.clientUnit.UpdateState(getUnitState(state), msg, map[string]interface{}{"streams": streamsPayload}) aUnit.mtx.Unlock() - // Unit should now be Healthy — the suppression flip took effect + // Unit should now be Healthy if cu.reportedState != client.UnitStateHealthy { - t.Errorf("expected Healthy after suppress flip, got %s", cu.reportedState) + t.Errorf("expected Healthy after muting degraded, got %s", cu.reportedState) } // Per-stream payload should still show Degraded @@ -455,10 +601,10 @@ func TestSuppressFlipRecomputesHealth(t *testing.T) { t.Errorf("expected per-stream Degraded after flip, got %q", streamStatus["status"]) } - // Now flip suppress back to false — unit should go Degraded again + // Flip report_degraded back to true — unit should go Degraded again aUnit.mtx.Lock() existing = aUnit.streamStates["stream-a"] - existing.suppressHealthDegradation = false + existing.statusReporting.reportDegraded = true aUnit.streamStates["stream-a"] = existing state, msg = aUnit.calcState() @@ -473,13 +619,16 @@ func TestSuppressFlipRecomputesHealth(t *testing.T) { aUnit.mtx.Unlock() if cu.reportedState != client.UnitStateDegraded { - t.Errorf("expected Degraded after unsuppress flip, got %s", cu.reportedState) + t.Errorf("expected Degraded after re-enabling reporting, got %s", cu.reportedState) } } -func TestSuppressedStreamRecovery(t *testing.T) { - suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ - "suppress_health_degradation": true, +func TestMutedStreamRecovery(t *testing.T) { + mutedSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": false, + "report_failed": false, + }, }) cu := &mockClientUnit{ @@ -487,7 +636,7 @@ func TestSuppressedStreamRecovery(t *testing.T) { Config: &proto.UnitExpectedConfig{ Id: "input-1", Streams: []*proto.Stream{ - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: mutedSource}, }, }, }, @@ -497,21 +646,24 @@ func TestSuppressedStreamRecovery(t *testing.T) { _ = aUnit.UpdateState(status.Running, "Healthy", nil) // Stream goes degraded — unit should stay healthy - aUnit.updateStateForStream("stream-suppressed", status.Degraded, "connection refused") + aUnit.updateStateForStream("stream-muted", status.Degraded, "connection refused") if cu.reportedState != client.UnitStateHealthy { t.Errorf("expected Healthy during degradation, got %s", cu.reportedState) } // Stream recovers — unit should still be healthy - aUnit.updateStateForStream("stream-suppressed", status.Running, "Healthy") + aUnit.updateStateForStream("stream-muted", status.Running, "Healthy") if cu.reportedState != client.UnitStateHealthy { t.Errorf("expected Healthy after recovery, got %s", cu.reportedState) } } -func TestSuppressedStreamStillReportsPerStreamStatus(t *testing.T) { - suppressedSource, _ := structpb.NewStruct(map[string]interface{}{ - "suppress_health_degradation": true, +func TestMutedStreamStillReportsPerStreamStatus(t *testing.T) { + mutedSource, _ := structpb.NewStruct(map[string]interface{}{ + "status_reporting": map[string]interface{}{ + "report_degraded": false, + "report_failed": false, + }, }) cu := &mockClientUnitWithPayload{ @@ -520,7 +672,7 @@ func TestSuppressedStreamStillReportsPerStreamStatus(t *testing.T) { Config: &proto.UnitExpectedConfig{ Id: "input-1", Streams: []*proto.Stream{ - {Id: "stream-suppressed", Source: suppressedSource}, + {Id: "stream-muted", Source: mutedSource}, }, }, }, @@ -529,9 +681,9 @@ func TestSuppressedStreamStillReportsPerStreamStatus(t *testing.T) { aUnit := newAgentUnit(cu, nil) _ = aUnit.UpdateState(status.Running, "Healthy", nil) - aUnit.updateStateForStream("stream-suppressed", status.Degraded, "connection refused") + aUnit.updateStateForStream("stream-muted", status.Degraded, "connection refused") - // Unit should be healthy (suppressed) + // Unit should be healthy (muted) if cu.reportedState != client.UnitStateHealthy { t.Errorf("expected unit Healthy, got %s", cu.reportedState) } @@ -541,9 +693,9 @@ func TestSuppressedStreamStillReportsPerStreamStatus(t *testing.T) { if !ok { t.Fatal("expected streams in payload") } - streamStatus, ok := streams["stream-suppressed"].(map[string]interface{}) + streamStatus, ok := streams["stream-muted"].(map[string]interface{}) if !ok { - t.Fatal("expected stream-suppressed in streams payload") + t.Fatal("expected stream-muted in streams payload") } if streamStatus["status"] != client.UnitStateDegraded.String() { t.Errorf("expected per-stream status %q, got %q", client.UnitStateDegraded.String(), streamStatus["status"])