Add suppress_health_degradation stream config to prevent optional streams 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 #49492

Fixes: elastic/elastic-agent#12885

Author: Oddly

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()
@@ -348,7 +367,11 @@ func (u *agentUnit) update(cu *client.Unit) {
 	newStreamStates, newStreamIDs := getStreamStates(cu.Expected())
 
 	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.
+			existing.suppressHealthDegradation = state.suppressHealthDegradation
+			u.streamStates[key] = existing
 			continue
 		}
 

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,177 @@ 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)
+			}
+		})
+	}
+}
+
 type mockClientUnit struct {
 	expected      client.Expected
 	reportedState client.UnitState