From b8c8b75e04674ce2afd5397a62734eeae4dda11e Mon Sep 17 00:00:00 2001 From: sneak Date: Thu, 20 Aug 2026 04:19:12 +0000 Subject: [PATCH] Expose delivery metrics on /metrics (closes #209) /metrics carried only the inbound HTTP surface, so a destination failing for an hour, a growing retry backlog and a stuck-open circuit breaker were all invisible: the receive side stays healthy in each case because it is. New internal/metrics registers, on the existing default registry that the go-http-metrics recorder and the promhttp handler already share: - webhooker_events_received_total - webhooker_delivery_attempts_total - webhooker_deliveries_succeeded_total - webhooker_deliveries_failed_total - webhooker_delivery_retries_total - webhooker_delivery_duration_seconds - webhooker_deliveries_pending / _retrying - webhooker_circuit_breakers_open The route mounting is untouched. Every delivery metric carries one label, target_type, whose domain is the four target-type constants; anything outside it collapses to "unknown" so no series can be minted from a UUID. Target ids, event ids and entrypoint ids are deliberately not labels. Instrumentation sits at the points every target type already passes through: processDelivery for the attempt counter and the duration histogram, updateDeliveryStatus for the outcome counters. The queue-depth gauges are counted out of the per-webhook databases by a 30s sampler rather than tracked as deltas, which would need seeding at startup and would drift on any transition that failed to persist. The open-breaker gauge is recounted from the target's breaker registry on every state change. --- README.md | 41 ++++ go.mod | 2 +- internal/delivery/engine.go | 38 ++- internal/delivery/export_test.go | 16 ++ internal/delivery/metrics_test.go | 378 ++++++++++++++++++++++++++++++ internal/delivery/queue_depth.go | 183 +++++++++++++++ internal/delivery/target_http.go | 29 +++ internal/handlers/handlers.go | 3 + internal/handlers/webhook.go | 5 + internal/metrics/metrics.go | 301 ++++++++++++++++++++++++ internal/metrics/metrics_test.go | 225 ++++++++++++++++++ 11 files changed, 1219 insertions(+), 2 deletions(-) create mode 100644 internal/delivery/metrics_test.go create mode 100644 internal/delivery/queue_depth.go create mode 100644 internal/metrics/metrics.go create mode 100644 internal/metrics/metrics_test.go diff --git a/README.md b/README.md index 2bb6e23..7a7e2b7 100644 --- a/README.md +++ b/README.md @@ -1122,6 +1122,44 @@ 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 dispatched to a target | +| `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 delivery attempt | +| `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. + ### Rate Limiting Global blanket rate limiting middleware (e.g., a per-IP throttle shared @@ -1747,6 +1785,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/ @@ -1764,6 +1803,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..055ef7e 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 + // mx 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. + mx *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, + mx: 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, @@ -826,6 +838,11 @@ func (e *Engine) failUnretryableRetry( target.Type, ) + // The delivery was loaded without its target relation, so + // attach it: updateDeliveryStatus reads the type to label the + // terminal failure it is about to count. + d.Target = *target + e.recordResult( webhookDB, d, @@ -844,12 +861,21 @@ func (e *Engine) failUnretryableRetry( // processDelivery dispatches a delivery to the target that // owns its type. Unknown target types fail the delivery. +// +// It is also where the attempt counter and the duration histogram +// are recorded, because it is the one point every target type +// passes through on every attempt: a target added later is +// instrumented without touching it, and the duration measured is +// the whole cost of the attempt rather than whatever each target +// happens to time for itself. func (e *Engine) processDelivery( ctx context.Context, webhookDB *gorm.DB, d *database.Delivery, task *Task, ) { + e.mx.DeliveryAttempted(d.Target.Type) + target, ok := e.targets[d.Target.Type] if !ok { e.log.Error( @@ -865,7 +891,13 @@ func (e *Engine) processDelivery( return } + start := time.Now() + target.Deliver(ctx, webhookDB, d, task, e) + + e.mx.ObserveDeliveryDuration( + d.Target.Type, time.Since(start), + ) } // recordResult persists a DeliveryResult row describing a @@ -901,12 +933,16 @@ 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. func (e *Engine) updateDeliveryStatus( webhookDB *gorm.DB, d *database.Delivery, status database.DeliveryStatus, ) { + e.mx.DeliveryStatusChanged(d.Target.Type, status) + err := webhookDB.Model(d). Update("status", status).Error if err != nil { diff --git a/internal/delivery/export_test.go b/internal/delivery/export_test.go index fc3f434..06f53eb 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, + mx: metrics.Default(), } e.initTargets(client) @@ -267,6 +269,7 @@ func NewTestEngineSmallRetry( e := &Engine{ log: log, retryCh: make(chan Task, 1), + mx: metrics.Default(), } e.initTargets(nil) @@ -289,12 +292,25 @@ func NewTestEngineWithDB( deliveryCh: make(chan Task, deliveryChannelSize), retryCh: make(chan Task, retryChannelSize), workers: workers, + mx: 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(mx *metrics.Set) { + e.mx = mx +} + +// 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..181f4bd --- /dev/null +++ b/internal/delivery/metrics_test.go @@ -0,0 +1,378 @@ +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" +) + +// 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() +} + +func mObservations( + t *testing.T, + reg *prometheus.Registry, + name, targetType string, +) uint64 { + t.Helper() + + return mFind(t, reg, name, targetType). + 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), + mObservations(t, reg, mDuration, mTypeHTTP)) + + 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), + mObservations(t, reg, mDuration, mTypeHTTP)) + + // 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_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) +} diff --git a/internal/delivery/queue_depth.go b/internal/delivery/queue_depth.go new file mode 100644 index 0000000..a9176c5 --- /dev/null +++ b/internal/delivery/queue_depth.go @@ -0,0 +1,183 @@ +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.mx.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 +// resolves to the empty type and lands in the unknown bucket rather +// than being dropped. +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_http.go b/internal/delivery/target_http.go index f39af45..9c9e614 100644 --- a/internal/delivery/target_http.go +++ b/internal/delivery/target_http.go @@ -107,6 +107,11 @@ 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() @@ -146,6 +151,8 @@ func (c *httpCore) circuitBreakerBlock( return false } + defer c.publishCircuitState(d.Target.Type) + remaining := cb.CooldownRemaining() c.eng.log.Info( @@ -215,6 +222,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.mx.SetCircuitBreakersOpen(targetType, open) +} + // remainingBackoff returns how long remains of the backoff // window for the last attempt of a recovered retrying // delivery. It implements rescheduler. diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 8d451bc..de7affb 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 + mx *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.mx = 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..a543916 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.mx.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..3078614 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,301 @@ +// 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 known target type 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, +) { + for _, t := range knownTargetTypes { + label := string(t) + + s.deliveriesPending.WithLabelValues(label). + Set(float64(pending[t])) + s.deliveriesRetrying.WithLabelValues(label). + Set(float64(retrying[t])) + } +} + +// 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. +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) + } +} + +// 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..9824d73 --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -0,0 +1,225 @@ +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" +) + +// 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, + []string{ + "http", "database", "log", "slack", "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_deliveries_pending", + "webhooker_deliveries_retrying", + "webhooker_circuit_breakers_open", + } { + assert.ElementsMatch(t, + []string{"http", "database", "log", "slack"}, + labelValues(t, reg, name), + "metric %s", name, + ) + } +} + +// 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) +}