From 353f438af9c872fd5e05c0143b1f86ee72dffb8b Mon Sep 17 00:00:00 2001 From: Sam Crauwels Date: Mon, 16 Mar 2026 12:51:28 +0100 Subject: [PATCH] Add optional flag to metricbeat modules to suppress health degradation on fetch errors Adds an `optional` config flag to metricbeat modules. 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 service integrations that may not be running on every enrolled host, without those absent services dragging the entire agent into a degraded state. The flag follows the same pattern as the existing `failure_threshold` setting: parsed from module config in `createWrapper`, stored on `metricSetWrapper`, and checked in `handleFetchError`. All periodic metricbeat modules get this for free. Fixes: https://github.com/elastic/elastic-agent/issues/12885 --- metricbeat/mb/module/wrapper.go | 10 ++++- metricbeat/mb/module/wrapper_internal_test.go | 44 +++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/metricbeat/mb/module/wrapper.go b/metricbeat/mb/module/wrapper.go index 69cb881935c7..771199ef43be 100644 --- a/metricbeat/mb/module/wrapper.go +++ b/metricbeat/mb/module/wrapper.go @@ -77,6 +77,7 @@ type metricSetWrapper struct { periodic bool // Set to true if this metricset is a periodic fetcher failureThreshold uint // threshold of consecutive errors needed to set the stream as degraded + optional bool // When true, fetch failures do not degrade agent health } // stats bundles common metricset stats. @@ -119,6 +120,7 @@ func createWrapper(module mb.Module, metricSets []mb.MetricSet, monitoring beatm var streamHealthSettings struct { FailureThreshold *uint `config:"failure_threshold"` + Optional *bool `config:"optional"` } err := module.UnpackConfig(&streamHealthSettings) @@ -131,12 +133,18 @@ func createWrapper(module mb.Module, metricSets []mb.MetricSet, monitoring beatm failureThreshold = *streamHealthSettings.FailureThreshold } + optional := false + if streamHealthSettings.Optional != nil { + optional = *streamHealthSettings.Optional + } + for i, metricSet := range metricSets { wrapper.metricSets[i] = &metricSetWrapper{ MetricSet: metricSet, module: wrapper, stats: getMetricSetStats(monitoring, wrapper.Name(), metricSet.Name()), failureThreshold: failureThreshold, + optional: optional, } } return wrapper, nil @@ -324,7 +332,7 @@ func (msw *metricSetWrapper) handleFetchError(err error, reporter mb.PushReporte default: reporter.Error(err) msw.stats.consecutiveFailures.Inc() - if msw.failureThreshold > 0 && msw.stats.consecutiveFailures != nil && uint(msw.stats.consecutiveFailures.Get()) >= msw.failureThreshold { + if !msw.optional && msw.failureThreshold > 0 && msw.stats.consecutiveFailures != nil && uint(msw.stats.consecutiveFailures.Get()) >= msw.failureThreshold { // mark it as degraded for any other issue encountered msw.module.UpdateStatus(status.Degraded, fmt.Sprintf("Error fetching data for metricset %s.%s: %v", msw.module.Name(), msw.Name(), err)) } diff --git a/metricbeat/mb/module/wrapper_internal_test.go b/metricbeat/mb/module/wrapper_internal_test.go index e4def57003dd..cb038bcd6b56 100644 --- a/metricbeat/mb/module/wrapper_internal_test.go +++ b/metricbeat/mb/module/wrapper_internal_test.go @@ -270,6 +270,50 @@ func TestWrapperHandleFetchErrorSync(t *testing.T) { } }, }, + { + name: "optional = true: status never becomes DEGRADED regardless of errors", + config: newConfig(t, map[string]interface{}{ + "module": mockModuleName, + "metricsets": []string{mockMetricSetName}, + "period": "100ms", + "hosts": []string{"testhost"}, + "optional": true, + }), + setup: func(t *testing.T, fetcher *mockReportingFetcher, pushReporter *mockPushReporterV2, statusReporter *mockStatusReporter) { + fetcher.On("Fetch", pushReporter).Return(fetchError).Times(5) + pushReporter.On("Error", fetchError).Return(true).Times(5) + }, + iterations: 5, + assertIteration: func(t *testing.T, i int, msWrapper *metricSetWrapper, fetcher *mockReportingFetcher, pushReporter *mockPushReporterV2, statusReporter *mockStatusReporter) { + t.Logf("Assertion after iteration %d", i) + assert.Truef(t, statusReporter.AssertNotCalled(t, "UpdateStatus", status.Degraded, mock.AnythingOfType("string")), "optional stream should not be degraded at iteration %d", i) + assert.Equal(t, uint64(i+1), msWrapper.stats.consecutiveFailures.Get(), "consecutive failures should be tracked even for optional metricsets") + }, + }, + { + name: "optional = true: status returns to Running after recovery", + config: newConfig(t, map[string]interface{}{ + "module": mockModuleName, + "metricsets": []string{mockMetricSetName}, + "period": "100ms", + "hosts": []string{"testhost"}, + "optional": true, + }), + setup: func(t *testing.T, fetcher *mockReportingFetcher, pushReporter *mockPushReporterV2, statusReporter *mockStatusReporter) { + fetcher.On("Fetch", pushReporter).Return(fetchError).Times(3) + fetcher.On("Fetch", pushReporter).Return(nil).Once() + pushReporter.On("Error", fetchError).Return(true).Times(3) + statusReporter.On("UpdateStatus", status.Running, mock.AnythingOfType("string")).Once() + }, + iterations: 4, + assertIteration: func(t *testing.T, i int, msWrapper *metricSetWrapper, fetcher *mockReportingFetcher, pushReporter *mockPushReporterV2, statusReporter *mockStatusReporter) { + t.Logf("Assertion after iteration %d", i) + assert.Truef(t, statusReporter.AssertNotCalled(t, "UpdateStatus", status.Degraded, mock.AnythingOfType("string")), "optional stream should never be degraded at iteration %d", i) + if i == 3 { + assert.Truef(t, statusReporter.AssertCalled(t, "UpdateStatus", status.Running, mock.AnythingOfType("string")), "stream should be running after recovery at iteration %d", i) + } + }, + }, } for _, tc := range testcases {