1package daemon23import (4 "context"5 "sync/atomic"6 "time"78 "github.com/prometheus/client_golang/prometheus"9)1011var queueMetricKinds = []string{"delivery", "push", "replication", "webhook", "job", "records-projection", "site-mirror", "git-metadata", "catalog-owner", "callback", "delivery-discovery", "notification-delivery", "notification-broadcast", "notification", "callback-delivery", "other"}1213type queueCollector struct {14 app *App15 pending *prometheus.Desc16 errors *prometheus.Desc17 failures atomic.Uint6418}1920// QueueCollector returns one process-wide collector aggregating all currently21// loaded tenants without exposing tenant identifiers as metric labels.22func (a *App) QueueCollector() prometheus.Collector {23 return &queueCollector{24 app: a,25 pending: prometheus.NewDesc("relay_work_pending", "Pending durable work intents across loaded tenants.", []string{"kind"}, nil),26 errors: prometheus.NewDesc("relay_work_collection_errors_total", "Failed durable work queue tenant scrapes.", nil, nil),27 }28}2930func (c *queueCollector) Describe(ch chan<- *prometheus.Desc) {31 ch <- c.pending32 ch <- c.errors33}3435func (c *queueCollector) Collect(ch chan<- prometheus.Metric) {36 counts := make(map[string]float64, len(queueMetricKinds))37 for _, kind := range queueMetricKinds {38 counts[kind] = 039 }40 loaded := c.tenants()41 ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)42 defer cancel()43 var failures uint6444 for _, tenant := range loaded {45 rows, err := tenant.store.DB().QueryContext(ctx, `SELECT kind,count(*) FROM work_intents WHERE state='pending' GROUP BY kind`)46 if err != nil {47 failures++48 continue49 }50 for rows.Next() {51 var kind string52 var count float6453 if err := rows.Scan(&kind, &count); err != nil {54 failures++55 break56 }57 counts[queueKind(kind)] += count58 }59 if err := rows.Err(); err != nil {60 failures++61 }62 _ = rows.Close()63 }64 c.failures.Add(failures)65 for _, kind := range queueMetricKinds {66 ch <- prometheus.MustNewConstMetric(c.pending, prometheus.GaugeValue, counts[kind], kind)67 }68 ch <- prometheus.MustNewConstMetric(c.errors, prometheus.CounterValue, float64(c.failures.Load()))69}7071func (c *queueCollector) tenants() []*Tenant {72 c.app.mu.Lock()73 defer c.app.mu.Unlock()74 loaded := make([]*Tenant, 0, len(c.app.tenants))75 for _, tenant := range c.app.tenants {76 loaded = append(loaded, tenant)77 }78 return loaded79}8081func queueKind(kind string) string {82 for _, known := range queueMetricKinds[:len(queueMetricKinds)-1] {83 if kind == known {84 return known85 }86 }87 return "other"88}8990var _ prometheus.Collector = (*queueCollector)(nil)