diff --git a/internal/common/config/validation.go b/internal/common/config/validation.go index 2892fe7560f..7ce0b6e947b 100644 --- a/internal/common/config/validation.go +++ b/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 { diff --git a/internal/common/config/validation_test.go b/internal/common/config/validation_test.go new file mode 100644 index 00000000000..98599eb7955 --- /dev/null +++ b/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)) +} diff --git a/internal/eventingester/configuration/types.go b/internal/eventingester/configuration/types.go index d0100fbb8c8..013b6f68ec3 100644 --- a/internal/eventingester/configuration/types.go +++ b/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 diff --git a/internal/eventingester/configuration/validation.go b/internal/eventingester/configuration/validation.go index 03940bb2251..808ded23586 100644 --- a/internal/eventingester/configuration/validation.go +++ b/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 diff --git a/internal/eventingester/configuration/validation_test.go b/internal/eventingester/configuration/validation_test.go new file mode 100644 index 00000000000..fe462ae230b --- /dev/null +++ b/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") +} diff --git a/internal/eventingester/metrics/redis/collector.go b/internal/eventingester/metrics/redis/collector.go index 911aa829cae..2625f7b4180 100644 --- a/internal/eventingester/metrics/redis/collector.go +++ b/internal/eventingester/metrics/redis/collector.go @@ -2,16 +2,22 @@ package redis import ( "context" + "errors" "fmt" + "io" "math/rand/v2" + "net" "sort" + "strings" "sync" "sync/atomic" "time" "github.com/prometheus/client_golang/prometheus" + "github.com/redis/go-redis/v9" "github.com/armadaproject/armada/internal/common/armadacontext" + log "github.com/armadaproject/armada/internal/common/logging" "github.com/armadaproject/armada/internal/eventingester/configuration" "github.com/armadaproject/armada/internal/eventingester/repository" "github.com/armadaproject/armada/internal/leaderelection" @@ -39,6 +45,16 @@ const ( RedisMetricsErrorsTotalMetricName = ArmadaRedisMetricsPrefix + "metrics_errors_total" RedisMetricsLastCollectionTimestampMetricName = ArmadaRedisMetricsPrefix + "metrics_last_collection_timestamp" RedisMetricsStreamScannedMetricName = ArmadaRedisMetricsPrefix + "metrics_streams_scanned_total" + + // Defaults applied when the corresponding config values are unset (zero) + defaultCollectionTimeout = 5 * time.Minute + defaultRetryInitialBackoff = configuration.DefaultRetryInitialBackoff + defaultRetryMaxBackoff = 30 * time.Second + defaultMaxRetries = 10 + + // Label values for the collection duration metric's "status" label + collectionStatusSuccess = "success" + collectionStatusError = "error" ) // ScannerInterface defines the interface for scanning Redis streams. @@ -71,7 +87,7 @@ type Collector struct { queueEventsGauge *prometheus.GaugeVec // Self-monitoring - collectionDuration prometheus.Histogram + collectionDuration *prometheus.HistogramVec errorsTotal prometheus.Counter lastCollectionTimestamp prometheus.Gauge streamsScannedGauge prometheus.Gauge @@ -159,10 +175,10 @@ func NewCollector(scanner ScannerInterface, config configuration.RedisMemoryMetr }, []string{"queue"}, ), - collectionDuration: prometheus.NewHistogram(prometheus.HistogramOpts{ + collectionDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: RedisMetricsCollectionDurationMetricName, Help: "Duration of Redis metrics collection cycles", - }), + }, []string{"status"}), errorsTotal: prometheus.NewCounter(prometheus.CounterOpts{ Name: RedisMetricsErrorsTotalMetricName, Help: "Total number of Redis metrics collection errors", @@ -288,24 +304,18 @@ func (c *Collector) collectOnce(ctx context.Context) error { } defer c.collectMu.Unlock() - // Reset all metrics for fresh collection cycle - c.resetMetricsForNewCycle() - start := time.Now() - // Scan all streams - streams, err := c.scanner.ScanAll(ctx) + streams, err := c.scanWithRetry(ctx) if err != nil { c.errorsTotal.Inc() - // Update self-monitoring even on error - c.collectionDuration.Observe(time.Since(start).Seconds()) - c.lastCollectionTimestamp.SetToCurrentTime() - c.streamsScannedGauge.Set(0) - // Collect snapshot with error metrics - c.collectSnapshot() + c.collectionDuration.WithLabelValues(collectionStatusError).Observe(time.Since(start).Seconds()) + c.collectSnapshot() // Update snapshot with self-monitoring metrics even on error return fmt.Errorf("scanner error: %w", err) } + c.resetMetricsForNewCycle() + // Sort for top-N computations byMemory := make([]repository.StreamInfo, len(streams)) copy(byMemory, streams) @@ -362,7 +372,7 @@ func (c *Collector) collectOnce(ctx context.Context) error { } // Update self-monitoring - c.collectionDuration.Observe(time.Since(start).Seconds()) + c.collectionDuration.WithLabelValues(collectionStatusSuccess).Observe(time.Since(start).Seconds()) c.lastCollectionTimestamp.SetToCurrentTime() c.streamsScannedGauge.Set(float64(len(streams))) @@ -372,6 +382,99 @@ func (c *Collector) collectOnce(ctx context.Context) error { return nil } +// scanWithRetry runs ScanAll with a per-attempt timeout, retrying retryable +// errors (e.g. timeouts, connection errors) with exponential backoff until +// MaxRetries is exhausted. Non-retryable errors and parent context +// cancellation are returned immediately. +func (c *Collector) scanWithRetry(ctx context.Context) ([]repository.StreamInfo, error) { + collectionTimeout := c.config.CollectionTimeout + if collectionTimeout == 0 { + collectionTimeout = defaultCollectionTimeout + } + initialBackoff := c.config.RetryInitialBackoff + if initialBackoff == 0 { + initialBackoff = defaultRetryInitialBackoff + } + maxBackoff := c.config.RetryMaxBackoff + if maxBackoff == 0 { + maxBackoff = defaultRetryMaxBackoff + } + // The default initial backoff may exceed a user-configured maximum; cap it + // so the first retry respects the configured bound. + initialBackoff = min(initialBackoff, maxBackoff) + maxRetries := c.config.MaxRetries + if maxRetries == 0 { + maxRetries = defaultMaxRetries + } + + attempts := maxRetries + 1 + if attempts < 1 { + attempts = 1 + } + + backoff := initialBackoff + var lastErr error + for attempt := range attempts { + if attempt > 0 { + log.WithError(lastErr).Warnf("retryable error scanning Redis streams, attempt %d/%d failed, retrying in %s", attempt, maxRetries, backoff) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(backoff): + } + backoff = min(2*backoff, maxBackoff) + } + + attemptCtx := ctx + var cancel context.CancelFunc + if collectionTimeout > 0 { + attemptCtx, cancel = context.WithTimeout(ctx, collectionTimeout) + } + streams, err := c.scanner.ScanAll(attemptCtx) + if cancel != nil { + cancel() + } + if err == nil { + return streams, nil + } + lastErr = err + + // Shutdown or fatal error: propagate immediately without retrying. + if ctx.Err() != nil { + return nil, ctx.Err() + } + if !isRetryableScanError(err) { + return nil, err + } + } + + return nil, fmt.Errorf("scan failed after %d attempts: %w", attempts, lastErr) +} + +// isRetryableScanError returns true for transient errors worth retrying, +// such as timeouts and connection failures (including EOF and connection +// resets on an established connection). +func isRetryableScanError(err error) bool { + if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, io.EOF) { + return true + } + var redisErr redis.Error + if errors.As(err, &redisErr) && strings.Contains(strings.ToLower(redisErr.Error()), "timeout") { + return true + } + var netErr net.Error + if errors.As(err, &netErr) { + return true + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "timeout") || + strings.Contains(msg, "connection refused") || + strings.Contains(msg, "connection pool timeout") || + strings.Contains(msg, "connection reset") || + strings.Contains(msg, "broken pipe") || + strings.Contains(msg, "eof") +} + // collectSnapshot collects all metrics into an atomic snapshot. func (c *Collector) collectSnapshot() { ch := make(chan prometheus.Metric, 10000) diff --git a/internal/eventingester/metrics/redis/collector_test.go b/internal/eventingester/metrics/redis/collector_test.go index ed46a5d33f8..ea802288835 100644 --- a/internal/eventingester/metrics/redis/collector_test.go +++ b/internal/eventingester/metrics/redis/collector_test.go @@ -2,11 +2,15 @@ package redis import ( "context" + "errors" "fmt" + "io" + "net" "sort" "strings" "sync" "sync/atomic" + "syscall" "testing" "time" @@ -233,19 +237,6 @@ func TestCollect_StaleLabelsCleared(t *testing.T) { }) } -func TestCollect_ScannerError(t *testing.T) { - ctx, cancel := armadacontext.WithCancel(armadacontext.Background()) - cancel() - withRedisClient(ctx, func(client redis.UniversalClient) { - collector := newRedisBackedCollector(client, testCollectorConfig(5), leaderelection.NewStandaloneLeaderController()) - - err := collector.collectOnce(ctx) - require.Error(t, err) - require.ErrorContains(t, err, "scanner error: context canceled") - require.Equal(t, 1.0, testutil.ToFloat64(collector.errorsTotal)) - }) -} - func TestCollect_SkipIfBusy(t *testing.T) { ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 30*time.Second) defer cancel() @@ -329,7 +320,32 @@ func TestCollect_ContextCancellation(t *testing.T) { err := collector.collectOnce(ctx) require.Error(t, err) require.ErrorContains(t, err, "scanner error: context canceled") - require.NotNil(t, collectMetrics(collector)) + + // The failed cycle publishes error telemetry through the snapshot. + // No successful cycle has run yet, so no business (stream/queue) + // metrics are served - only self-monitoring error metrics. + metrics := collectMetrics(collector) + require.NotEmpty(t, metrics) + + businessMetricNames := []string{ + RedisStreamMemoryBytesMetricName, + RedisStreamEventCountMetricName, + RedisStreamAgeSecondsMetricName, + RedisQueueStreamsMetricName, + RedisQueueMemoryBytesMetricName, + RedisQueueEventsMetricName, + } + foundErrorsTotal := false + for _, m := range metrics { + desc := m.Desc().String() + for _, name := range businessMetricNames { + require.NotContains(t, desc, fmt.Sprintf("%q", name)) + } + if strings.Contains(desc, fmt.Sprintf("%q", RedisMetricsErrorsTotalMetricName)) { + foundErrorsTotal = true + } + } + require.True(t, foundErrorsTotal, "expected error telemetry to be served after failed collection") }) } @@ -979,3 +995,275 @@ func seedRedisStream(t *testing.T, client redis.UniversalClient, ctx context.Con } return streamKey } + +type scriptedMockScanner struct { + script []scriptedScanResult + calls atomic.Int64 +} + +type scriptedScanResult struct { + streams []repository.StreamInfo + err error +} + +func (m *scriptedMockScanner) ScanAll(ctx context.Context) ([]repository.StreamInfo, error) { + call := int(m.calls.Add(1)) + idx := min(call-1, len(m.script)-1) + result := m.script[idx] + return result.streams, result.err +} + +func testStreams(count int) []repository.StreamInfo { + streams := make([]repository.StreamInfo, count) + for i := range count { + queue := fmt.Sprintf("queue-%d", i) + jobSetId := fmt.Sprintf("jobset-%d", i) + streams[i] = repository.StreamInfo{ + Key: fmt.Sprintf("%s%s:%s", constants.EventStreamPrefix, queue, jobSetId), + Queue: queue, + JobSetId: jobSetId, + Length: int64(100 * (i + 1)), + MemoryBytes: int64(1024 * (i + 1)), + AgeSeconds: float64(60 * (i + 1)), + } + } + return streams +} + +func retryConfig() configuration.RedisMemoryMetricsConfig { + config := testCollectorConfig(5) + config.RetryInitialBackoff = 1 * time.Millisecond + config.RetryMaxBackoff = 5 * time.Millisecond + return config +} + +func TestCollect_KeepsStaleMetricsOnError(t *testing.T) { + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 10*time.Second) + defer cancel() + + scanner := &scriptedMockScanner{ + script: []scriptedScanResult{ + {streams: testStreams(3)}, + {err: errors.New("xinfo stream error for key \"Events:gone:gone\": ERR no such key")}, + }, + } + + collector := NewCollector(scanner, retryConfig(), leaderelection.NewStandaloneLeaderController()) + + require.NoError(t, collector.collectOnce(ctx)) + firstSnapshot := gaugeMetricValues(t, collector) + require.Len(t, metricKeys(firstSnapshot), 7) // 3 top-N + 3 queue aggregates + last-collection timestamp + + require.Error(t, collector.collectOnce(ctx)) + require.Equal(t, 1.0, testutil.ToFloat64(collector.errorsTotal)) + + staleSnapshot := gaugeMetricValues(t, collector) + require.Equal(t, firstSnapshot, staleSnapshot) +} + +func TestCollect_RetriesOnTransientErrors(t *testing.T) { + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 10*time.Second) + defer cancel() + + tests := map[string]struct { + script []scriptedScanResult + maxRetries int + expectCalls int64 + expectErr bool + }{ + "timeout then success": { + script: []scriptedScanResult{ + {err: fmt.Errorf("scan error: %w", context.DeadlineExceeded)}, + {err: fmt.Errorf("scan error: %w", context.DeadlineExceeded)}, + {streams: testStreams(2)}, + }, + maxRetries: 3, + expectCalls: 3, + }, + "eof then success": { + script: []scriptedScanResult{ + {err: io.EOF}, + {streams: testStreams(2)}, + }, + maxRetries: 3, + expectCalls: 2, + }, + "connection reset then success": { + script: []scriptedScanResult{ + {err: &net.OpError{Op: "read", Net: "tcp", Err: syscall.ECONNRESET}}, + {streams: testStreams(2)}, + }, + maxRetries: 3, + expectCalls: 2, + }, + "gives up after max retries": { + script: []scriptedScanResult{ + {streams: testStreams(1)}, + {err: fmt.Errorf("scan error: %w", context.DeadlineExceeded)}, + }, + maxRetries: 2, + expectCalls: 4, + expectErr: true, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + scanner := &scriptedMockScanner{script: tc.script} + config := retryConfig() + config.MaxRetries = tc.maxRetries + collector := NewCollector(scanner, config, leaderelection.NewStandaloneLeaderController()) + + if tc.expectErr { + // Seed a successful cycle first so stale snapshot is preserved + require.NoError(t, collector.collectOnce(ctx)) + err := collector.collectOnce(ctx) + require.Error(t, err) + require.ErrorContains(t, err, "scan failed after") + require.Equal(t, 1.0, testutil.ToFloat64(collector.errorsTotal)) + } else { + require.NoError(t, collector.collectOnce(ctx)) + require.Equal(t, 0.0, testutil.ToFloat64(collector.errorsTotal)) + } + require.Equal(t, tc.expectCalls, scanner.calls.Load()) + }) + } +} + +func TestIsRetryableScanError(t *testing.T) { + connectionResetErr := &net.OpError{Op: "read", Net: "tcp", Err: syscall.ECONNRESET} + dnsTimeoutErr := &net.DNSError{Err: "timeout", Name: "redis", IsTimeout: true} + + tests := map[string]struct { + err error + expected bool + }{ + "deadline exceeded": { + err: context.DeadlineExceeded, + expected: true, + }, + "wrapped deadline exceeded": { + err: fmt.Errorf("scan error: %w", context.DeadlineExceeded), + expected: true, + }, + "bare EOF": { + err: io.EOF, + expected: true, + }, + "wrapped EOF": { + err: fmt.Errorf("scan error: %w", io.EOF), + expected: true, + }, + "connection reset": { + err: connectionResetErr, + expected: true, + }, + "wrapped connection reset": { + err: fmt.Errorf("scan error: %w", connectionResetErr), + expected: true, + }, + "dns timeout": { + err: dnsTimeoutErr, + expected: true, + }, + "timeout string": { + err: errors.New("i/o timeout"), + expected: true, + }, + "connection refused string": { + err: errors.New("dial tcp 127.0.0.1:6379: connect: connection refused"), + expected: true, + }, + "connection pool timeout string": { + err: errors.New("redis: connection pool timeout"), + expected: true, + }, + "eof string": { + err: fmt.Errorf("scan error: read tcp 127.0.0.1:12345->127.0.0.1:6379: EOF"), + expected: true, + }, + "use of closed network connection string": { + err: errors.New("use of closed network connection"), + expected: false, + }, + "non-retryable application error": { + err: errors.New("ERR no such key"), + expected: false, + }, + "nil error": { + err: nil, + expected: false, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + if tc.err == nil { + return + } + require.Equal(t, tc.expected, isRetryableScanError(tc.err)) + }) + } +} + +func TestCollect_RetriesOnEOF(t *testing.T) { + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 10*time.Second) + defer cancel() + + scanner := &scriptedMockScanner{ + script: []scriptedScanResult{ + {err: io.EOF}, + {streams: testStreams(2)}, + }, + } + + collector := NewCollector(scanner, retryConfig(), leaderelection.NewStandaloneLeaderController()) + + require.NoError(t, collector.collectOnce(ctx)) + require.Equal(t, int64(2), scanner.calls.Load()) + require.Equal(t, 0.0, testutil.ToFloat64(collector.errorsTotal)) +} + +func TestCollect_RetriesOnConnectionReset(t *testing.T) { + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 10*time.Second) + defer cancel() + + scanner := &scriptedMockScanner{ + script: []scriptedScanResult{ + {err: &net.OpError{Op: "read", Net: "tcp", Err: syscall.ECONNRESET}}, + {streams: testStreams(2)}, + }, + } + + collector := NewCollector(scanner, retryConfig(), leaderelection.NewStandaloneLeaderController()) + + require.NoError(t, collector.collectOnce(ctx)) + require.Equal(t, int64(2), scanner.calls.Load()) + require.Equal(t, 0.0, testutil.ToFloat64(collector.errorsTotal)) +} + +func TestCollect_DefaultInitialBackoffCappedAtConfiguredMax(t *testing.T) { + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 10*time.Second) + defer cancel() + + scanner := &scriptedMockScanner{ + script: []scriptedScanResult{ + {err: errors.New("i/o timeout")}, + {streams: testStreams(1)}, + }, + } + + // Initial backoff unset (500ms default) but max backoff set well below it; + // without the cap the first retry would wait 500ms. + config := retryConfig() + config.RetryInitialBackoff = 0 + config.RetryMaxBackoff = 20 * time.Millisecond + collector := NewCollector(scanner, config, leaderelection.NewStandaloneLeaderController()) + + start := time.Now() + require.NoError(t, collector.collectOnce(ctx)) + elapsed := time.Since(start) + + require.Equal(t, int64(2), scanner.calls.Load()) + require.Less(t, elapsed, 250*time.Millisecond, "first retry waited longer than the configured max backoff") +} diff --git a/internal/eventingester/repository/scanner.go b/internal/eventingester/repository/scanner.go index 4600eb8322e..7426c1ec8f5 100644 --- a/internal/eventingester/repository/scanner.go +++ b/internal/eventingester/repository/scanner.go @@ -131,7 +131,11 @@ func (s *Scanner) executePipelineBatch(ctx context.Context, keys []string) ([]St info, err := infoCmd.Result() if err != nil { - if err == redis.Nil || strings.Contains(err.Error(), "NOSTREAM") || strings.Contains(err.Error(), "WRONGTYPE") { + errMsg := strings.ToLower(err.Error()) + if err == redis.Nil || + strings.Contains(errMsg, "nostream") || + strings.Contains(errMsg, "wrongtype") || + strings.Contains(errMsg, "no such key") { continue } return nil, fmt.Errorf("xinfo stream error for key %q: %w", key, err) diff --git a/internal/eventingester/repository/scanner_test.go b/internal/eventingester/repository/scanner_test.go index c383971f056..aa63d6bc967 100644 --- a/internal/eventingester/repository/scanner_test.go +++ b/internal/eventingester/repository/scanner_test.go @@ -118,6 +118,31 @@ func TestScanAll_ContextCancelled(t *testing.T) { }) } +func TestScanAll_VanishedKeySkipped(t *testing.T) { + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 30*time.Second) + defer cancel() + withRedisClient(ctx, func(client redis.UniversalClient) { + vanishedKey := seedRedisStream(t, client, ctx, "queue-a", "jobset-a", 10) + survivingKey := seedRedisStream(t, client, ctx, "queue-b", "jobset-b", 10) + + // Delete one key after it would have been returned by SCAN + require.NoError(t, client.Del(ctx, vanishedKey).Err()) + + config := configuration.RedisMemoryMetricsConfig{ + ScanBatchSize: 10, + PipelineBatchSize: 5, + InterBatchDelay: 0, + } + + scanner := NewScanner(client, config) + results, err := scanner.executePipelineBatch(ctx, []string{vanishedKey, survivingKey}) + + require.NoError(t, err) + require.Len(t, results, 1) + assert.Equal(t, survivingKey, results[0].Key) + }) +} + func withRedisClient(ctx *armadacontext.Context, action func(client redis.UniversalClient)) { client := redis.NewClient(&redis.Options{Addr: "localhost:6379", DB: 8}) defer client.FlushDB(ctx)