Expose delivery metrics on /metrics (closes #209) (#224)

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.
+		}
+	}
+}