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