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 {