Harden redis metrics collection against transient errors

internal/common/config/validation.go

@@ -14,18 +14,25 @@ type Config interface {
 }
 
 func FormatValidationErrors(err error) error {
-	var validationErrors error
-	for _, err := range err.(validator.ValidationErrors) {
+	if err == nil {
+		return nil
+	}
+	var ve validator.ValidationErrors
+	if !errors.As(err, &ve) {
+		return err
+	}
+	var formatted error
+	for _, err := range ve {
 		fieldName := stripPrefix(err.Namespace())
 		tag := err.Tag()
 		switch tag {
 		case "required":
-			validationErrors = errors.Join(validationErrors, fmt.Errorf("ConfigError: Field %s is required but was not found", fieldName))
+			formatted = errors.Join(formatted, fmt.Errorf("ConfigError: Field %s is required but was not found", fieldName))
 		default:
-			validationErrors = errors.Join(validationErrors, fmt.Errorf("ConfigError: Field %s has invalid value %s: %s", fieldName, err.Value(), tag))
+			formatted = errors.Join(formatted, fmt.Errorf("ConfigError: Field %s has invalid value %s: %s", fieldName, err.Value(), tag))
 		}
 	}
-	return validationErrors
+	return formatted
 }
 
 func stripPrefix(s string) string {

internal/common/config/validation_test.go

@@ -0,0 +1,25 @@
+package config
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/go-playground/validator/v10"
+	"github.com/stretchr/testify/assert"
+)
+
+type formatTestStruct struct {
+	Field string `validate:"required"`
+}
+
+func TestFormatValidationErrors(t *testing.T) {
+	newValidationErrors := func() error { return validator.New().Struct(formatTestStruct{}) }
+	assert.EqualError(t, FormatValidationErrors(newValidationErrors()),
+		"ConfigError: Field Field is required but was not found")
+	assert.EqualError(t, FormatValidationErrors(fmt.Errorf("invalid config: %w", newValidationErrors())),
+		"ConfigError: Field Field is required but was not found")
+	assert.EqualError(t, FormatValidationErrors(errors.New("some other error")), "some other error")
+	assert.EqualError(t, FormatValidationErrors(errors.Join(errors.New("a"), errors.New("b"))), "a\nb")
+	assert.Nil(t, FormatValidationErrors(nil))
+}

internal/eventingester/configuration/types.go

@@ -60,6 +60,19 @@ type RedisMemoryMetricsConfig struct {
 	InterBatchDelay           time.Duration
 	MemoryUsageSamples        int
 	Leader                    leaderelection.Config
+	// CollectionTimeout is the maximum duration of a single scan attempt.
+	// If zero, a default of 5 minutes is used. Set to a negative value to disable.
+	CollectionTimeout time.Duration
+	// RetryInitialBackoff is the initial backoff between scan attempts after a retryable error.
+	// If zero, a default of 500ms is used. It is capped at RetryMaxBackoff.
+	// Validation rejects configs where the initial backoff exceeds RetryMaxBackoff.
+	RetryInitialBackoff time.Duration
+	// RetryMaxBackoff is the maximum backoff between scan attempts.
+	// If zero, a default of 30s is used.
+	RetryMaxBackoff time.Duration
+	// MaxRetries is the maximum number of retries per collection cycle after the first attempt.
+	// If zero, a default of 10 is used. Set to a negative value to disable retries.
+	MaxRetries int
 }
 
 // TODO: unpack this into just EventExpirtation

internal/eventingester/configuration/validation.go

@@ -1,16 +1,48 @@
 package configuration
 
 import (
+	"fmt"
+	"time"
+
 	"github.com/go-playground/validator/v10"
 
 	commonconfig "github.com/armadaproject/armada/internal/common/config"
 )
 
+// DefaultRetryInitialBackoff is used when RetryInitialBackoff is unset.
+// It lives here so validation can reason about effective values without
+// importing the collector package (which depends on this one).
+const DefaultRetryInitialBackoff = 500 * time.Millisecond
+
 func (c EventIngesterConfiguration) Validate() error {
 	validate := validator.New()
+	validate.RegisterStructValidation(redisMemoryMetricsConfigValidation, RedisMemoryMetricsConfig{})
 	return validate.Struct(c)
 }
 
+func redisMemoryMetricsConfigValidation(sl validator.StructLevel) {
+	c := sl.Current().Interface().(RedisMemoryMetricsConfig)
+
+	if c.RetryInitialBackoff < 0 {
+		sl.ReportError(c.RetryInitialBackoff, "RetryInitialBackoff", "", "retryInitialBackoff must be non-negative", "")
+	}
+	if c.RetryMaxBackoff < 0 {
+		sl.ReportError(c.RetryMaxBackoff, "RetryMaxBackoff", "", "retryMaxBackoff must be non-negative", "")
+	}
+
+	effectiveInitialBackoff := c.RetryInitialBackoff
+	if effectiveInitialBackoff == 0 {
+		effectiveInitialBackoff = DefaultRetryInitialBackoff
+	}
+	if c.RetryMaxBackoff > 0 && effectiveInitialBackoff > c.RetryMaxBackoff {
+		if c.RetryInitialBackoff == 0 {
+			sl.ReportError(c.RetryMaxBackoff, "RetryMaxBackoff", "", fmt.Sprintf("retryMaxBackoff (%s) is below the default retryInitialBackoff (%s); set retryInitialBackoff explicitly or raise retryMaxBackoff", c.RetryMaxBackoff, DefaultRetryInitialBackoff), "")
+		} else {
+			sl.ReportError(c.RetryInitialBackoff, "RetryInitialBackoff", "", fmt.Sprintf("retryInitialBackoff (%s) must not exceed retryMaxBackoff (%s)", c.RetryInitialBackoff, c.RetryMaxBackoff), "")
+		}
+	}
+}
+
 func (c *EventIngesterConfiguration) Mutate() (commonconfig.Config, error) {
 	c.Observability.ApplyResourceDefaults("eventingester")
 	return c, nil

internal/eventingester/configuration/validation_test.go

@@ -0,0 +1,104 @@
+package configuration
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/armadaproject/armada/internal/leaderelection"
+)
+
+func validRedisMemoryMetricsConfig() RedisMemoryMetricsConfig {
+	return RedisMemoryMetricsConfig{
+		Enabled:             true,
+		CollectionInterval:  time.Minute,
+		TopN:                10,
+		RetryInitialBackoff: 500 * time.Millisecond,
+		RetryMaxBackoff:     30 * time.Second,
+		Leader:              leaderelection.Config{Mode: leaderelection.ModeStandalone},
+	}
+}
+
+func validEventIngesterConfiguration() EventIngesterConfiguration {
+	return EventIngesterConfiguration{
+		Metrics: MetricsConfig{
+			Redis: validRedisMemoryMetricsConfig(),
+		},
+	}
+}
+
+func TestValidate_AcceptsValidRetryConfig(t *testing.T) {
+	require.NoError(t, validEventIngesterConfiguration().Validate())
+}
+
+func TestValidate_AllowsZeroBackoffs(t *testing.T) {
+	config := validEventIngesterConfiguration()
+	config.Metrics.Redis = RedisMemoryMetricsConfig{
+		Leader: leaderelection.Config{Mode: leaderelection.ModeStandalone},
+	}
+	require.NoError(t, config.Validate())
+}
+
+func TestValidate_RejectsNegativeInitialBackoff(t *testing.T) {
+	redisConfig := validRedisMemoryMetricsConfig()
+	redisConfig.RetryInitialBackoff = -1 * time.Second
+	config := EventIngesterConfiguration{Metrics: MetricsConfig{Redis: redisConfig}}
+
+	err := config.Validate()
+	require.Error(t, err)
+	require.ErrorContains(t, err, "retryInitialBackoff must be non-negative")
+}
+
+func TestValidate_RejectsNegativeMaxBackoff(t *testing.T) {
+	redisConfig := validRedisMemoryMetricsConfig()
+	redisConfig.RetryMaxBackoff = -1 * time.Second
+	config := EventIngesterConfiguration{Metrics: MetricsConfig{Redis: redisConfig}}
+
+	err := config.Validate()
+	require.Error(t, err)
+	require.ErrorContains(t, err, "retryMaxBackoff must be non-negative")
+}
+
+func TestValidate_RejectsInitialBackoffAboveMax(t *testing.T) {
+	redisConfig := validRedisMemoryMetricsConfig()
+	redisConfig.RetryInitialBackoff = 40 * time.Second
+	redisConfig.RetryMaxBackoff = 30 * time.Second
+	config := EventIngesterConfiguration{Metrics: MetricsConfig{Redis: redisConfig}}
+
+	err := config.Validate()
+	require.Error(t, err)
+	require.ErrorContains(t, err, "retryInitialBackoff (40s) must not exceed retryMaxBackoff (30s)")
+}
+
+func TestValidate_RejectsUnsetInitialBackoffWithMaxBelowDefault(t *testing.T) {
+	redisConfig := validRedisMemoryMetricsConfig()
+	redisConfig.RetryInitialBackoff = 0
+	redisConfig.RetryMaxBackoff = 100 * time.Millisecond
+	config := EventIngesterConfiguration{Metrics: MetricsConfig{Redis: redisConfig}}
+
+	err := config.Validate()
+	require.Error(t, err)
+	require.ErrorContains(t, err, "below the default retryInitialBackoff")
+}
+
+func TestValidate_AcceptsUnsetInitialBackoffWithMaxAtDefault(t *testing.T) {
+	redisConfig := validRedisMemoryMetricsConfig()
+	redisConfig.RetryInitialBackoff = 0
+	redisConfig.RetryMaxBackoff = DefaultRetryInitialBackoff
+	config := EventIngesterConfiguration{Metrics: MetricsConfig{Redis: redisConfig}}
+
+	require.NoError(t, config.Validate())
+}
+
+func TestValidate_ReportsAllViolations(t *testing.T) {
+	redisConfig := validRedisMemoryMetricsConfig()
+	redisConfig.RetryInitialBackoff = -1 * time.Second
+	redisConfig.RetryMaxBackoff = -2 * time.Second
+	config := EventIngesterConfiguration{Metrics: MetricsConfig{Redis: redisConfig}}
+
+	err := config.Validate()
+	require.Error(t, err)
+	require.ErrorContains(t, err, "retryInitialBackoff must be non-negative")
+	require.ErrorContains(t, err, "retryMaxBackoff must be non-negative")
+}