feat(health): Add max time between metrics check

Author: tulku

plugins/outputs/health/README.md

@@ -46,6 +46,12 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   # tls_cert = "/etc/telegraf/cert.pem"
   # tls_key = "/etc/telegraf/key.pem"
 
+  ## Maximum expected time between metrics being written
+  ## Enforces an unhealthy state if there was no new metric seen for at least
+  ## the specified time. The check is disabled by default and only used if a
+  ## positive time is specified.
+  # max_time_between_metrics = "0s"
+
   ## NOTE: Due to the way TOML is parsed, tables must be at the END of the
   ## plugin definition, otherwise additional config options are read as part of
   ## the table
@@ -67,6 +73,20 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ##   field = "buffer_size"
 ```
 
+### Maximum time between metrics
+
+The health plugin can assert that metrics are being delivered to it at an
+expected rate when setting `max_time_between_metrics` to a positive number.
+The check measures the time between consecutive writes to the plugin and
+compares it to the defined `max_time_between_metrics`. When the time
+elapsed between writes is greater than the configured maximum time, the plugin
+will report an unhealthy status. As soon as metrics are written again to the
+plugin, the health status will reset to healthy. Before the first metrics are
+received, the status will be healthy.
+
+Note that the metric timestamps are not taken into account, rather the time they
+are written to the plugin.
+
 ### compares
 
 The `compares` check is used to assert basic mathematical relationships.  Use

plugins/outputs/health/health.go

@@ -41,17 +41,20 @@ type Health struct {
 	BasicPassword  string          `toml:"basic_password"`
 	common_tls.ServerConfig
 
-	Compares []*Compares     `toml:"compares"`
-	Contains []*Contains     `toml:"contains"`
+	Compares              []*Compares     `toml:"compares"`
+	Contains              []*Contains     `toml:"contains"`
+	MaxTimeBetweenMetrics config.Duration `toml:"max_time_between_metrics"`
+
 	Log      telegraf.Logger `toml:"-"`
 	checkers []Checker
 
-	wg      sync.WaitGroup
-	server  *http.Server
-	origin  string
-	network string
-	address string
-	tlsConf *tls.Config
+	wg             sync.WaitGroup
+	server         *http.Server
+	origin         string
+	network        string
+	address        string
+	tlsConf        *tls.Config
+	lastMetricTime time.Time
 
 	mu      sync.Mutex
 	healthy bool
@@ -143,7 +146,13 @@ func (h *Health) listen() (net.Listener, error) {
 
 func (h *Health) ServeHTTP(rw http.ResponseWriter, _ *http.Request) {
 	var code = http.StatusOK
-	if !h.isHealthy() {
+
+	healthy := h.isHealthy()
+	if h.MaxTimeBetweenMetrics > 0 && !h.lastMetricTime.IsZero() {
+		healthy = healthy && time.Since(h.lastMetricTime) < time.Duration(h.MaxTimeBetweenMetrics)
+	}
+
+	if !healthy {
 		code = http.StatusServiceUnavailable
 	}
 
@@ -153,6 +162,7 @@ func (h *Health) ServeHTTP(rw http.ResponseWriter, _ *http.Request) {
 
 // Write runs all checks over the metric batch and adjust health state.
 func (h *Health) Write(metrics []telegraf.Metric) error {
+	h.lastMetricTime = time.Now()
 	healthy := true
 	for _, checker := range h.checkers {
 		success := checker.Check(metrics)
@@ -219,10 +229,12 @@ func (h *Health) isHealthy() bool {
 
 func NewHealth() *Health {
 	return &Health{
-		ServiceAddress: defaultServiceAddress,
-		ReadTimeout:    config.Duration(defaultReadTimeout),
-		WriteTimeout:   config.Duration(defaultWriteTimeout),
-		healthy:        true,
+		ServiceAddress:        defaultServiceAddress,
+		ReadTimeout:           config.Duration(defaultReadTimeout),
+		WriteTimeout:          config.Duration(defaultWriteTimeout),
+		MaxTimeBetweenMetrics: 0,
+		healthy:               true,
+		lastMetricTime:        time.Time{},
 	}
 }
 

plugins/outputs/health/health_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/plugins/outputs/health"
 	"github.com/influxdata/telegraf/testutil"
 )
@@ -214,3 +215,86 @@ func TestInitServiceAddress(t *testing.T) {
 		})
 	}
 }
+
+func TestTimeBetweenMetrics(t *testing.T) {
+	now := time.Now()
+	past := time.Time{}.AddDate(2002, 0, 0)
+	tests := []struct {
+		name                  string
+		maxTimeBetweenMetrics config.Duration
+		metrics               []telegraf.Metric
+		expectedCode          int
+	}{
+		{
+			name:                  "healthy when disabled and old metric",
+			maxTimeBetweenMetrics: config.Duration(0),
+			metrics: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]any{
+						"time_idle": 42,
+					},
+					past),
+			},
+			expectedCode: 200,
+		},
+		{
+			name:                  "healthy when enabled and recent metric",
+			maxTimeBetweenMetrics: config.Duration(5 * time.Second),
+			metrics: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]any{
+						"time_idle": 42,
+					},
+					now),
+			},
+			expectedCode: 200,
+		},
+		{
+			name:                  "unhealthy when enabled and old metric",
+			maxTimeBetweenMetrics: config.Duration(1 * time.Nanosecond),
+			metrics: []telegraf.Metric{
+				testutil.MustMetric(
+					"cpu",
+					map[string]string{},
+					map[string]any{
+						"time_idle": 42,
+					},
+					past),
+			},
+			expectedCode: 503,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			dut := health.NewHealth()
+			dut.ServiceAddress = "tcp://127.0.0.1:0"
+			dut.Log = testutil.Logger{}
+			dut.MaxTimeBetweenMetrics = tt.maxTimeBetweenMetrics
+
+			err := dut.Init()
+			require.NoError(t, err)
+
+			err = dut.Connect()
+			require.NoError(t, err)
+
+			err = dut.Write(tt.metrics)
+			require.NoError(t, err)
+
+			resp, err := http.Get(dut.Origin())
+			require.NoError(t, err)
+			defer resp.Body.Close()
+			require.Equal(t, tt.expectedCode, resp.StatusCode)
+
+			_, err = io.ReadAll(resp.Body)
+			require.NoError(t, err)
+
+			err = dut.Close()
+			require.NoError(t, err)
+		})
+	}
+}

plugins/outputs/health/sample.conf

@@ -21,6 +21,12 @@
   # tls_cert = "/etc/telegraf/cert.pem"
   # tls_key = "/etc/telegraf/key.pem"
 
+  ## Maximum expected time between metrics being written
+  ## Enforces an unhealthy state if there was no new metric seen for at least
+  ## the specified time. The check is disabled by default and only used if a
+  ## positive time is specified.
+  # max_time_between_metrics = "0s"
+
   ## NOTE: Due to the way TOML is parsed, tables must be at the END of the
   ## plugin definition, otherwise additional config options are read as part of
   ## the table