diff --git a/cmd/e2e/configs/basic.go b/cmd/e2e/configs/basic.go index 5356be12e..28ed2eaf6 100644 --- a/cmd/e2e/configs/basic.go +++ b/cmd/e2e/configs/basic.go @@ -87,7 +87,7 @@ func Basic(t *testing.T, opts BasicOpts) config.Config { c.LogBatchSize = 100 c.DeploymentID = opts.DeploymentID c.Alert.AutoDisableDestination = true - c.Alert.ConsecutiveFailureCount = 20 + c.Alert.ConsecutiveFailureCount = config.NewOptionalString("20") // Setup cleanup t.Cleanup(func() { diff --git a/cmd/e2e/regressions_test.go b/cmd/e2e/regressions_test.go index 10b2c37b3..155bb738c 100644 --- a/cmd/e2e/regressions_test.go +++ b/cmd/e2e/regressions_test.go @@ -91,7 +91,7 @@ func TestE2E_Regression_AutoDisableWithoutCallbackURL(t *testing.T) { LogStorage: configs.LogStorageTypePostgres, }) cfg.Alert.AutoDisableDestination = true - cfg.Alert.ConsecutiveFailureCount = 20 + cfg.Alert.ConsecutiveFailureCount = config.NewOptionalString("20") require.NoError(t, cfg.Validate(config.Flags{})) configs.ApplyMigrations(t, &cfg) diff --git a/docs/apis/openapi.yaml b/docs/apis/openapi.yaml index 8e0304694..a8285cc79 100644 --- a/docs/apis/openapi.yaml +++ b/docs/apis/openapi.yaml @@ -3094,10 +3094,25 @@ components: properties: ALERT_AUTO_DISABLE_DESTINATION: type: string + description: >- + If "true", automatically disables a destination once + ALERT_CONSECUTIVE_FAILURE_COUNT is reached. Has no effect when + consecutive-failure alerting is disabled. ALERT_CONSECUTIVE_FAILURE_COUNT: type: string + description: >- + Consecutive delivery failures before alerting on a destination (and + disabling it when ALERT_AUTO_DISABLE_DESTINATION is "true"). Omit for + the default of 100; set to an empty string to disable + consecutive-failure alerting entirely. ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS: type: string + description: >- + Suppression window in seconds for exhausted_retries alerts: the first + exhaustion per destination alerts and subsequent ones within the + window are suppressed ("0" = no suppression). Omit for the default of + 3600; set to an empty string to disable exhausted_retries alerting + entirely. DELIVERY_TIMEOUT_SECONDS: type: string DESTINATIONS_AWS_KINESIS_METADATA_IN_PAYLOAD: diff --git a/docs/content/self-hosting/configuration.mdoc b/docs/content/self-hosting/configuration.mdoc index 07c4a3228..d932a62fc 100644 --- a/docs/content/self-hosting/configuration.mdoc +++ b/docs/content/self-hosting/configuration.mdoc @@ -98,9 +98,9 @@ Choose one for event log persistence: | Variable | Default | Description | |----------|---------|-------------| -| `ALERT_CALLBACK_URL` | — | URL to POST to when an alert fires | -| `ALERT_CONSECUTIVE_FAILURE_COUNT` | `20` | Consecutive failures before alert triggers | -| `ALERT_AUTO_DISABLE_DESTINATION` | `true` | Auto-disable destination when failure count reaches 100% | +| `ALERT_CONSECUTIVE_FAILURE_COUNT` | `100` | Consecutive delivery failures before alerting on a destination (and disabling it when `ALERT_AUTO_DISABLE_DESTINATION` is `true`). Leave unset for the default of `100`; set to an empty string to disable consecutive-failure alerting entirely. | +| `ALERT_AUTO_DISABLE_DESTINATION` | `false` | Auto-disable a destination once `ALERT_CONSECUTIVE_FAILURE_COUNT` is reached. Has no effect when consecutive-failure alerting is disabled. | +| `ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS` | `3600` | Suppression window (seconds) for `exhausted_retries` alerts: the first exhaustion per destination alerts and subsequent ones within the window are suppressed (`0` = no suppression, alert on every exhaustion). Leave unset for the default of `3600`; set to an empty string to disable `exhausted_retries` alerting entirely. | ## Destinations @@ -159,7 +159,7 @@ portal: organization_name: "Acme Corp" accent_color: "#6122E7" -alerts: - callback_url: "https://yourapp.com/api/outpost-alerts" +alert: consecutive_failure_count: 50 + exhausted_retries_window_seconds: 3600 ``` diff --git a/examples/docker-compose/compose.yml b/examples/docker-compose/compose.yml index 4eeeda9a2..1618b9afc 100644 --- a/examples/docker-compose/compose.yml +++ b/examples/docker-compose/compose.yml @@ -1,7 +1,7 @@ name: outpost-example services: migrate: - image: hookdeck/outpost:v1.0.5 + image: hookdeck/outpost:v1.0.6 command: migrate apply --yes env_file: .env depends_on: @@ -10,7 +10,7 @@ services: restart: "no" api: - image: hookdeck/outpost:v1.0.5 + image: hookdeck/outpost:v1.0.6 env_file: .env depends_on: migrate: @@ -25,7 +25,7 @@ services: - 3333:3333 delivery: - image: hookdeck/outpost:v1.0.5 + image: hookdeck/outpost:v1.0.6 env_file: .env depends_on: migrate: @@ -38,7 +38,7 @@ services: SERVICE: delivery log: - image: hookdeck/outpost:v1.0.5 + image: hookdeck/outpost:v1.0.6 env_file: .env depends_on: migrate: diff --git a/internal/alert/monitor.go b/internal/alert/monitor.go index 00d79cfdb..4d9f45a22 100644 --- a/internal/alert/monitor.go +++ b/internal/alert/monitor.go @@ -90,6 +90,23 @@ func WithExhaustedRetriesIdempotence(idemp idempotence.Idempotence) AlertOption } } +// WithConsecutiveFailureEnabled toggles consecutive-failure alerting. When set +// to false the monitor never tracks or alerts on consecutive failures (and +// therefore never auto-disables). Defaults to true. +func WithConsecutiveFailureEnabled(enabled bool) AlertOption { + return func(m *alertMonitor) { + m.consecutiveFailureEnabled = enabled + } +} + +// WithExhaustedRetriesEnabled toggles exhausted_retries alerting. When set to +// false the monitor never emits exhausted_retries alerts. Defaults to true. +func WithExhaustedRetriesEnabled(enabled bool) AlertOption { + return func(m *alertMonitor) { + m.exhaustedRetriesEnabled = enabled + } +} + // DeliveryAttempt represents a single delivery attempt type DeliveryAttempt struct { Event *models.Event @@ -109,6 +126,9 @@ type alertMonitor struct { alertThresholds []int retryMaxLimit int exhaustedRetryIdemp idempotence.Idempotence + + consecutiveFailureEnabled bool + exhaustedRetriesEnabled bool } // NewAlertMonitor creates a new alert monitor. Emitter and retryMaxLimit are @@ -119,10 +139,12 @@ func NewAlertMonitor(logger *logging.Logger, redisClient redis.Cmdable, emitter panic("alert: NewAlertMonitor requires a non-nil emitter") } alertMonitor := &alertMonitor{ - logger: logger, - emitter: emitter, - retryMaxLimit: retryMaxLimit, - alertThresholds: []int{50, 70, 90, 100}, // default thresholds + logger: logger, + emitter: emitter, + retryMaxLimit: retryMaxLimit, + alertThresholds: []int{50, 70, 90, 100}, // default thresholds + consecutiveFailureEnabled: true, + exhaustedRetriesEnabled: true, } for _, opt := range opts { @@ -142,93 +164,32 @@ func NewAlertMonitor(logger *logging.Logger, redisClient redis.Cmdable, emitter func (m *alertMonitor) HandleAttempt(ctx context.Context, attempt DeliveryAttempt) error { if attempt.Attempt.Status == models.AttemptStatusSuccess { + // Nothing is tracked when consecutive-failure alerting is disabled, so + // there is no count to reset. + if !m.consecutiveFailureEnabled { + return nil + } return m.store.ResetConsecutiveFailureCount(ctx, attempt.Destination.TenantID, attempt.Destination.ID) } - // Consecutive failure tracking - res, err := m.store.IncrementConsecutiveFailureCount(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID) - if err != nil { - return fmt.Errorf("failed to get alert state: %w", err) - } - - // Replayed attempt (MQ redelivery, producer re-publish) that already - // completed a full evaluation — skip re-emitting alerts. Attempts that - // were counted but never marked evaluated (eval failed mid-way and the - // message was nacked) fall through and re-run as recovery. - if !res.NewlyCounted && res.AlreadyEvaluated { - m.logger.Ctx(ctx).Debug("skipping replayed attempt: already evaluated", - zap.String("attempt_id", attempt.Attempt.ID), - zap.String("tenant_id", attempt.Destination.TenantID), - zap.String("destination_id", attempt.Destination.ID), - ) - return nil - } - - count := res.Count - level, shouldAlert := m.evaluator.ShouldAlert(count) - if shouldAlert { - // At 100% threshold, disable the destination and emit disabled alert. - // Both operations are idempotent on replay: DisableDestination is a no-op - // if already disabled, and consumers deduplicate events by ID. - if level == 100 && m.disabler != nil { - if err := m.disabler.DisableDestination(ctx, attempt.Destination.TenantID, attempt.Destination.ID); err != nil { - return fmt.Errorf("failed to disable destination: %w", err) - } - - now := time.Now() - attempt.Destination.DisabledAt = &now - - m.logger.Ctx(ctx).Audit("destination disabled", - zap.String("attempt_id", attempt.Attempt.ID), - zap.String("event_id", attempt.Event.ID), - zap.String("tenant_id", attempt.Destination.TenantID), - zap.String("destination_id", attempt.Destination.ID), - zap.String("destination_type", attempt.Destination.Type), - ) - - disabledData := DestinationDisabledData{ - TenantID: attempt.Destination.TenantID, - Destination: attempt.Destination, - DisabledAt: now, - Reason: "consecutive_failure", - Event: attempt.Event, - Attempt: attempt.Attempt, - } - if err := m.emitter.Emit(ctx, opevents.TopicAlertDestinationDisabled, attempt.Destination.TenantID, disabledData); err != nil { - return fmt.Errorf("failed to emit destination disabled alert: %w", err) - } - } - - // Emit consecutive failure alert - cfData := ConsecutiveFailureData{ - TenantID: attempt.Destination.TenantID, - Event: attempt.Event, - Attempt: attempt.Attempt, - Destination: attempt.Destination, - ConsecutiveFailures: ConsecutiveFailures{ - Current: count, - Max: m.autoDisableFailureCount, - Threshold: level, - }, + if m.consecutiveFailureEnabled { + // A replayed attempt that already completed evaluation skips the rest of + // the pipeline (exhausted-retries check and the evaluated mark), matching + // the original single-pass behavior. + done, err := m.handleConsecutiveFailure(ctx, attempt) + if err != nil { + return err } - if err := m.emitter.Emit(ctx, opevents.TopicAlertConsecutiveFailure, attempt.Destination.TenantID, cfData); err != nil { - return fmt.Errorf("failed to emit consecutive failure alert: %w", err) + if done { + return nil } - - m.logger.Ctx(ctx).Audit("alert sent", - zap.String("topic", opevents.TopicAlertConsecutiveFailure), - zap.String("attempt_id", attempt.Attempt.ID), - zap.String("event_id", attempt.Event.ID), - zap.String("tenant_id", attempt.Destination.TenantID), - zap.String("destination_id", attempt.Destination.ID), - zap.String("destination_type", attempt.Destination.Type), - ) } // Exhausted retries check (independent of consecutive failure thresholds). // Attempt is 1-indexed: with retryMaxLimit=10, attempt 11 is the final one. - // Skip if retryMaxLimit=0 (retries disabled — no exhausted state to report). - if m.retryMaxLimit > 0 && attempt.Event.EligibleForRetry && attempt.Attempt.AttemptNumber > m.retryMaxLimit { + // Skip if retryMaxLimit=0 (retries disabled — no exhausted state to report) + // or if exhausted-retries alerting is disabled. + if m.exhaustedRetriesEnabled && m.retryMaxLimit > 0 && attempt.Event.EligibleForRetry && attempt.Attempt.AttemptNumber > m.retryMaxLimit { erData := ExhaustedRetriesData{ TenantID: attempt.Destination.TenantID, Event: attempt.Event, @@ -263,17 +224,110 @@ func (m *alertMonitor) HandleAttempt(ctx context.Context, attempt DeliveryAttemp } } - // Mark the attempt fully evaluated so replays skip re-emitting alerts. + // Mark the attempt fully evaluated so replays skip re-emitting alerts. Only + // relevant when consecutive-failure tracking ran — it is the consumer of the + // evaluated mark (the replay short-circuit above). // Non-fatal: on failure the attempt simply re-evaluates on replay, which // matches the previous behavior (emit/disable are idempotent-by-design). - if err := m.store.MarkAttemptEvaluated(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID); err != nil { - m.logger.Ctx(ctx).Warn("failed to mark attempt evaluated", - zap.Error(err), + if m.consecutiveFailureEnabled { + if err := m.store.MarkAttemptEvaluated(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID); err != nil { + m.logger.Ctx(ctx).Warn("failed to mark attempt evaluated", + zap.Error(err), + zap.String("attempt_id", attempt.Attempt.ID), + zap.String("tenant_id", attempt.Destination.TenantID), + zap.String("destination_id", attempt.Destination.ID), + ) + } + } + + return nil +} + +// handleConsecutiveFailure runs consecutive-failure tracking, alerting and +// auto-disable for a failed attempt. It returns done=true when the attempt is a +// replay that already completed evaluation, signalling the caller to stop +// processing (skip the exhausted-retries check and the evaluated mark). +func (m *alertMonitor) handleConsecutiveFailure(ctx context.Context, attempt DeliveryAttempt) (done bool, err error) { + res, err := m.store.IncrementConsecutiveFailureCount(ctx, attempt.Destination.TenantID, attempt.Destination.ID, attempt.Attempt.ID) + if err != nil { + return false, fmt.Errorf("failed to get alert state: %w", err) + } + + // Replayed attempt (MQ redelivery, producer re-publish) that already + // completed a full evaluation — skip re-emitting alerts. Attempts that + // were counted but never marked evaluated (eval failed mid-way and the + // message was nacked) fall through and re-run as recovery. + if !res.NewlyCounted && res.AlreadyEvaluated { + m.logger.Ctx(ctx).Debug("skipping replayed attempt: already evaluated", zap.String("attempt_id", attempt.Attempt.ID), zap.String("tenant_id", attempt.Destination.TenantID), zap.String("destination_id", attempt.Destination.ID), ) + return true, nil } - return nil + count := res.Count + level, shouldAlert := m.evaluator.ShouldAlert(count) + if !shouldAlert { + return false, nil + } + + // At 100% threshold, disable the destination and emit disabled alert. + // Both operations are idempotent on replay: DisableDestination is a no-op + // if already disabled, and consumers deduplicate events by ID. + if level == 100 && m.disabler != nil { + if err := m.disabler.DisableDestination(ctx, attempt.Destination.TenantID, attempt.Destination.ID); err != nil { + return false, fmt.Errorf("failed to disable destination: %w", err) + } + + now := time.Now() + attempt.Destination.DisabledAt = &now + + m.logger.Ctx(ctx).Audit("destination disabled", + zap.String("attempt_id", attempt.Attempt.ID), + zap.String("event_id", attempt.Event.ID), + zap.String("tenant_id", attempt.Destination.TenantID), + zap.String("destination_id", attempt.Destination.ID), + zap.String("destination_type", attempt.Destination.Type), + ) + + disabledData := DestinationDisabledData{ + TenantID: attempt.Destination.TenantID, + Destination: attempt.Destination, + DisabledAt: now, + Reason: "consecutive_failure", + Event: attempt.Event, + Attempt: attempt.Attempt, + } + if err := m.emitter.Emit(ctx, opevents.TopicAlertDestinationDisabled, attempt.Destination.TenantID, disabledData); err != nil { + return false, fmt.Errorf("failed to emit destination disabled alert: %w", err) + } + } + + // Emit consecutive failure alert + cfData := ConsecutiveFailureData{ + TenantID: attempt.Destination.TenantID, + Event: attempt.Event, + Attempt: attempt.Attempt, + Destination: attempt.Destination, + ConsecutiveFailures: ConsecutiveFailures{ + Current: count, + Max: m.autoDisableFailureCount, + Threshold: level, + }, + } + if err := m.emitter.Emit(ctx, opevents.TopicAlertConsecutiveFailure, attempt.Destination.TenantID, cfData); err != nil { + return false, fmt.Errorf("failed to emit consecutive failure alert: %w", err) + } + + m.logger.Ctx(ctx).Audit("alert sent", + zap.String("topic", opevents.TopicAlertConsecutiveFailure), + zap.String("attempt_id", attempt.Attempt.ID), + zap.String("event_id", attempt.Event.ID), + zap.String("tenant_id", attempt.Destination.TenantID), + zap.String("destination_id", attempt.Destination.ID), + zap.String("destination_type", attempt.Destination.Type), + ) + + return false, nil } diff --git a/internal/alert/monitor_test.go b/internal/alert/monitor_test.go index 7df1f7cd6..a412f6eb6 100644 --- a/internal/alert/monitor_test.go +++ b/internal/alert/monitor_test.go @@ -600,3 +600,135 @@ func countEmitCalls(emitter *mockAlertEmitter, topic string) int { } return count } + +func TestAlertMonitor_ConsecutiveFailures_GateDisabled(t *testing.T) { + // With consecutive-failure alerting disabled, failures never emit cf/disabled + // alerts and never auto-disable, even past the threshold. + t.Parallel() + ctx := context.Background() + logger := testutil.CreateTestLogger(t) + redisClient := testutil.CreateTestRedisClient(t) + emitter := &mockAlertEmitter{} + emitter.On("Emit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) + disabler := &mockDestinationDisabler{} + disabler.On("DisableDestination", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + monitor := alert.NewAlertMonitor( + logger, + redisClient, + emitter, + 10, + alert.WithDisabler(disabler), + alert.WithAutoDisableFailureCount(5), + alert.WithConsecutiveFailureEnabled(false), + ) + + dest := &alert.AlertDestination{ID: "dest_cf_off", TenantID: "tenant_cf_off"} + event := &models.Event{Topic: "test.event"} + + // Well past the threshold of 5. + for i := 1; i <= 10; i++ { + attempt := alert.DeliveryAttempt{ + Event: event, + Destination: dest, + Attempt: &models.Attempt{ + ID: fmt.Sprintf("att_%d", i), + Status: "failed", + Code: "500", + Time: time.Now(), + }, + } + require.NoError(t, monitor.HandleAttempt(ctx, attempt)) + } + + require.Equal(t, 0, countEmitCalls(emitter, "alert.destination.consecutive_failure"), + "no consecutive_failure alerts when gate disabled") + require.Equal(t, 0, countEmitCalls(emitter, "alert.destination.disabled"), + "no disabled alerts when gate disabled") + disabler.AssertNotCalled(t, "DisableDestination", mock.Anything, mock.Anything, mock.Anything) +} + +func TestAlertMonitor_ExhaustedRetries_GateDisabled(t *testing.T) { + // With exhausted-retries alerting disabled, exceeding the retry limit emits + // nothing, even though retries are enabled and the event is eligible. + t.Parallel() + ctx := context.Background() + logger := testutil.CreateTestLogger(t) + redisClient := testutil.CreateTestRedisClient(t) + emitter := &mockAlertEmitter{} + emitter.On("Emit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + monitor := alert.NewAlertMonitor( + logger, + redisClient, + emitter, + 3, + alert.WithAutoDisableFailureCount(100), + alert.WithExhaustedRetriesEnabled(false), + ) + + dest := &alert.AlertDestination{ID: "dest_er_off", TenantID: "tenant_er_off"} + event := &models.Event{Topic: "test.event", EligibleForRetry: true} + + attempt := alert.DeliveryAttempt{ + Event: event, + Destination: dest, + Attempt: &models.Attempt{ + ID: "att_4", + AttemptNumber: 4, + Status: "failed", + Code: "500", + Time: time.Now(), + }, + } + require.NoError(t, monitor.HandleAttempt(ctx, attempt)) + + require.Equal(t, 0, countEmitCalls(emitter, "alert.attempt.exhausted_retries"), + "no exhausted_retries alerts when gate disabled") +} + +func TestAlertMonitor_Gates_Independent(t *testing.T) { + // Consecutive-failure gate off but exhausted-retries gate on: exhausted_retries + // still fires while consecutive_failure stays silent. + t.Parallel() + ctx := context.Background() + logger := testutil.CreateTestLogger(t) + redisClient := testutil.CreateTestRedisClient(t) + emitter := &mockAlertEmitter{} + emitter.On("Emit", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + monitor := alert.NewAlertMonitor( + logger, + redisClient, + emitter, + 3, + alert.WithAutoDisableFailureCount(5), + alert.WithConsecutiveFailureEnabled(false), + alert.WithExhaustedRetriesEnabled(true), + ) + + dest := &alert.AlertDestination{ID: "dest_mix", TenantID: "tenant_mix"} + event := &models.Event{Topic: "test.event", EligibleForRetry: true} + + // Enough failures to cross the cf threshold, with the last one exceeding the + // retry limit so exhausted_retries is eligible. + for i := 1; i <= 6; i++ { + attempt := alert.DeliveryAttempt{ + Event: event, + Destination: dest, + Attempt: &models.Attempt{ + ID: fmt.Sprintf("att_%d", i), + AttemptNumber: i, + Status: "failed", + Code: "500", + Time: time.Now(), + }, + } + require.NoError(t, monitor.HandleAttempt(ctx, attempt)) + } + + require.Equal(t, 0, countEmitCalls(emitter, "alert.destination.consecutive_failure"), + "consecutive_failure stays silent when its gate is off") + require.Greater(t, countEmitCalls(emitter, "alert.attempt.exhausted_retries"), 0, + "exhausted_retries still fires when its gate is on") +} diff --git a/internal/alert/settings.go b/internal/alert/settings.go new file mode 100644 index 000000000..78e3d4e67 --- /dev/null +++ b/internal/alert/settings.go @@ -0,0 +1,34 @@ +package alert + +// Default alert values, applied when the corresponding config value is unset. +const ( + DefaultConsecutiveFailureCount = 100 + DefaultExhaustedRetriesWindowSeconds = 3600 +) + +// Settings is the resolved, operational alert configuration consumed by the +// service builder. The config package produces it from raw env/yaml values via +// AlertConfig.ToConfig, so the rest of the codebase never deals with the raw +// unset / empty / value strings. +type Settings struct { + ConsecutiveFailure ConsecutiveFailureSetting + ExhaustedRetries ExhaustedRetriesSetting + AutoDisableDestination bool +} + +// ConsecutiveFailureSetting controls consecutive-failure alerting. When Enabled +// is false the monitor never tracks or alerts on consecutive failures, and +// therefore never auto-disables a destination regardless of AutoDisableDestination. +type ConsecutiveFailureSetting struct { + Enabled bool + Count int +} + +// ExhaustedRetriesSetting controls exhausted-retries alerting. When Enabled is +// false the monitor never emits exhausted_retries alerts. WindowSeconds is the +// suppression window for duplicate alerts; 0 means no suppression (alert on +// every exhaustion). +type ExhaustedRetriesSetting struct { + Enabled bool + WindowSeconds int +} diff --git a/internal/config/alert_test.go b/internal/config/alert_test.go new file mode 100644 index 000000000..2850f901c --- /dev/null +++ b/internal/config/alert_test.go @@ -0,0 +1,211 @@ +package config_test + +import ( + "testing" + + "github.com/hookdeck/outpost/internal/alert" + "github.com/hookdeck/outpost/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestAlertConfig_ParsePaths verifies the three-state contract for an alert +// setting — unset means "use the default", empty means "disabled", and a value +// means "use that value" — and that it holds identically whether the setting +// comes from the YAML config file or an environment variable. +// +// It runs the full parse path (file/env -> config -> resolved settings) so that +// the two surfaces are exercised exactly as a deployment would configure them. +// The empty-string-disables case in particular must behave the same on both. +func TestAlertConfig_ParsePaths(t *testing.T) { + parseYAML := func(t *testing.T, yamlBody string) alert.Settings { + t.Helper() + m := &mockOS{ + files: map[string][]byte{"/c.yaml": []byte(yamlBody)}, + envVars: map[string]string{}, + } + cfg, err := config.ParseWithoutValidation(config.Flags{Config: "/c.yaml"}, m) + require.NoError(t, err) + s, err := cfg.Alert.ToConfig() + require.NoError(t, err) + return s + } + parseEnv := func(t *testing.T, env map[string]string) alert.Settings { + t.Helper() + m := &mockOS{files: map[string][]byte{}, envVars: env} + cfg, err := config.ParseWithoutValidation(config.Flags{}, m) + require.NoError(t, err) + s, err := cfg.Alert.ToConfig() + require.NoError(t, err) + return s + } + parseBoth := func(t *testing.T, yamlBody string, env map[string]string) alert.Settings { + t.Helper() + m := &mockOS{files: map[string][]byte{"/c.yaml": []byte(yamlBody)}, envVars: env} + cfg, err := config.ParseWithoutValidation(config.Flags{Config: "/c.yaml"}, m) + require.NoError(t, err) + s, err := cfg.Alert.ToConfig() + require.NoError(t, err) + return s + } + cf := func(s alert.Settings) alert.ConsecutiveFailureSetting { return s.ConsecutiveFailure } + + t.Run("yaml: alert key absent -> default", func(t *testing.T) { + got := cf(parseYAML(t, "log_level: debug\n")) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: true, Count: 100}, got) + }) + t.Run("yaml: set to empty string -> disabled", func(t *testing.T) { + got := cf(parseYAML(t, "alert:\n consecutive_failure_count: \"\"\n")) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: false, Count: 0}, got) + }) + t.Run("yaml: key present but no value -> default", func(t *testing.T) { + // `key:` with nothing after it is not the same as an empty string; it + // reads as "not provided", so it falls back to the default rather than + // disabling. Only an explicit empty string disables. + got := cf(parseYAML(t, "alert:\n consecutive_failure_count:\n")) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: true, Count: 100}, got) + }) + t.Run("yaml: value -> value", func(t *testing.T) { + got := cf(parseYAML(t, "alert:\n consecutive_failure_count: \"5\"\n")) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: true, Count: 5}, got) + }) + + t.Run("env: var absent -> default", func(t *testing.T) { + got := cf(parseEnv(t, map[string]string{})) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: true, Count: 100}, got) + }) + t.Run("env: set to empty string -> disabled", func(t *testing.T) { + // Setting the env var to an empty string must disable the dimension, the + // same way `key: ""` does in YAML — an empty env var is a deliberate + // "off", distinct from leaving the var unset (which uses the default). + got := cf(parseEnv(t, map[string]string{"ALERT_CONSECUTIVE_FAILURE_COUNT": ""})) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: false, Count: 0}, got) + }) + t.Run("env: value -> value", func(t *testing.T) { + got := cf(parseEnv(t, map[string]string{"ALERT_CONSECUTIVE_FAILURE_COUNT": "5"})) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: true, Count: 5}, got) + }) + + // When a setting is provided in both places, the env var wins over YAML. + t.Run("both: env value overrides yaml value", func(t *testing.T) { + got := cf(parseBoth(t, + "alert:\n consecutive_failure_count: \"5\"\n", + map[string]string{"ALERT_CONSECUTIVE_FAILURE_COUNT": "9"})) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: true, Count: 9}, got) + }) + t.Run("both: empty env overrides yaml value (disables)", func(t *testing.T) { + got := cf(parseBoth(t, + "alert:\n consecutive_failure_count: \"5\"\n", + map[string]string{"ALERT_CONSECUTIVE_FAILURE_COUNT": ""})) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: false, Count: 0}, got) + }) + t.Run("both: env value overrides yaml disable", func(t *testing.T) { + got := cf(parseBoth(t, + "alert:\n consecutive_failure_count: \"\"\n", + map[string]string{"ALERT_CONSECUTIVE_FAILURE_COUNT": "9"})) + assert.Equal(t, alert.ConsecutiveFailureSetting{Enabled: true, Count: 9}, got) + }) +} + +func TestAlertConfig_ToConfig(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg config.AlertConfig + want alert.Settings + wantErr bool + }{ + { + name: "unset uses defaults", + cfg: config.AlertConfig{}, + want: alert.Settings{ + ConsecutiveFailure: alert.ConsecutiveFailureSetting{Enabled: true, Count: 100}, + ExhaustedRetries: alert.ExhaustedRetriesSetting{Enabled: true, WindowSeconds: 3600}, + }, + }, + { + name: "empty string disables both dimensions", + cfg: config.AlertConfig{ + ConsecutiveFailureCount: config.NewOptionalString(""), + ExhaustedRetriesWindowSeconds: config.NewOptionalString(""), + }, + want: alert.Settings{ + ConsecutiveFailure: alert.ConsecutiveFailureSetting{Enabled: false, Count: 0}, + ExhaustedRetries: alert.ExhaustedRetriesSetting{Enabled: false, WindowSeconds: 0}, + }, + }, + { + name: "explicit values", + cfg: config.AlertConfig{ + ConsecutiveFailureCount: config.NewOptionalString("50"), + ExhaustedRetriesWindowSeconds: config.NewOptionalString("120"), + }, + want: alert.Settings{ + ConsecutiveFailure: alert.ConsecutiveFailureSetting{Enabled: true, Count: 50}, + ExhaustedRetries: alert.ExhaustedRetriesSetting{Enabled: true, WindowSeconds: 120}, + }, + }, + { + name: "surrounding whitespace is trimmed", + cfg: config.AlertConfig{ + ConsecutiveFailureCount: config.NewOptionalString(" 50 "), + }, + want: alert.Settings{ + ConsecutiveFailure: alert.ConsecutiveFailureSetting{Enabled: true, Count: 50}, + ExhaustedRetries: alert.ExhaustedRetriesSetting{Enabled: true, WindowSeconds: 3600}, + }, + }, + { + name: "exhausted window zero means enabled with no suppression", + cfg: config.AlertConfig{ExhaustedRetriesWindowSeconds: config.NewOptionalString("0")}, + want: alert.Settings{ + ConsecutiveFailure: alert.ConsecutiveFailureSetting{Enabled: true, Count: 100}, + ExhaustedRetries: alert.ExhaustedRetriesSetting{Enabled: true, WindowSeconds: 0}, + }, + }, + { + name: "auto_disable_destination is carried through", + cfg: config.AlertConfig{AutoDisableDestination: true}, + want: alert.Settings{ + ConsecutiveFailure: alert.ConsecutiveFailureSetting{Enabled: true, Count: 100}, + ExhaustedRetries: alert.ExhaustedRetriesSetting{Enabled: true, WindowSeconds: 3600}, + AutoDisableDestination: true, + }, + }, + { + name: "consecutive zero is invalid (min 1)", + cfg: config.AlertConfig{ConsecutiveFailureCount: config.NewOptionalString("0")}, + wantErr: true, + }, + { + name: "consecutive negative is invalid", + cfg: config.AlertConfig{ConsecutiveFailureCount: config.NewOptionalString("-1")}, + wantErr: true, + }, + { + name: "consecutive non-numeric is invalid", + cfg: config.AlertConfig{ConsecutiveFailureCount: config.NewOptionalString("abc")}, + wantErr: true, + }, + { + name: "exhausted negative is invalid", + cfg: config.AlertConfig{ExhaustedRetriesWindowSeconds: config.NewOptionalString("-5")}, + wantErr: true, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got, err := tt.cfg.ToConfig() + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 9bd97c69f..4c08bf9c7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -3,10 +3,12 @@ package config import ( "errors" "fmt" + "strconv" "strings" "time" "github.com/caarlos0/env/v9" + "github.com/hookdeck/outpost/internal/alert" "github.com/hookdeck/outpost/internal/backoff" "github.com/hookdeck/outpost/internal/clickhouse" "github.com/hookdeck/outpost/internal/migrator" @@ -190,11 +192,9 @@ func (c *Config) InitDefaults() { }, } - c.Alert = AlertConfig{ - ConsecutiveFailureCount: 100, - AutoDisableDestination: false, - ExhaustedRetriesWindowSeconds: 3600, - } + // Alert: ConsecutiveFailureCount / ExhaustedRetriesWindowSeconds are left + // unset here so their defaults (and the empty-string "disabled" sentinel) + // are applied by AlertConfig.ToConfig rather than baked in as int zero values. c.Telemetry = TelemetryConfig{ Disabled: false, @@ -278,11 +278,33 @@ func (c *Config) parseEnvVariables(osInterface OSInterface) error { envMap[env[:i]] = env[i+1:] } } - return env.ParseWithOptions(c, env.Options{Environment: envMap}) + if err := env.ParseWithOptions(c, env.Options{Environment: envMap}); err != nil { + return err + } + } else { + // For real OS, use env.Parse directly + if err := env.Parse(c); err != nil { + return err + } } - // For real OS, use env.Parse directly - return env.Parse(c) + c.captureEmptyAlertEnv(osInterface) + return nil +} + +// captureEmptyAlertEnv honors "an empty env var disables this alert dimension". +// caarlos0/env ignores a present-but-empty env var (it never invokes the field's +// unmarshaler), so it can't set these to empty on its own. We detect presence +// via LookupEnv and apply the empty value explicitly. A present env var takes +// precedence over a YAML value (env > yaml); present non-empty values are +// already bound by caarlos0/env above. +func (c *Config) captureEmptyAlertEnv(osInterface OSInterface) { + if v, ok := osInterface.LookupEnv("ALERT_CONSECUTIVE_FAILURE_COUNT"); ok && v == "" { + c.Alert.ConsecutiveFailureCount = NewOptionalString("") + } + if v, ok := osInterface.LookupEnv("ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS"); ok && v == "" { + c.Alert.ExhaustedRetriesWindowSeconds = NewOptionalString("") + } } func (c *Config) normalizeTopics() { @@ -475,9 +497,62 @@ func (c *OperatorEventsConfig) ToConfig() opevents.Config { } type AlertConfig struct { - ConsecutiveFailureCount int `yaml:"consecutive_failure_count" env:"ALERT_CONSECUTIVE_FAILURE_COUNT" desc:"Number of consecutive delivery failures for a destination before triggering an alert and potentially disabling it." required:"N"` - AutoDisableDestination bool `yaml:"auto_disable_destination" env:"ALERT_AUTO_DISABLE_DESTINATION" desc:"If true, automatically disables a destination after 'consecutive_failure_count' is reached." required:"N"` - ExhaustedRetriesWindowSeconds int `yaml:"exhausted_retries_window_seconds" env:"ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS" desc:"Suppression window in seconds for exhausted_retries alerts. First exhaustion per destination emits an alert; subsequent exhaustions within the window are suppressed." required:"N"` + ConsecutiveFailureCount OptionalString `yaml:"consecutive_failure_count" env:"ALERT_CONSECUTIVE_FAILURE_COUNT" desc:"Number of consecutive delivery failures before alerting on a destination and, with auto_disable_destination, disabling it. Leave unset for the default of 100; set to an empty string to disable consecutive-failure alerting entirely." required:"N"` + AutoDisableDestination bool `yaml:"auto_disable_destination" env:"ALERT_AUTO_DISABLE_DESTINATION" desc:"If true, automatically disables a destination when consecutive_failure_count is reached. Has no effect when consecutive-failure alerting is disabled." required:"N"` + ExhaustedRetriesWindowSeconds OptionalString `yaml:"exhausted_retries_window_seconds" env:"ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS" desc:"Suppression window in seconds for exhausted_retries alerts; the first exhaustion per destination emits an alert and subsequent ones within the window are suppressed (0 = no suppression). Leave unset for the default of 3600; set to an empty string to disable exhausted_retries alerting entirely." required:"N"` +} + +// ToConfig resolves the raw alert config into operational alert.Settings. For +// the two count fields the rule is: unset (nil) uses the built-in default, an +// empty string disables that alert dimension, and any other value must parse to +// a non-negative integer. It returns an error on a non-numeric or out-of-range +// value so Validate can reject it at startup. +func (c *AlertConfig) ToConfig() (alert.Settings, error) { + consecutive, err := resolveAlertCount(c.ConsecutiveFailureCount, alert.DefaultConsecutiveFailureCount, 1) + if err != nil { + return alert.Settings{}, fmt.Errorf("alert.consecutive_failure_count: %w", err) + } + exhausted, err := resolveAlertCount(c.ExhaustedRetriesWindowSeconds, alert.DefaultExhaustedRetriesWindowSeconds, 0) + if err != nil { + return alert.Settings{}, fmt.Errorf("alert.exhausted_retries_window_seconds: %w", err) + } + return alert.Settings{ + ConsecutiveFailure: alert.ConsecutiveFailureSetting{ + Enabled: consecutive.enabled, + Count: consecutive.value, + }, + ExhaustedRetries: alert.ExhaustedRetriesSetting{ + Enabled: exhausted.enabled, + WindowSeconds: exhausted.value, + }, + AutoDisableDestination: c.AutoDisableDestination, + }, nil +} + +type resolvedAlertCount struct { + enabled bool + value int +} + +// resolveAlertCount applies the unset/empty/value rule to a single raw field. +// unset -> {enabled, defaultValue}; "" -> {disabled, 0}; else parse and require +// value >= min. +func resolveAlertCount(raw OptionalString, defaultValue, min int) (resolvedAlertCount, error) { + value, set := raw.Get() + if !set { + return resolvedAlertCount{enabled: true, value: defaultValue}, nil + } + if value == "" { + return resolvedAlertCount{enabled: false, value: 0}, nil + } + n, err := strconv.Atoi(strings.TrimSpace(value)) + if err != nil { + return resolvedAlertCount{}, fmt.Errorf("must be an integer, an empty string to disable, or unset for the default: %q", value) + } + if n < min { + return resolvedAlertCount{}, fmt.Errorf("must be >= %d, got %d", min, n) + } + return resolvedAlertCount{enabled: true, value: n}, nil } // ConfigFilePath returns the path of the config file that was used diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5968e3c85..ee14dae23 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -19,6 +19,11 @@ func (m *mockOS) Getenv(key string) string { return m.envVars[key] } +func (m *mockOS) LookupEnv(key string) (string, bool) { + v, ok := m.envVars[key] + return v, ok +} + func (m *mockOS) Stat(name string) (os.FileInfo, error) { if err, ok := m.statErrs[name]; ok { return nil, err diff --git a/internal/config/logging.go b/internal/config/logging.go index 36e23ea91..116c3f0f6 100644 --- a/internal/config/logging.go +++ b/internal/config/logging.go @@ -20,6 +20,10 @@ import ( // // See contributing/config.md for detailed guidelines on configuration logging. func (c *Config) LogConfigurationSummary() []zap.Field { + // Resolve alert settings so the summary logs the effective values. The error + // is ignored — config is already validated by the time this runs. + alertSettings, _ := c.Alert.ToConfig() + fields := []zap.Field{ // General zap.String("service", c.Service), @@ -92,8 +96,11 @@ func (c *Config) LogConfigurationSummary() []zap.Field { zap.Bool("telemetry_disabled", c.Telemetry.Disabled || c.DisableTelemetry), // Alert - zap.Int("alert_consecutive_failure_count", c.Alert.ConsecutiveFailureCount), + zap.Bool("alert_consecutive_failure_enabled", alertSettings.ConsecutiveFailure.Enabled), + zap.Int("alert_consecutive_failure_count", alertSettings.ConsecutiveFailure.Count), zap.Bool("alert_auto_disable_destination", c.Alert.AutoDisableDestination), + zap.Bool("alert_exhausted_retries_enabled", alertSettings.ExhaustedRetries.Enabled), + zap.Int("alert_exhausted_retries_window_seconds", alertSettings.ExhaustedRetries.WindowSeconds), // ID Generation zap.String("idgen_type", c.IDGen.Type), diff --git a/internal/config/optional.go b/internal/config/optional.go new file mode 100644 index 000000000..d19907a1d --- /dev/null +++ b/internal/config/optional.go @@ -0,0 +1,46 @@ +package config + +import "gopkg.in/yaml.v3" + +// OptionalString is a config value that records whether it was provided at all, +// distinguishing "unset" from "explicitly set to empty string". This lets a +// single field express three states — unset, empty, value — which a plain +// string cannot. +// +// It implements both encoding.TextUnmarshaler (so caarlos0/env binds it as a +// scalar — avoiding the pointer-recursion crash a *string triggers) and +// yaml.Unmarshaler (so YAML can express the empty state via `key: ""`). +// +// One gap remains: caarlos0/env ignores a present-but-empty env var entirely and +// never invokes UnmarshalText for it, so the empty-env case is handled +// explicitly during parsing via OSInterface.LookupEnv (see captureEmptyEnv). +type OptionalString struct { + set bool + value string +} + +// NewOptionalString returns a set OptionalString. Intended for tests and +// programmatic config construction. +func NewOptionalString(value string) OptionalString { + return OptionalString{set: true, value: value} +} + +// Get returns the value and whether it was set. +func (o OptionalString) Get() (string, bool) { + return o.value, o.set +} + +func (o *OptionalString) UnmarshalText(b []byte) error { + o.set = true + o.value = string(b) + return nil +} + +func (o *OptionalString) UnmarshalYAML(node *yaml.Node) error { + // A bare `key:` (no value) is YAML null — treat as unset, matching an absent key. + if node.Tag == "!!null" { + return nil + } + o.set = true + return node.Decode(&o.value) +} diff --git a/internal/config/os.go b/internal/config/os.go index 692db43cd..8f3e6778e 100644 --- a/internal/config/os.go +++ b/internal/config/os.go @@ -4,6 +4,10 @@ import "os" type OSInterface interface { Getenv(key string) string + // LookupEnv reports whether the variable is present (ok) in addition to its + // value, so callers can distinguish "unset" from "set to empty string" — + // a distinction caarlos0/env does not surface. + LookupEnv(key string) (string, bool) Stat(name string) (os.FileInfo, error) ReadFile(name string) ([]byte, error) Environ() []string @@ -17,6 +21,10 @@ func (d *defaultOSImpl) Getenv(key string) string { return os.Getenv(key) } +func (d *defaultOSImpl) LookupEnv(key string) (string, bool) { + return os.LookupEnv(key) +} + func (d *defaultOSImpl) Stat(name string) (os.FileInfo, error) { return os.Stat(name) } diff --git a/internal/config/validation.go b/internal/config/validation.go index 5b9a7c5b6..377c65d80 100644 --- a/internal/config/validation.go +++ b/internal/config/validation.go @@ -49,11 +49,24 @@ func (c *Config) Validate(flags Flags) error { return err } + if err := c.validateAlert(); err != nil { + return err + } + // Mark as validated if we get here c.validated = true return nil } +// validateAlert rejects malformed alert config (non-numeric or out-of-range +// consecutive_failure_count / exhausted_retries_window_seconds) at startup. +func (c *Config) validateAlert() error { + if _, err := c.Alert.ToConfig(); err != nil { + return err + } + return nil +} + // validateService validates the service configuration func (c *Config) validateService(flags Flags) error { // Parse service type from flag & env diff --git a/internal/services/builder.go b/internal/services/builder.go index dd57b44c8..fa8c1f824 100644 --- a/internal/services/builder.go +++ b/internal/services/builder.go @@ -351,14 +351,26 @@ func (b *ServiceBuilder) BuildLogWorker(baseRouter *gin.Engine) error { } emitter := opevents.NewEmitter(sink, b.cfg.DeploymentID, oeCfg.Topics) + alertSettings, err := b.cfg.Alert.ToConfig() + if err != nil { + return fmt.Errorf("failed to resolve alert config: %w", err) + } + var disabler alert.DestinationDisabler - if b.cfg.Alert.AutoDisableDestination { + if alertSettings.AutoDisableDestination { disabler = newDestinationDisabler(svc.tenantStore) } - exhaustedRetriesIdemp := idempotence.New(svc.redisClient, - idempotence.WithSuccessfulTTL(time.Duration(b.cfg.Alert.ExhaustedRetriesWindowSeconds)*time.Second), - idempotence.WithDeploymentID(b.cfg.DeploymentID), - ) + + // Build a suppression window only when exhausted-retries alerting is enabled + // with a positive window. A zero window means "alert on every exhaustion" + // (no suppression), so we leave the idempotence instance nil. + var exhaustedRetriesIdemp idempotence.Idempotence + if alertSettings.ExhaustedRetries.Enabled && alertSettings.ExhaustedRetries.WindowSeconds > 0 { + exhaustedRetriesIdemp = idempotence.New(svc.redisClient, + idempotence.WithSuccessfulTTL(time.Duration(alertSettings.ExhaustedRetries.WindowSeconds)*time.Second), + idempotence.WithDeploymentID(b.cfg.DeploymentID), + ) + } _, retryMaxLimit := b.cfg.GetRetryBackoff() alertMonitor := alert.NewAlertMonitor( b.logger, @@ -367,7 +379,9 @@ func (b *ServiceBuilder) BuildLogWorker(baseRouter *gin.Engine) error { retryMaxLimit, alert.WithDisabler(disabler), alert.WithExhaustedRetriesIdempotence(exhaustedRetriesIdemp), - alert.WithAutoDisableFailureCount(b.cfg.Alert.ConsecutiveFailureCount), + alert.WithConsecutiveFailureEnabled(alertSettings.ConsecutiveFailure.Enabled), + alert.WithAutoDisableFailureCount(alertSettings.ConsecutiveFailure.Count), + alert.WithExhaustedRetriesEnabled(alertSettings.ExhaustedRetries.Enabled), alert.WithDeploymentID(b.cfg.DeploymentID), )