diff --git a/README.md b/README.md index 06acc8b..d17cc7d 100644 --- a/README.md +++ b/README.md @@ -1134,6 +1134,52 @@ delivery as `retrying` and schedules a retry timer for after the remaining cooldown period. This ensures no deliveries are lost — they're just delayed until the target is healthy again. +### Metrics + +`/metrics` serves one Prometheus registry behind basic auth (see +[Infrastructure Endpoints](#infrastructure-endpoints)). Alongside the +inbound HTTP metrics recorded by the middleware, it exposes the +delivery pipeline — the part of the service that can be failing while +the receive side looks perfectly healthy, because it is: events are +arriving and being stored, they are just not getting anywhere. + +| Metric | Type | Meaning | +| ------ | ---- | ------- | +| `webhooker_events_received_total` | counter | Events received and durably stored. Compare against the delivery counters on one dashboard | +| `webhooker_delivery_attempts_total` | counter | Delivery attempts actually dispatched to a target. A delivery an open circuit breaker refused is not one: it is counted as a retry instead | +| `webhooker_deliveries_succeeded_total` | counter | Deliveries that reached `delivered` | +| `webhooker_deliveries_failed_total` | counter | Deliveries that failed terminally and will not be retried | +| `webhooker_delivery_retries_total` | counter | Deliveries put back into `retrying` | +| `webhooker_delivery_duration_seconds` | histogram | Wall time of a single dispatched delivery attempt, the same duration the attempt's `DeliveryResult` records | +| `webhooker_deliveries_pending` | gauge | Deliveries currently in `pending` | +| `webhooker_deliveries_retrying` | gauge | Deliveries currently in `retrying` | +| `webhooker_circuit_breakers_open` | gauge | Circuit breakers currently open | + +Every delivery metric carries exactly one label, `target_type`, and +cardinality is the whole reason for that restriction. A target type is +one of four compile-time constants, so the label domain is bounded by +construction; a value outside that set collapses to `unknown` rather +than minting a series of its own. Target ids, event ids and entrypoint +ids are deliberately not labels: they are UUIDs minted per operator +action or per inbound request, a series is never reclaimed once it +exists, and labelling by any of them would make `/metrics` a memory +leak that grows with traffic. + +The two queue-depth gauges are counted out of the databases by a +sampler that runs every 30 seconds for as long as the delivery engine +does, rather than tracked as deltas alongside the status transitions: a +delta would have to be seeded at startup from rows a previous process +wrote, and would drift permanently on any transition that failed to +persist. + +Those two gauges also publish an `unknown` series, from startup rather +than on first occurrence. Deliveries queued against a target that has +since been deleted are counted there: that backlog is the one nobody is +watching, so it is the one that must not silently vanish from the +gauge. The outcome counters move only after the status change has been +written, so a transition the database rejected is never reported as an +outcome that happened. + ### Rate Limiting Global blanket rate limiting middleware (e.g., a per-IP throttle shared @@ -1759,6 +1805,7 @@ webhooker/ │ │ ├── target_log.go # Log target (stdout) │ │ ├── target_config_view.go # Masked target config for templates │ │ ├── archive_sweeper.go # Periodic pruning of idle archives +│ │ ├── queue_depth.go # Periodic sampler behind the queue-depth gauges │ │ ├── url_mask.go # Strips credentials from *url.Error │ │ └── ssrf.go # SSRF prevention (IP validation, safe HTTP transport) │ ├── handlers/ @@ -1776,6 +1823,8 @@ webhooker/ │ │ └── lifecycle.go # Shared stop-hook waiter, bounded by the stop context │ ├── logger/ │ │ └── logger.go # slog setup with TTY detection +│ ├── metrics/ +│ │ └── metrics.go # Delivery Prometheus collectors, labelled by target type │ ├── middleware/ │ │ ├── middleware.go # Logging, CORS, Auth, Metrics, MetricsAuth, SecurityHeaders, MaxBodySize │ │ ├── csrf.go # CSRF protection middleware (gorilla/csrf) diff --git a/go.mod b/go.mod index c51849e..d2c5db6 100644 --- a/go.mod +++ b/go.mod @@ -13,6 +13,7 @@ require ( github.com/gorilla/sessions v1.4.0 github.com/joho/godotenv v1.5.1 github.com/prometheus/client_golang v1.18.0 + github.com/prometheus/client_model v0.5.0 github.com/slok/go-http-metrics v0.11.0 github.com/stretchr/testify v1.8.4 go.uber.org/fx v1.20.1 @@ -38,7 +39,6 @@ require ( github.com/mattn/go-sqlite3 v1.14.17 // indirect github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect - github.com/prometheus/client_model v0.5.0 // indirect github.com/prometheus/common v0.45.0 // indirect github.com/prometheus/procfs v0.12.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect diff --git a/internal/delivery/engine.go b/internal/delivery/engine.go index a35ebd1..12f238d 100644 --- a/internal/delivery/engine.go +++ b/internal/delivery/engine.go @@ -15,6 +15,7 @@ import ( "sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/lifecycle" "sneak.berlin/go/webhooker/internal/logger" + "sneak.berlin/go/webhooker/internal/metrics" ) const ( @@ -139,6 +140,12 @@ type Engine struct { retryCh chan Task workers int + // mtr is the delivery metric set. Production wires the + // process-wide one; a test can substitute a set registered on + // a private registry so its assertions are not disturbed by + // deliveries other tests are making at the same time. + mtr *metrics.Set + // targets maps each target type to its implementation. targets map[database.TargetType]Target @@ -164,6 +171,7 @@ func New( deliveryCh: make(chan Task, deliveryChannelSize), retryCh: make(chan Task, retryChannelSize), workers: defaultWorkers, + mtr: metrics.Default(), } e.initTargets(&http.Client{ @@ -283,6 +291,10 @@ func (e *Engine) start() { go e.retrySweep(ctx) + e.wg.Add(1) + + go e.queueDepthSampler(ctx) + e.log.Info( "delivery engine started", "workers", e.workers, @@ -837,8 +849,15 @@ func (e *Engine) failUnretryableRetry( 0, ) + // The type is passed rather than assigned onto d: the delivery + // is loaded here without its target relation, and populating + // d.Target would make GORM's SaveBeforeAssociations upsert the + // whole target row — plaintext config, which for a slack target + // is the credential — into the per-webhook event database. See + // https://git.eeqj.de/sneak/webhooker/issues/206. e.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusFailed, + webhookDB, d, target.Type, + database.DeliveryStatusFailed, ) } @@ -859,7 +878,8 @@ func (e *Engine) processDelivery( ) e.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusFailed, + webhookDB, d, d.Target.Type, + database.DeliveryStatusFailed, ) return @@ -868,6 +888,24 @@ func (e *Engine) processDelivery( target.Deliver(ctx, webhookDB, d, task, e) } +// observeAttempt counts one delivery attempt that was actually +// dispatched to a target, and records how long it took. +// +// It is called from the dispatch paths rather than from around +// Target.Deliver, because Deliver is also entered for deliveries +// that never reach the wire: a delivery an open circuit breaker +// refuses sends nothing, records no DeliveryResult, and is +// rescheduled. Counting those would climb the attempts counter with +// no traffic behind it and fill the duration histogram with +// microsecond samples, which would make the delivery-duration +// quantiles improve during exactly the outage they exist to reveal. +func (e *Engine) observeAttempt( + t database.TargetType, dur time.Duration, +) { + e.mtr.DeliveryAttempted(t) + e.mtr.ObserveDeliveryDuration(t, dur) +} + // recordResult persists a DeliveryResult row describing a // single attempt. It is a cross-target helper the targets // call. @@ -901,10 +939,22 @@ func (e *Engine) recordResult( } // updateDeliveryStatus persists a new status for a delivery. -// It is a cross-target helper the targets call. +// It is a cross-target helper the targets call, and therefore the +// single point where a delivery's outcome — delivered, terminally +// failed, or put back into retry — is counted. +// +// The target type is a parameter rather than read off d.Target +// because one caller — failUnretryableRetry — deliberately holds a +// delivery loaded without its target relation, and must keep it that +// way: a populated d.Target makes GORM upsert the target row, config +// and all, into the per-webhook database. +// +// The counter moves only after the row is written, so a transition +// the database rejected is not claimed as an outcome that happened. func (e *Engine) updateDeliveryStatus( webhookDB *gorm.DB, d *database.Delivery, + targetType database.TargetType, status database.DeliveryStatus, ) { err := webhookDB.Model(d). @@ -916,7 +966,11 @@ func (e *Engine) updateDeliveryStatus( "status", status, "error", err, ) + + return } + + e.mtr.DeliveryStatusChanged(targetType, status) } func truncate(s string, maxLen int) string { diff --git a/internal/delivery/engine_integration_test.go b/internal/delivery/engine_integration_test.go index 966ef0b..0f4738b 100644 --- a/internal/delivery/engine_integration_test.go +++ b/internal/delivery/engine_integration_test.go @@ -886,6 +886,82 @@ func TestSweepSingleRetry_TypeNoLongerRetries( ) } +// TestFailUnretryableRetry_WritesNoTargetRow proves the +// orphaned-retry terminal path leaves no target row — and so no +// plaintext target config — in the per-webhook event database. +// +// That path loads the delivery without its Target relation on +// purpose. Populating d.Target makes GORM's SaveBeforeAssociations +// upsert the whole target row on the status UPDATE, which for a slack +// target writes the incoming-webhook credential into events-*.db. +// See https://git.eeqj.de/sneak/webhooker/issues/206. +func TestFailUnretryableRetry_WritesNoTargetRow( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "no-target-row", + ) + + targetID := uuid.New().String() + + // A Slack incoming-webhook URL: the target config IS the + // credential, which is what makes a leaked target row a + // disclosure rather than a curiosity. + hookURL := "https://hooks.slack.com/services/T00/B00/x" + + iCreateTarget(t, s.MainDB, targetID, + s.WebhookID, "credential-bearing", + database.TargetTypeLog, iHTTPConfig(hookURL), 5, + ) + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"orphaned":"retry"}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusRetrying, + ) + + iSeedFailedResult(t, s.WebhookDB, d.ID) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + iAssertStatus(t, s.WebhookDB, d.ID, + database.DeliveryStatusFailed, + ) + + // The table exists in the per-webhook database because GORM + // migrates the Delivery relation's model alongside it. It must + // stay empty. + var targetRows int64 + + require.NoError(t, s.WebhookDB. + Table("targets"). + Count(&targetRows).Error) + + assert.Zero(t, targetRows, + "orphaned-retry terminal failure wrote a target row "+ + "into the per-webhook event database", + ) + + var configs []string + + require.NoError(t, s.WebhookDB. + Table("targets"). + Pluck("config", &configs).Error) + + assert.NotContains( + t, strings.Join(configs, " "), hookURL, + ) +} + func TestRecoverSingleRetry_UnknownTargetType( t *testing.T, ) { diff --git a/internal/delivery/export_test.go b/internal/delivery/export_test.go index fc3f434..ca37c59 100644 --- a/internal/delivery/export_test.go +++ b/internal/delivery/export_test.go @@ -10,6 +10,7 @@ import ( "go.uber.org/fx" "gorm.io/gorm" "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/metrics" ) // ErrExportArchiveWriterEvicted exposes the sentinel returned by @@ -253,6 +254,7 @@ func NewTestEngine( deliveryCh: make(chan Task, deliveryChannelSize), retryCh: make(chan Task, retryChannelSize), workers: workers, + mtr: metrics.Default(), } e.initTargets(client) @@ -267,6 +269,7 @@ func NewTestEngineSmallRetry( e := &Engine{ log: log, retryCh: make(chan Task, 1), + mtr: metrics.Default(), } e.initTargets(nil) @@ -289,12 +292,25 @@ func NewTestEngineWithDB( deliveryCh: make(chan Task, deliveryChannelSize), retryCh: make(chan Task, retryChannelSize), workers: workers, + mtr: metrics.Default(), } e.initTargets(client) return e } +// ExportSetMetrics substitutes the engine's metric set, so a test can +// assert on collectors registered on a private registry instead of +// the process-wide ones every other test is also moving. +func (e *Engine) ExportSetMetrics(mtr *metrics.Set) { + e.mtr = mtr +} + +// ExportSampleQueueDepths runs one queue depth sample synchronously. +func (e *Engine) ExportSampleQueueDepths(ctx context.Context) { + e.sampleQueueDepths(ctx) +} + // NewTestCircuitBreaker creates a CircuitBreaker with // custom settings for testing. func NewTestCircuitBreaker( diff --git a/internal/delivery/metrics_test.go b/internal/delivery/metrics_test.go new file mode 100644 index 0000000..284114a --- /dev/null +++ b/internal/delivery/metrics_test.go @@ -0,0 +1,545 @@ +package delivery_test + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/delivery" + "sneak.berlin/go/webhooker/internal/metrics" +) + +// Metric names as exposed on /metrics. +const ( + mAttempts = "webhooker_delivery_attempts_total" + mSucceeded = "webhooker_deliveries_succeeded_total" + mFailed = "webhooker_deliveries_failed_total" + mRetries = "webhooker_delivery_retries_total" + mDuration = "webhooker_delivery_duration_seconds" + mPending = "webhooker_deliveries_pending" + mRetrying = "webhooker_deliveries_retrying" + mBreakers = "webhooker_circuit_breakers_open" +) + +const ( + mTypeHTTP = "http" + mTypeLog = "log" + mTypeUnknown = "unknown" +) + +// mIsolate gives the setup's engine a metric set registered on a +// private registry. The process-wide collectors are moved by every +// other delivery test running in parallel, so exact assertions are +// only possible against a registry this test owns. +func mIsolate( + t *testing.T, s iSetup, +) *prometheus.Registry { + t.Helper() + + reg := prometheus.NewRegistry() + s.Engine.ExportSetMetrics(metrics.New(reg)) + + return reg +} + +// mFind returns the series of the named metric carrying the given +// target_type label. +func mFind( + t *testing.T, + reg *prometheus.Registry, + name, targetType string, +) *dto.Metric { + t.Helper() + + families, err := reg.Gather() + require.NoError(t, err) + + for _, fam := range families { + if fam.GetName() != name { + continue + } + + for _, m := range fam.GetMetric() { + for _, label := range m.GetLabel() { + if label.GetName() == "target_type" && + label.GetValue() == targetType { + return m + } + } + } + } + + t.Fatalf( + "metric %s{target_type=%q} not found", + name, targetType, + ) + + return nil +} + +func mCounter( + t *testing.T, + reg *prometheus.Registry, + name, targetType string, +) float64 { + t.Helper() + + return mFind(t, reg, name, targetType). + GetCounter().GetValue() +} + +func mGauge( + t *testing.T, + reg *prometheus.Registry, + name, targetType string, +) float64 { + t.Helper() + + return mFind(t, reg, name, targetType). + GetGauge().GetValue() +} + +// mHTTPDurations returns how many samples the delivery duration +// histogram holds for the http target type, which is the type every +// test here times. +func mHTTPDurations( + t *testing.T, reg *prometheus.Registry, +) uint64 { + t.Helper() + + return mFind(t, reg, mDuration, mTypeHTTP). + GetHistogram().GetSampleCount() +} + +// TestDeliveryMetrics_SuccessAndRetryExhaustion drives one delivery +// that succeeds and one that fails every attempt until its retries +// are exhausted, and asserts every delivery counter across both. +func TestDeliveryMetrics_SuccessAndRetryExhaustion( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + reg := mIsolate(t, s) + + mDeliverOK(t, s) + + assert.InDelta(t, 1.0, + mCounter(t, reg, mAttempts, mTypeHTTP), 0) + assert.InDelta(t, 1.0, + mCounter(t, reg, mSucceeded, mTypeHTTP), 0) + assert.InDelta(t, 0.0, + mCounter(t, reg, mFailed, mTypeHTTP), 0) + assert.InDelta(t, 0.0, + mCounter(t, reg, mRetries, mTypeHTTP), 0) + assert.Equal(t, uint64(1), + mHTTPDurations(t, reg)) + + mExhaustRetries(t, s) + + // Two further attempts: the first is retried, the second is + // the last one allowed and fails the delivery terminally. + assert.InDelta(t, 3.0, + mCounter(t, reg, mAttempts, mTypeHTTP), 0) + assert.InDelta(t, 1.0, + mCounter(t, reg, mSucceeded, mTypeHTTP), 0) + assert.InDelta(t, 1.0, + mCounter(t, reg, mRetries, mTypeHTTP), 0) + assert.InDelta(t, 1.0, + mCounter(t, reg, mFailed, mTypeHTTP), 0) + assert.Equal(t, uint64(3), + mHTTPDurations(t, reg)) + + // Two consecutive failures are below the trip threshold. + assert.InDelta(t, 0.0, + mGauge(t, reg, mBreakers, mTypeHTTP), 0) + + // The label is the target type and nothing finer: two http + // targets shared one series, and no other type's moved. + assert.InDelta(t, 0.0, + mCounter(t, reg, mAttempts, mTypeLog), 0) + assert.InDelta(t, 0.0, + mCounter(t, reg, mFailed, mTypeLog), 0) +} + +// mDeliverOK delivers one event to a target that answers 200. +func mDeliverOK(t *testing.T, s iSetup) { + t.Helper() + + ts := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }, + )) + defer ts.Close() + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"ok":true}`, + ) + targetID := uuid.New().String() + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + body := event.Body + task := iTask( + d, event, s.WebhookID, targetID, + "metrics-ok", iHTTPConfig(ts.URL), 3, 1, &body, + ) + + s.Engine.ExportProcessNewTask(context.TODO(), &task) + + iAssertStatus(t, s.WebhookDB, d.ID, + database.DeliveryStatusDelivered, + ) +} + +// mExhaustRetries delivers to a target that answers 500 with a +// two-attempt budget, driving both attempts so the delivery ends +// terminally failed. +func mExhaustRetries(t *testing.T, s iSetup) { + t.Helper() + + ts := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + }, + )) + defer ts.Close() + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"ok":false}`, + ) + targetID := uuid.New().String() + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + body := event.Body + cfg := iHTTPConfig(ts.URL) + + first := iTask( + d, event, s.WebhookID, targetID, + "metrics-fail", cfg, 2, 1, &body, + ) + + s.Engine.ExportProcessNewTask(context.TODO(), &first) + + iAssertStatus(t, s.WebhookDB, d.ID, + database.DeliveryStatusRetrying, + ) + + // The engine's own scheduler would re-enqueue this after the + // backoff; driving the second attempt directly keeps the test + // deterministic and off the wall clock. + second := iTask( + d, event, s.WebhookID, targetID, + "metrics-fail", cfg, 2, 2, &body, + ) + + s.Engine.ExportProcessRetryTask( + context.TODO(), &second, + ) + + iAssertStatus(t, s.WebhookDB, d.ID, + database.DeliveryStatusFailed, + ) +} + +// TestDeliveryMetrics_CircuitBreakerGauge proves the open-breaker +// gauge follows a breaker that trips. +func TestDeliveryMetrics_CircuitBreakerGauge(t *testing.T) { + t.Parallel() + + s := newISetup(t) + reg := mIsolate(t, s) + + ts := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + }, + )) + defer ts.Close() + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"trip":true}`, + ) + targetID := uuid.New().String() + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + body := event.Body + cfg := iHTTPConfig(ts.URL) + + // A retry budget above the failure threshold, so the breaker + // rather than the budget is what stops the delivery. + maxRetries := delivery.ExportDefaultFailureThreshold + 5 + + first := iTask( + d, event, s.WebhookID, targetID, + "metrics-trip", cfg, maxRetries, 1, &body, + ) + + s.Engine.ExportProcessNewTask(context.TODO(), &first) + + assert.InDelta(t, 0.0, + mGauge(t, reg, mBreakers, mTypeHTTP), 0) + + for attempt := 2; attempt <= delivery. + ExportDefaultFailureThreshold; attempt++ { + task := iTask( + d, event, s.WebhookID, targetID, + "metrics-trip", cfg, maxRetries, attempt, &body, + ) + + s.Engine.ExportProcessRetryTask( + context.TODO(), &task, + ) + } + + assert.InDelta(t, 1.0, + mGauge(t, reg, mBreakers, mTypeHTTP), 0) +} + +// TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt proves a delivery +// an open circuit breaker refuses is neither counted as an attempt +// nor observed in the duration histogram. +// +// It sends nothing and records no result row, so counting it would +// climb the attempts counter with no traffic behind it and pull the +// duration quantiles down with near-zero samples for as long as the +// breaker stayed open — the metric moving the wrong way during the +// outage it exists to reveal. +func TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + reg := mIsolate(t, s) + + ts := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + }, + )) + defer ts.Close() + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"blocked":true}`, + ) + targetID := uuid.New().String() + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + body := event.Body + cfg := iHTTPConfig(ts.URL) + maxRetries := delivery.ExportDefaultFailureThreshold + 5 + + first := iTask( + d, event, s.WebhookID, targetID, + "metrics-blocked", cfg, maxRetries, 1, &body, + ) + + s.Engine.ExportProcessNewTask(context.TODO(), &first) + + for attempt := 2; attempt <= delivery. + ExportDefaultFailureThreshold; attempt++ { + task := iTask( + d, event, s.WebhookID, targetID, + "metrics-blocked", cfg, maxRetries, attempt, &body, + ) + + s.Engine.ExportProcessRetryTask(context.TODO(), &task) + } + + require.InDelta(t, 1.0, + mGauge(t, reg, mBreakers, mTypeHTTP), 0, + "breaker should be open before the blocked attempt") + + threshold := float64( + delivery.ExportDefaultFailureThreshold, + ) + + assert.InDelta(t, threshold, + mCounter(t, reg, mAttempts, mTypeHTTP), 0) + assert.Equal(t, uint64(threshold), + mHTTPDurations(t, reg)) + + retriesBefore := mCounter(t, reg, mRetries, mTypeHTTP) + + blocked := iTask( + d, event, s.WebhookID, targetID, + "metrics-blocked", cfg, maxRetries, + delivery.ExportDefaultFailureThreshold+1, &body, + ) + + s.Engine.ExportProcessRetryTask(context.TODO(), &blocked) + + // The breaker refused it: rescheduled, so the retry counter + // moved, but nothing was attempted or timed. + assert.InDelta(t, retriesBefore+1, + mCounter(t, reg, mRetries, mTypeHTTP), 0) + assert.InDelta(t, threshold, + mCounter(t, reg, mAttempts, mTypeHTTP), 0) + assert.Equal(t, uint64(threshold), + mHTTPDurations(t, reg)) +} + +// TestDeliveryMetrics_OrphanedRetryFailureLabelled proves the +// terminal failure of a delivery whose target no longer retries is +// counted against the target's real type, not against unknown. The +// type is threaded in as an argument because populating d.Target on +// that path would write the target row into the per-webhook database +// (https://git.eeqj.de/sneak/webhooker/issues/206). +func TestDeliveryMetrics_OrphanedRetryFailureLabelled( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + reg := mIsolate(t, s) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "orphaned-label", + ) + + deliveryID := iSeedRetryingWithType( + t, s, database.TargetTypeLog, + ) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + iAssertStatus(t, s.WebhookDB, deliveryID, + database.DeliveryStatusFailed, + ) + + assert.InDelta(t, 1.0, + mCounter(t, reg, mFailed, mTypeLog), 0) +} + +// TestDeliveryMetrics_QueueDepthGauges proves the sampler publishes +// the queued deliveries it finds in the per-webhook databases, and +// that a drained queue reads zero rather than keeping its last +// value. +func TestDeliveryMetrics_QueueDepthGauges(t *testing.T) { + t.Parallel() + + s := newISetup(t) + reg := mIsolate(t, s) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "queue-depth", + ) + + targetID := uuid.New().String() + + iCreateTarget(t, s.MainDB, targetID, s.WebhookID, + "queue-depth-target", database.TargetTypeHTTP, + iHTTPConfig("https://example.com/hook"), 3, + ) + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"queued":true}`, + ) + + pending := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + retrying := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusRetrying, + ) + + s.Engine.ExportSampleQueueDepths(context.Background()) + + assert.InDelta(t, 2.0, + mGauge(t, reg, mPending, mTypeHTTP), 0) + assert.InDelta(t, 1.0, + mGauge(t, reg, mRetrying, mTypeHTTP), 0) + assert.InDelta(t, 0.0, + mGauge(t, reg, mPending, mTypeLog), 0) + + require.NoError(t, s.WebhookDB. + Model(&database.Delivery{}). + Where("id IN ?", []string{pending.ID, retrying.ID}). + Update( + "status", database.DeliveryStatusDelivered, + ).Error) + + s.Engine.ExportSampleQueueDepths(context.Background()) + + assert.InDelta(t, 1.0, + mGauge(t, reg, mPending, mTypeHTTP), 0) + assert.InDelta(t, 0.0, + mGauge(t, reg, mRetrying, mTypeHTTP), 0) +} + +// TestDeliveryMetrics_QueueDepthDeletedTarget proves a backlog queued +// against a target that has since been deleted stays visible, in the +// unknown series, instead of being dropped. That backlog is the one +// nobody is watching, so losing it would defeat the queue-depth +// alerting this metric exists for. +func TestDeliveryMetrics_QueueDepthDeletedTarget(t *testing.T) { + t.Parallel() + + s := newISetup(t) + reg := mIsolate(t, s) + + iCreateWebhook( + t, s.MainDB, s.WebhookID, "deleted-target", + ) + + // No target row is created: this is a delivery whose target was + // deleted out from under it. + targetID := uuid.New().String() + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"orphan":true}`, + ) + + iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusRetrying, + ) + + s.Engine.ExportSampleQueueDepths(context.Background()) + + assert.InDelta(t, 1.0, + mGauge(t, reg, mPending, mTypeUnknown), 0) + assert.InDelta(t, 1.0, + mGauge(t, reg, mRetrying, mTypeUnknown), 0) + assert.InDelta(t, 0.0, + mGauge(t, reg, mPending, mTypeHTTP), 0) +} diff --git a/internal/delivery/queue_depth.go b/internal/delivery/queue_depth.go new file mode 100644 index 0000000..760a5a6 --- /dev/null +++ b/internal/delivery/queue_depth.go @@ -0,0 +1,187 @@ +package delivery + +import ( + "context" + "fmt" + "time" + + "sneak.berlin/go/webhooker/internal/database" +) + +// queueDepthSampleInterval is how often the pending and retrying +// queue depths are counted and published as gauges. +const queueDepthSampleInterval = 30 * time.Second + +// queueDepthSampler publishes the pending and retrying queue depths +// on a timer for as long as the engine runs. +// +// The depths are counted out of the databases rather than tracked as +// deltas alongside the status transitions. A delta counter would have +// to be seeded correctly at startup from rows written by a previous +// process, and would drift permanently on any transition that failed +// to persist. Counting is the measurement that cannot go wrong, and +// it is the same whole-database walk the retry sweep already makes. +func (e *Engine) queueDepthSampler(ctx context.Context) { + defer e.wg.Done() + + ticker := time.NewTicker(queueDepthSampleInterval) + defer ticker.Stop() + + e.sampleQueueDepths(ctx) + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + e.sampleQueueDepths(ctx) + } + } +} + +// sampleQueueDepths counts every queued delivery across all +// per-webhook databases and publishes the result. +func (e *Engine) sampleQueueDepths(ctx context.Context) { + if e.database == nil || e.dbManager == nil { + return + } + + types, err := e.targetTypesByID() + if err != nil { + e.log.Error( + "queue depth sample: failed to load target types", + "error", err, + ) + + return + } + + var webhookIDs []string + + err = e.database.DB(). + Model(&database.Webhook{}). + Pluck("id", &webhookIDs).Error + if err != nil { + e.log.Error( + "queue depth sample: failed to query webhook IDs", + "error", err, + ) + + return + } + + pending := make(map[database.TargetType]int) + retrying := make(map[database.TargetType]int) + + for _, webhookID := range webhookIDs { + select { + case <-ctx.Done(): + return + default: + } + + if !e.dbManager.DBExists(webhookID) { + continue + } + + e.sampleWebhookQueueDepths( + webhookID, types, pending, retrying, + ) + } + + e.mtr.SetQueueDepths(pending, retrying) +} + +// targetTypesByID maps every configured target id to its type. The +// deliveries live in the per-webhook databases but carry only a +// target id, so the type label has to come from the main database. +func (e *Engine) targetTypesByID() ( + map[string]database.TargetType, error, +) { + var rows []struct { + ID string + Type database.TargetType + } + + err := e.database.DB(). + Model(&database.Target{}). + Select("id", "type"). + Scan(&rows).Error + if err != nil { + return nil, fmt.Errorf("loading targets: %w", err) + } + + types := make(map[string]database.TargetType, len(rows)) + + for _, row := range rows { + types[row.ID] = row.Type + } + + return types, nil +} + +// sampleWebhookQueueDepths adds one webhook's queued deliveries into +// the running totals. +// +// A delivery whose target has since been deleted is not in the type +// map and so counts under the empty target type. Set.SetQueueDepths +// folds that into the unknown series rather than dropping it: a +// backlog stuck behind a deleted target is a backlog that still needs +// to be alertable. +func (e *Engine) sampleWebhookQueueDepths( + webhookID string, + types map[string]database.TargetType, + pending, retrying map[database.TargetType]int, +) { + webhookDB, err := e.dbManager.GetDB(webhookID) + if err != nil { + e.log.Error( + "queue depth sample: failed to get webhook database", + "webhook_id", webhookID, + "error", err, + ) + + return + } + + var rows []struct { + TargetID string + Status database.DeliveryStatus + Depth int + } + + err = webhookDB. + Model(&database.Delivery{}). + Select("target_id", "status", "count(*) as depth"). + Where("status IN ?", []database.DeliveryStatus{ + database.DeliveryStatusPending, + database.DeliveryStatusRetrying, + }). + Group("target_id, status"). + Scan(&rows).Error + if err != nil { + e.log.Error( + "queue depth sample: "+ + "failed to count queued deliveries", + "webhook_id", webhookID, + "error", err, + ) + + return + } + + for _, row := range rows { + targetType := types[row.TargetID] + + switch row.Status { + case database.DeliveryStatusPending: + pending[targetType] += row.Depth + case database.DeliveryStatusRetrying: + retrying[targetType] += row.Depth + case database.DeliveryStatusDelivered, + database.DeliveryStatusFailed: + // Excluded by the query above: a delivery that has + // reached a terminal state is not queued. + } + } +} diff --git a/internal/delivery/target.go b/internal/delivery/target.go index 9e131b2..17c3bf5 100644 --- a/internal/delivery/target.go +++ b/internal/delivery/target.go @@ -27,6 +27,12 @@ type Scheduler interface { // own circuit breaker, and reschedules via the injected // Scheduler. Fire-and-forget targets simply record a single // attempt. +// +// An implementation reports each attempt it actually dispatches to +// Engine.observeAttempt, alongside the DeliveryResult it records for +// it. Deliver is also entered for attempts that never happen — an +// open circuit breaker refuses one — so the count cannot be taken +// from around this call. type Target interface { Deliver( ctx context.Context, @@ -74,6 +80,12 @@ type attemptResult struct { errMsg string } +// elapsed returns how long the attempt took. The field is stored in +// milliseconds because that is what DeliveryResult persists. +func (r attemptResult) elapsed() time.Duration { + return time.Duration(r.duration) * time.Millisecond +} + // initTargets builds the target registry, wiring each target // to the engine's persistence helpers and giving the HTTP and // Slack targets the shared SSRF-safe client. It is called by diff --git a/internal/delivery/target_database.go b/internal/delivery/target_database.go index db29520..04ccaf6 100644 --- a/internal/delivery/target_database.go +++ b/internal/delivery/target_database.go @@ -42,7 +42,14 @@ func (t *databaseTarget) Deliver( _ *Task, _ Scheduler, ) { + start := time.Now() + err := t.archive(d) + + elapsed := time.Since(start) + + t.eng.observeAttempt(d.Target.Type, elapsed) + if err != nil { t.eng.log.Error( "failed to archive event to database target", @@ -53,22 +60,25 @@ func (t *databaseTarget) Deliver( t.eng.recordResult( webhookDB, d, 1, false, 0, "", - err.Error(), 0, + err.Error(), elapsed.Milliseconds(), ) t.eng.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusFailed, + webhookDB, d, d.Target.Type, + database.DeliveryStatusFailed, ) return } t.eng.recordResult( - webhookDB, d, 1, true, 0, "", "", 0, + webhookDB, d, 1, true, 0, "", "", + elapsed.Milliseconds(), ) t.eng.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusDelivered, + webhookDB, d, d.Target.Type, + database.DeliveryStatusDelivered, ) } diff --git a/internal/delivery/target_http.go b/internal/delivery/target_http.go index f39af45..c8d025a 100644 --- a/internal/delivery/target_http.go +++ b/internal/delivery/target_http.go @@ -74,6 +74,8 @@ func (c *httpCore) fireAndForget( d *database.Delivery, res attemptResult, ) { + c.eng.observeAttempt(d.Target.Type, res.elapsed()) + c.eng.recordResult( webhookDB, d, 1, res.success, res.statusCode, res.respBody, res.errMsg, @@ -82,7 +84,7 @@ func (c *httpCore) fireAndForget( if res.success { c.eng.updateDeliveryStatus( - webhookDB, d, + webhookDB, d, d.Target.Type, database.DeliveryStatusDelivered, ) @@ -90,7 +92,8 @@ func (c *httpCore) fireAndForget( } c.eng.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusFailed, + webhookDB, d, d.Target.Type, + database.DeliveryStatusFailed, ) } @@ -107,10 +110,17 @@ func (c *httpCore) withRetry( return } + // Allow may have moved the breaker to half-open, and the + // attempt below may open or close it, so the gauge is + // republished on every exit from here. + defer c.publishCircuitState(d.Target.Type) + attemptNum := task.AttemptNum res := attempt() + c.eng.observeAttempt(d.Target.Type, res.elapsed()) + c.eng.recordResult( webhookDB, d, attemptNum, res.success, res.statusCode, res.respBody, res.errMsg, @@ -121,7 +131,7 @@ func (c *httpCore) withRetry( cb.RecordSuccess() c.eng.updateDeliveryStatus( - webhookDB, d, + webhookDB, d, d.Target.Type, database.DeliveryStatusDelivered, ) @@ -146,6 +156,8 @@ func (c *httpCore) circuitBreakerBlock( return false } + defer c.publishCircuitState(d.Target.Type) + remaining := cb.CooldownRemaining() c.eng.log.Info( @@ -157,7 +169,7 @@ func (c *httpCore) circuitBreakerBlock( ) c.eng.updateDeliveryStatus( - webhookDB, d, + webhookDB, d, d.Target.Type, database.DeliveryStatusRetrying, ) @@ -177,7 +189,7 @@ func (c *httpCore) handleRetry( ) { if attemptNum >= maxRetries { c.eng.updateDeliveryStatus( - webhookDB, d, + webhookDB, d, d.Target.Type, database.DeliveryStatusFailed, ) @@ -185,7 +197,8 @@ func (c *httpCore) handleRetry( } c.eng.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusRetrying, + webhookDB, d, d.Target.Type, + database.DeliveryStatusRetrying, ) backoff := calcBackoff(attemptNum) @@ -215,6 +228,28 @@ func (c *httpCore) getCircuitBreaker( return cb } +// publishCircuitState recounts this core's open breakers and +// publishes the gauge. Each core holds the breakers of exactly one +// target type, so the recount is over that type's targets alone. +// Counting rather than adjusting a delta keeps the gauge honest +// however a breaker changed state. +func (c *httpCore) publishCircuitState( + targetType database.TargetType, +) { + open := 0 + + c.circuitBreakers.Range(func(_, val any) bool { + cb, ok := val.(*CircuitBreaker) + if ok && cb.State() == CircuitOpen { + open++ + } + + return true + }) + + c.eng.mtr.SetCircuitBreakersOpen(targetType, open) +} + // remainingBackoff returns how long remains of the backoff // window for the last attempt of a recovered retrying // delivery. It implements rescheduler. @@ -302,7 +337,8 @@ func (t *httpTarget) Deliver( ) t.eng.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusFailed, + webhookDB, d, d.Target.Type, + database.DeliveryStatusFailed, ) return diff --git a/internal/delivery/target_log.go b/internal/delivery/target_log.go index ce2d690..26aa7e2 100644 --- a/internal/delivery/target_log.go +++ b/internal/delivery/target_log.go @@ -2,6 +2,7 @@ package delivery import ( "context" + "time" "gorm.io/gorm" "sneak.berlin/go/webhooker/internal/database" @@ -34,6 +35,8 @@ func (t *logTarget) Deliver( _ *Task, _ Scheduler, ) { + start := time.Now() + t.eng.log.Info( "webhook event delivered to log target", "delivery_id", d.ID, @@ -48,11 +51,17 @@ func (t *logTarget) Deliver( "body", d.Event.Body, ) + elapsed := time.Since(start) + + t.eng.observeAttempt(d.Target.Type, elapsed) + t.eng.recordResult( - webhookDB, d, 1, true, 0, "", "", 0, + webhookDB, d, 1, true, 0, "", "", + elapsed.Milliseconds(), ) t.eng.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusDelivered, + webhookDB, d, d.Target.Type, + database.DeliveryStatusDelivered, ) } diff --git a/internal/delivery/target_slack.go b/internal/delivery/target_slack.go index 5f98359..3f855d1 100644 --- a/internal/delivery/target_slack.go +++ b/internal/delivery/target_slack.go @@ -101,7 +101,8 @@ func (t *slackTarget) failConfig( ) t.eng.updateDeliveryStatus( - webhookDB, d, database.DeliveryStatusFailed, + webhookDB, d, d.Target.Type, + database.DeliveryStatusFailed, ) } diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 8d451bc..9100867 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -18,6 +18,7 @@ import ( "sneak.berlin/go/webhooker/internal/globals" "sneak.berlin/go/webhooker/internal/healthcheck" "sneak.berlin/go/webhooker/internal/logger" + "sneak.berlin/go/webhooker/internal/metrics" "sneak.berlin/go/webhooker/internal/middleware" "sneak.berlin/go/webhooker/internal/session" "sneak.berlin/go/webhooker/templates" @@ -73,6 +74,7 @@ type Handlers struct { mw *middleware.Middleware notifier delivery.Notifier evictor delivery.WebhookEvictor + mtr *metrics.Set templates map[string]*template.Template // dummyVerifications counts the equivalent-cost verifications @@ -114,6 +116,7 @@ func New( s.mw = params.Middleware s.notifier = params.Notifier s.evictor = params.Evictor + s.mtr = metrics.Default() // Parse all page templates once at startup s.templates = map[string]*template.Template{ diff --git a/internal/handlers/webhook.go b/internal/handlers/webhook.go index 1b1543f..920e68b 100644 --- a/internal/handlers/webhook.go +++ b/internal/handlers/webhook.go @@ -217,6 +217,11 @@ func (h *Handlers) createAndDeliverEvent( return } + // Counted here, after the commit: an event is received once it + // is durably stored, which is what the delivery counters are + // compared against on a dashboard. + h.mtr.EventReceived() + h.finishWebhookResponse(w, event, entrypoint, tasks) } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 0000000..b69538f --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,344 @@ +// Package metrics defines the Prometheus collectors describing +// webhooker's delivery pipeline: how many events arrive, how many +// deliveries are attempted, how they end, how long they take, how +// deep the queues are, and how many circuit breakers are open. +// +// The inbound HTTP metrics come from the go-http-metrics recorder in +// internal/middleware and land on prometheus.DefaultRegisterer. These +// collectors register there too, so both surfaces are gathered by the +// one promhttp handler mounted on the authenticated /metrics route. +package metrics + +import ( + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "sneak.berlin/go/webhooker/internal/database" +) + +// namespace prefixes every collector defined here. +const namespace = "webhooker" + +// targetTypeLabel is the only label any delivery metric carries, and +// cardinality is the whole reason for that. +// +// A target type is one of four compile-time constants, so the label +// domain is bounded by construction. Target ids, event ids and +// entrypoint ids are not: they are UUIDs minted per operator action +// or per inbound request, a series is never reclaimed once it exists, +// and labelling by any of them makes /metrics a memory leak that +// grows with traffic. normalizeTargetType enforces the bound at every +// call site — a type the registry does not know collapses into +// unknownTargetType rather than minting a series of its own. +const targetTypeLabel = "target_type" + +// unknownTargetType is the bucket for a target type outside the known +// set, so an unrecognised value cannot mint a new series. +const unknownTargetType = "unknown" + +// Delivery duration buckets, exponential from 5ms so the last bucket +// (about 98s) sits above the 30s outbound HTTP client timeout. +const ( + durationBucketStart = 0.005 + durationBucketFactor = 3 + durationBucketCount = 10 +) + +// knownTargetTypes is the fixed label domain: the target types the +// delivery engine implements. +// +//nolint:gochecknoglobals // the label domain, built once per process +var knownTargetTypes = []database.TargetType{ + database.TargetTypeHTTP, + database.TargetTypeDatabase, + database.TargetTypeLog, + database.TargetTypeSlack, +} + +// defaultSet is the process-wide metric set, registered on the same +// registry the HTTP middleware and the /metrics handler already use. +// It is built on first use rather than in an init so that a test +// binary that never touches metrics never registers them. +// +//nolint:gochecknoglobals // one process-wide registration, by design +var defaultSet = sync.OnceValue(func() *Set { + return New(prometheus.DefaultRegisterer) +}) + +// Default returns the process-wide metric set. +func Default() *Set { + return defaultSet() +} + +// Set is one registered group of webhooker's delivery collectors. +// Production uses the single Default set; tests build their own +// against a private registry so assertions are not disturbed by +// deliveries other tests are making concurrently. +type Set struct { + eventsReceived prometheus.Counter + deliveryAttempts *prometheus.CounterVec + deliveriesSucceeded *prometheus.CounterVec + deliveriesFailed *prometheus.CounterVec + deliveryRetries *prometheus.CounterVec + deliveryDuration *prometheus.HistogramVec + deliveriesPending *prometheus.GaugeVec + deliveriesRetrying *prometheus.GaugeVec + circuitBreakersOpen *prometheus.GaugeVec +} + +// New registers a full set of delivery collectors on reg and returns +// it. It panics if reg already holds them, which is the intended +// behaviour for a duplicate registration. +func New(reg prometheus.Registerer) *Set { + factory := promauto.With(reg) + + s := &Set{ + eventsReceived: factory.NewCounter( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "events_received_total", + Help: "Webhook events received and " + + "stored, so the receive and deliver " + + "sides can be compared.", + }, + ), + deliveryDuration: factory.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: namespace, + Name: "delivery_duration_seconds", + Help: "Wall time of a single delivery " + + "attempt, by target type.", + Buckets: prometheus.ExponentialBuckets( + durationBucketStart, + durationBucketFactor, + durationBucketCount, + ), + }, + []string{targetTypeLabel}, + ), + } + + s.registerCounters(factory) + s.registerGauges(factory) + s.initSeries() + + return s +} + +// EventReceived counts one inbound webhook event stored. +func (s *Set) EventReceived() { + s.eventsReceived.Inc() +} + +// DeliveryAttempted counts one delivery attempt dispatched to a +// target. +func (s *Set) DeliveryAttempted(t database.TargetType) { + s.deliveryAttempts. + WithLabelValues(normalizeTargetType(t)). + Inc() +} + +// ObserveDeliveryDuration records how long one delivery attempt took. +func (s *Set) ObserveDeliveryDuration( + t database.TargetType, d time.Duration, +) { + s.deliveryDuration. + WithLabelValues(normalizeTargetType(t)). + Observe(d.Seconds()) +} + +// DeliveryStatusChanged counts a delivery's transition into a new +// status. The mapping from status to counter lives here, next to the +// collectors, so the engine has a single call for every transition it +// persists. A move back to pending is not an outcome and counts +// nothing. +func (s *Set) DeliveryStatusChanged( + t database.TargetType, status database.DeliveryStatus, +) { + label := normalizeTargetType(t) + + switch status { + case database.DeliveryStatusDelivered: + s.deliveriesSucceeded.WithLabelValues(label).Inc() + case database.DeliveryStatusFailed: + s.deliveriesFailed.WithLabelValues(label).Inc() + case database.DeliveryStatusRetrying: + s.deliveryRetries.WithLabelValues(label).Inc() + case database.DeliveryStatusPending: + } +} + +// SetQueueDepths publishes the pending and retrying queue depths from +// one sample. Every label in the queue domain is written on every +// call, so a type whose queue has drained reads zero instead of +// holding its last value forever. +func (s *Set) SetQueueDepths( + pending, retrying map[database.TargetType]int, +) { + pendingByLabel := foldToLabels(pending) + retryingByLabel := foldToLabels(retrying) + + for _, label := range queueDepthLabels() { + s.deliveriesPending.WithLabelValues(label). + Set(float64(pendingByLabel[label])) + s.deliveriesRetrying.WithLabelValues(label). + Set(float64(retryingByLabel[label])) + } +} + +// queueDepthLabels is the label domain of the two queue-depth gauges: +// the known target types plus unknown. +// +// Unknown is a real bucket here, not a safety net. A delivery queued +// against a target that has since been deleted carries a target id no +// longer in the targets table, so the sample resolves it to the empty +// type; folding it into unknown is what keeps that backlog visible. +// Dropping it would hide the one queue nobody is watching. +func queueDepthLabels() []string { + labels := make([]string, 0, len(knownTargetTypes)+1) + + for _, t := range knownTargetTypes { + labels = append(labels, string(t)) + } + + return append(labels, unknownTargetType) +} + +// foldToLabels collapses a per-target-type count onto the bounded +// label domain, summing everything outside the known set into +// unknown. +func foldToLabels( + counts map[database.TargetType]int, +) map[string]int { + byLabel := make(map[string]int, len(counts)) + + for t, n := range counts { + byLabel[normalizeTargetType(t)] += n + } + + return byLabel +} + +// SetCircuitBreakersOpen publishes how many of a target type's +// circuit breakers are currently open. +func (s *Set) SetCircuitBreakersOpen( + t database.TargetType, open int, +) { + s.circuitBreakersOpen. + WithLabelValues(normalizeTargetType(t)). + Set(float64(open)) +} + +func (s *Set) registerCounters(factory promauto.Factory) { + s.deliveryAttempts = factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "delivery_attempts_total", + Help: "Delivery attempts dispatched to a " + + "target, by target type.", + }, + []string{targetTypeLabel}, + ) + + s.deliveriesSucceeded = factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "deliveries_succeeded_total", + Help: "Deliveries that reached the delivered " + + "state, by target type.", + }, + []string{targetTypeLabel}, + ) + + s.deliveriesFailed = factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "deliveries_failed_total", + Help: "Deliveries that failed terminally and " + + "will not be retried, by target type.", + }, + []string{targetTypeLabel}, + ) + + s.deliveryRetries = factory.NewCounterVec( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "delivery_retries_total", + Help: "Deliveries put back into the retrying " + + "state, by target type.", + }, + []string{targetTypeLabel}, + ) +} + +func (s *Set) registerGauges(factory promauto.Factory) { + s.deliveriesPending = factory.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Name: "deliveries_pending", + Help: "Deliveries currently in the pending " + + "state, by target type.", + }, + []string{targetTypeLabel}, + ) + + s.deliveriesRetrying = factory.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Name: "deliveries_retrying", + Help: "Deliveries currently in the retrying " + + "state, by target type.", + }, + []string{targetTypeLabel}, + ) + + s.circuitBreakersOpen = factory.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: namespace, + Name: "circuit_breakers_open", + Help: "Delivery circuit breakers currently " + + "open, by target type.", + }, + []string{targetTypeLabel}, + ) +} + +// initSeries materialises every known-target-type series at zero, so +// a dashboard and an alert rule see a target type that has not +// delivered yet rather than a missing series. +// +// The queue-depth gauges additionally get their unknown series, which +// holds deliveries queued against a deleted target. That backlog can +// predate the process — it is read out of the databases, not counted +// from transitions — so its series has to exist from the first scrape +// rather than appearing only once a backlog has already built up. +func (s *Set) initSeries() { + for _, t := range knownTargetTypes { + label := string(t) + + s.deliveryAttempts.WithLabelValues(label) + s.deliveriesSucceeded.WithLabelValues(label) + s.deliveriesFailed.WithLabelValues(label) + s.deliveryRetries.WithLabelValues(label) + s.deliveriesPending.WithLabelValues(label) + s.deliveriesRetrying.WithLabelValues(label) + s.circuitBreakersOpen.WithLabelValues(label) + } + + s.deliveriesPending.WithLabelValues(unknownTargetType) + s.deliveriesRetrying.WithLabelValues(unknownTargetType) +} + +// normalizeTargetType maps a target type onto the bounded label +// domain, collapsing anything outside it to unknownTargetType. +func normalizeTargetType(t database.TargetType) string { + for _, known := range knownTargetTypes { + if t == known { + return string(known) + } + } + + return unknownTargetType +} diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go new file mode 100644 index 0000000..a8b3a64 --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -0,0 +1,285 @@ +package metrics_test + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/metrics" +) + +// knownLabels is the target_type label domain built from the target +// types the delivery engine implements. +func knownLabels() []string { + return []string{"http", "database", "log", "slack"} +} + +// labelValues returns the target_type label values a metric family +// currently carries. +func labelValues( + t *testing.T, reg *prometheus.Registry, name string, +) []string { + t.Helper() + + families, err := reg.Gather() + require.NoError(t, err) + + var values []string + + for _, fam := range families { + if fam.GetName() != name { + continue + } + + for _, m := range fam.GetMetric() { + for _, label := range m.GetLabel() { + if label.GetName() == "target_type" { + values = append( + values, label.GetValue(), + ) + } + } + } + } + + return values +} + +func gaugeValue( + t *testing.T, + reg *prometheus.Registry, + name, targetType string, +) float64 { + t.Helper() + + families, err := reg.Gather() + require.NoError(t, err) + + for _, fam := range families { + if fam.GetName() != name { + continue + } + + for _, m := range fam.GetMetric() { + if hasTargetType(m, targetType) { + return m.GetGauge().GetValue() + } + } + } + + t.Fatalf( + "gauge %s{target_type=%q} not found", + name, targetType, + ) + + return 0 +} + +func hasTargetType(m *dto.Metric, targetType string) bool { + for _, label := range m.GetLabel() { + if label.GetName() == "target_type" && + label.GetValue() == targetType { + return true + } + } + + return false +} + +// TestUnknownTargetTypeCollapses is the cardinality guard: a target +// type outside the known set must not mint a series of its own, or +// /metrics grows without bound. +func TestUnknownTargetTypeCollapses(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + set := metrics.New(reg) + + for _, bogus := range []string{ + "a1b2c3d4-0000-0000-0000-000000000001", + "a1b2c3d4-0000-0000-0000-000000000002", + "webhook-forwarder", + } { + set.DeliveryAttempted(database.TargetType(bogus)) + } + + values := labelValues( + t, reg, "webhooker_delivery_attempts_total", + ) + + assert.ElementsMatch(t, + append(knownLabels(), "unknown"), + values, + ) +} + +// TestSetQueueDepthsZeroesDrainedTypes proves a queue that has +// drained reads zero rather than holding its last sample forever. +func TestSetQueueDepthsZeroesDrainedTypes(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + set := metrics.New(reg) + + set.SetQueueDepths( + map[database.TargetType]int{ + database.TargetTypeHTTP: 7, + }, + map[database.TargetType]int{ + database.TargetTypeSlack: 2, + }, + ) + + assert.InDelta(t, 7.0, gaugeValue( + t, reg, "webhooker_deliveries_pending", "http", + ), 0) + assert.InDelta(t, 2.0, gaugeValue( + t, reg, "webhooker_deliveries_retrying", "slack", + ), 0) + + set.SetQueueDepths( + map[database.TargetType]int{}, + map[database.TargetType]int{}, + ) + + assert.InDelta(t, 0.0, gaugeValue( + t, reg, "webhooker_deliveries_pending", "http", + ), 0) + assert.InDelta(t, 0.0, gaugeValue( + t, reg, "webhooker_deliveries_retrying", "slack", + ), 0) +} + +// TestKnownSeriesExistBeforeAnyDelivery proves every known target +// type is published at zero from registration, so an alert rule does +// not have to cope with a missing series. +func TestKnownSeriesExistBeforeAnyDelivery(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + metrics.New(reg) + + for _, name := range []string{ + "webhooker_delivery_attempts_total", + "webhooker_deliveries_succeeded_total", + "webhooker_deliveries_failed_total", + "webhooker_delivery_retries_total", + "webhooker_circuit_breakers_open", + } { + assert.ElementsMatch(t, + knownLabels(), + labelValues(t, reg, name), + "metric %s", name, + ) + } + + // The queue gauges additionally publish unknown from + // registration: a backlog queued against a deleted target lands + // there, and it can predate the process, so the series has to + // exist before the first sample rather than appearing only once + // something is already stuck. + for _, name := range []string{ + "webhooker_deliveries_pending", + "webhooker_deliveries_retrying", + } { + assert.ElementsMatch(t, + append(knownLabels(), "unknown"), + labelValues(t, reg, name), + "metric %s", name, + ) + } +} + +// TestSetQueueDepthsFoldsUnknownTypes proves a queued delivery whose +// target type is not a known one — a target deleted out from under it +// resolves to the empty type — is summed into the unknown series +// instead of being dropped, and that the fold is a sum rather than a +// last-writer-wins. +func TestSetQueueDepthsFoldsUnknownTypes(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + set := metrics.New(reg) + + set.SetQueueDepths( + map[database.TargetType]int{ + database.TargetTypeHTTP: 1, + database.TargetType(""): 4, + database.TargetType("retired-type"): 3, + }, + map[database.TargetType]int{ + database.TargetType(""): 2, + }, + ) + + assert.InDelta(t, 7.0, gaugeValue( + t, reg, "webhooker_deliveries_pending", "unknown", + ), 0) + assert.InDelta(t, 2.0, gaugeValue( + t, reg, "webhooker_deliveries_retrying", "unknown", + ), 0) + assert.InDelta(t, 1.0, gaugeValue( + t, reg, "webhooker_deliveries_pending", "http", + ), 0) + + set.SetQueueDepths( + map[database.TargetType]int{}, + map[database.TargetType]int{}, + ) + + assert.InDelta(t, 0.0, gaugeValue( + t, reg, "webhooker_deliveries_pending", "unknown", + ), 0) +} + +// TestDeliveryStatusChangedCounts maps each persisted status onto the +// counter it moves. +func TestDeliveryStatusChangedCounts(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + set := metrics.New(reg) + + set.DeliveryStatusChanged( + database.TargetTypeLog, + database.DeliveryStatusDelivered, + ) + set.DeliveryStatusChanged( + database.TargetTypeLog, + database.DeliveryStatusFailed, + ) + set.DeliveryStatusChanged( + database.TargetTypeLog, + database.DeliveryStatusRetrying, + ) + set.DeliveryStatusChanged( + database.TargetTypeLog, + database.DeliveryStatusPending, + ) + + families, err := reg.Gather() + require.NoError(t, err) + + counts := map[string]float64{} + + for _, fam := range families { + for _, m := range fam.GetMetric() { + if hasTargetType(m, "log") { + counts[fam.GetName()] = + m.GetCounter().GetValue() + } + } + } + + assert.InDelta(t, 1.0, + counts["webhooker_deliveries_succeeded_total"], 0) + assert.InDelta(t, 1.0, + counts["webhooker_deliveries_failed_total"], 0) + assert.InDelta(t, 1.0, + counts["webhooker_delivery_retries_total"], 0) + assert.InDelta(t, 0.0, + counts["webhooker_delivery_attempts_total"], 0) +}