| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync/atomic" |
| 6 | "time" |
| 7 | |
| 8 | "github.com/prometheus/client_golang/prometheus" |
| 9 | ) |
| 10 | |
| 11 | var 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" } |
| 12 | |
| 13 | type queueCollector struct { |
| 14 | app *App |
| 15 | pending *prometheus.Desc |
| 16 | errors *prometheus.Desc |
| 17 | failures atomic.Uint64 |
| 18 | } |
| 19 | |
| 20 | // QueueCollector returns one process-wide collector aggregating all currently |
| 21 | // loaded tenants without exposing tenant identifiers as metric labels. |
| 22 | func (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 | } |
| 29 | |
| 30 | func (c *queueCollector) Describe(ch chan<- *prometheus.Desc) { |
| 31 | ch <- c.pending |
| 32 | ch <- c.errors |
| 33 | } |
| 34 | |
| 35 | func (c *queueCollector) Collect(ch chan<- prometheus.Metric) { |
| 36 | counts := make(map[string]float64, len(queueMetricKinds)) |
| 37 | for _, kind := range queueMetricKinds { |
| 38 | counts[kind] = 0 |
| 39 | } |
| 40 | loaded := c.tenants() |
| 41 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| 42 | defer cancel() |
| 43 | var failures uint64 |
| 44 | 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 | continue |
| 49 | } |
| 50 | for rows.Next() { |
| 51 | var kind string |
| 52 | var count float64 |
| 53 | if err := rows.Scan(&kind, &count); err != nil { |
| 54 | failures++ |
| 55 | break |
| 56 | } |
| 57 | counts[queueKind(kind)] += count |
| 58 | } |
| 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 | } |
| 70 | |
| 71 | func (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 loaded |
| 79 | } |
| 80 | |
| 81 | func queueKind(kind string) string { |
| 82 | for _, known := range queueMetricKinds[:len(queueMetricKinds)-1] { |
| 83 | if kind == known { |
| 84 | return known |
| 85 | } |
| 86 | } |
| 87 | return "other" |
| 88 | } |
| 89 | |
| 90 | var _ prometheus.Collector = (*queueCollector)(nil) |