awa_metrics/
lib.rs

1//! Internal OpenTelemetry metric definitions shared across Awa crates.
2//!
3//! `AwaMetrics` is the single source of truth for Awa's OTel metric names and
4//! attribute sets. It lives in this crate (rather than in `awa-worker`) so
5//! `awa-ui` and `awa-cli` callers can emit the same counters as the worker
6//! without pulling in the dispatcher/runtime crate graph.
7//!
8//! Metrics are published via the global OTel meter provider — callers
9//! configure their exporter (Prometheus, OTLP, etc.) before starting the
10//! client.
11//!
12//! All metrics use the `awa` meter name and follow OpenTelemetry naming and
13//! unit guidance:
14//! - Dot-separated hierarchical namespaces (`awa.job.*`, `awa.dispatch.*`)
15//! - Singular nouns for namespaces (not pluralized)
16//! - Units declared via `.with_unit()` using UCUM notation
17//! - No unit suffix in metric names (exporters append automatically)
18//!
19//! Metric-name string constants live in [`names`] for tests asserting
20//! against an OTLP exporter.
21
22use awa_model::storage::StorageStatus;
23use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter, UpDownCounter};
24use std::time::Duration;
25
26/// Public metric-name constants. Useful for tests asserting against an OTLP
27/// exporter without copy-pasting string literals.
28pub mod names {
29    pub const JOB_INSERTED: &str = "awa.job.inserted";
30    pub const ENQUEUE_BATCH_SIZE: &str = "awa.enqueue.batch_size";
31    pub const ENQUEUE_DURATION: &str = "awa.enqueue.duration";
32    pub const JOB_COMPLETED: &str = "awa.job.completed";
33    pub const JOB_FAILED: &str = "awa.job.failed";
34    pub const JOB_RETRIED: &str = "awa.job.retried";
35    pub const JOB_CANCELLED: &str = "awa.job.cancelled";
36    pub const JOB_CLAIMED: &str = "awa.job.claimed";
37    pub const JOB_DURATION: &str = "awa.job.duration";
38    pub const JOB_IN_FLIGHT: &str = "awa.job.in_flight";
39    pub const JOB_WAIT_DURATION: &str = "awa.job.wait_duration";
40    pub const JOB_WAITING_EXTERNAL: &str = "awa.job.waiting_external";
41    pub const JOB_DLQ_MOVED: &str = "awa.job.dlq_moved";
42    pub const JOB_DLQ_RETRIED: &str = "awa.job.dlq_retried";
43    pub const JOB_DLQ_PURGED: &str = "awa.job.dlq_purged";
44    pub const JOB_DLQ_DEPTH: &str = "awa.job.dlq_depth";
45    pub const QUEUE_DEPTH: &str = "awa.queue.depth";
46    pub const QUEUE_LAG: &str = "awa.queue.lag";
47    pub const QUEUE_INFO: &str = "awa.queue.info";
48    pub const JOB_KIND_INFO: &str = "awa.job_kind.info";
49    pub const DISPATCH_CLAIM_BATCHES: &str = "awa.dispatch.claim_batches";
50    pub const DISPATCH_WAKEUPS: &str = "awa.dispatch.wakeups";
51    pub const DISPATCH_WAKE_TO_CLAIM_DURATION: &str = "awa.dispatch.wake_to_claim_duration";
52    pub const DISPATCH_CAPACITY_AVAILABLE: &str = "awa.dispatch.capacity_available";
53    pub const DISPATCH_EMPTY_CLAIMS: &str = "awa.dispatch.empty_claims";
54    pub const DISPATCH_UNUSED_PERMITS: &str = "awa.dispatch.unused_permits";
55    pub const DISPATCH_RATE_LIMITED: &str = "awa.dispatch.rate_limited";
56    pub const DISPATCH_CLAIM_BATCH_SIZE: &str = "awa.dispatch.claim_batch_size";
57    pub const DISPATCH_CLAIM_DURATION: &str = "awa.dispatch.claim_duration";
58    pub const COMPLETION_FLUSHES: &str = "awa.completion.flushes";
59    pub const COMPLETION_FLUSH_BATCH_SIZE: &str = "awa.completion.flush_batch_size";
60    pub const COMPLETION_FLUSH_DURATION: &str = "awa.completion.flush_duration";
61    pub const HEARTBEAT_BATCHES: &str = "awa.heartbeat.batches";
62    pub const MAINTENANCE_RESCUES: &str = "awa.maintenance.rescues";
63    pub const MAINTENANCE_PROMOTE_BATCHES: &str = "awa.maintenance.promote_batches";
64    pub const MAINTENANCE_PROMOTE_BATCH_SIZE: &str = "awa.maintenance.promote_batch_size";
65    pub const MAINTENANCE_PROMOTE_DURATION: &str = "awa.maintenance.promote_duration";
66    pub const MAINTENANCE_BRANCH_DURATION: &str = "awa.maintenance.branch.duration";
67    pub const MAINTENANCE_BRANCH_OVERRUN: &str = "awa.maintenance.branch.overrun";
68    pub const MAINTENANCE_ROTATE_ATTEMPTS: &str = "awa.maintenance.rotate.attempts";
69    pub const MAINTENANCE_ROTATE_SKIPPED_ROWS: &str = "awa.maintenance.rotate.skipped_rows";
70    pub const MAINTENANCE_PRUNE_ATTEMPTS: &str = "awa.maintenance.prune.attempts";
71    pub const MAINTENANCE_PRUNE_SKIPPED_ROWS: &str = "awa.maintenance.prune.skipped_rows";
72    pub const STORAGE_TRANSITION_READY: &str = "awa.storage.transition_ready";
73    pub const STORAGE_CANONICAL_LIVE_BACKLOG: &str = "awa.storage.canonical_live_backlog";
74    pub const STORAGE_LIVE_RUNTIME_CAPABILITY: &str = "awa.storage.live_runtime_capability";
75    pub const STORAGE_STATE: &str = "awa.storage.state";
76    pub const RING_CURRENT_SLOT: &str = "awa.ring.current_slot";
77    pub const RING_GENERATION: &str = "awa.ring.generation";
78}
79
80const WAIT_DURATION_BUCKETS_SECONDS: [f64; 14] = [
81    0.001, 0.005, 0.010, 0.025, 0.050, 0.100, 0.250, 0.500, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0,
82];
83
84/// Awa worker metrics backed by OpenTelemetry.
85#[derive(Clone)]
86pub struct AwaMetrics {
87    /// Total jobs inserted.
88    pub jobs_inserted: Counter<u64>,
89    /// Per-batch size distribution for the direct queue-storage COPY enqueue
90    /// path (`QueueStorage::enqueue_params_copy` / Python
91    /// `Client.enqueue_many_copy`). Lets producers see how chunky their
92    /// batches actually land — useful when a target enqueue rate isn't being
93    /// hit and you need to disambiguate "batches are tiny" from "batches are
94    /// slow."
95    pub enqueue_batch_size: Histogram<u64>,
96    /// Per-batch wall-clock duration for the direct queue-storage COPY
97    /// enqueue path. Pair with [`enqueue_batch_size`](Self::enqueue_batch_size)
98    /// to read jobs-per-second and per-batch latency in one Grafana row.
99    pub enqueue_duration_seconds: Histogram<f64>,
100    /// Total jobs completed successfully.
101    pub jobs_completed: Counter<u64>,
102    /// Total jobs that failed (terminal).
103    pub jobs_failed: Counter<u64>,
104    /// Total jobs marked retryable.
105    pub jobs_retried: Counter<u64>,
106    /// Total jobs cancelled.
107    pub jobs_cancelled: Counter<u64>,
108    /// Total jobs claimed (dequeued) for execution.
109    pub jobs_claimed: Counter<u64>,
110    /// Number of dispatcher claim queries executed.
111    pub claim_batches: Counter<u64>,
112    /// Number of dispatcher wake-ups by reason.
113    pub dispatch_wakeups: Counter<u64>,
114    /// Time from wake-up to the first claim attempt.
115    pub dispatch_wake_to_claim_seconds: Histogram<f64>,
116    /// Number of permits available when a dispatcher wake is processed.
117    pub dispatch_capacity_available: Histogram<u64>,
118    /// Number of wakes that found no jobs despite available capacity.
119    pub dispatch_empty_claims: Counter<u64>,
120    /// Number of pre-acquired permits released unused after a claim round.
121    pub dispatch_unused_permits: Counter<u64>,
122    /// Number of wakes that were blocked by rate limiting.
123    pub dispatch_rate_limited: Counter<u64>,
124    /// Claim batch size distribution.
125    pub claim_batch_size: Histogram<u64>,
126    /// Claim query duration.
127    pub claim_duration_seconds: Histogram<f64>,
128    /// Job execution duration.
129    pub job_duration_seconds: Histogram<f64>,
130    /// Number of completion batch flushes executed.
131    pub completion_flushes: Counter<u64>,
132    /// Completion flush batch size distribution.
133    pub completion_flush_batch_size: Histogram<u64>,
134    /// Completion flush duration.
135    pub completion_flush_duration_seconds: Histogram<f64>,
136    /// Number of scheduled/retryable promotion batches executed.
137    pub promotion_batches: Counter<u64>,
138    /// Promotion batch size distribution.
139    pub promotion_batch_size: Histogram<u64>,
140    /// Promotion query duration.
141    pub promotion_duration_seconds: Histogram<f64>,
142    /// Current in-flight jobs (can go up and down).
143    pub jobs_in_flight: UpDownCounter<i64>,
144    /// Total heartbeat batches sent.
145    pub heartbeat_batches: Counter<u64>,
146    /// Total maintenance rescue operations.
147    pub maintenance_rescues: Counter<u64>,
148    /// Total jobs parked for external callback.
149    pub jobs_waiting_external: Counter<u64>,
150    /// Current queue depth per state — how many jobs are in each state per queue.
151    pub queue_depth: Gauge<i64>,
152    /// Queue lag — age of the oldest available job per queue.
153    pub queue_lag_seconds: Gauge<f64>,
154    /// Time from job creation to claim — the user-visible queuing latency.
155    pub wait_duration_seconds: Histogram<f64>,
156    /// Total jobs moved into the Dead Letter Queue.
157    pub dlq_moved: Counter<u64>,
158    /// Total jobs retried out of the Dead Letter Queue.
159    pub dlq_retried: Counter<u64>,
160    /// Total DLQ rows purged.
161    pub dlq_purged: Counter<u64>,
162    /// Current DLQ depth per queue.
163    pub dlq_depth: Gauge<i64>,
164    /// Info gauge for declared queue descriptors — value is always 1, the
165    /// useful payload is the attribute set (display_name, owner, tags).
166    /// Dashboards join it into throughput / latency panels with a
167    /// `* on(awa_job_queue) group_left(awa_queue_display_name, awa_queue_owner)`
168    /// Prometheus expression, which keeps descriptor fields out of the
169    /// high-cardinality per-metric label set.
170    pub queue_info: Gauge<i64>,
171    /// Info gauge for declared job-kind descriptors. Same pattern as
172    /// [`queue_info`][Self::queue_info].
173    pub job_kind_info: Gauge<i64>,
174    /// Readiness gauge for storage transition actions such as
175    /// `enter_mixed_transition` and `finalize` (1 = ready, 0 = blocked).
176    pub storage_transition_ready: Gauge<i64>,
177    /// Current canonical live backlog observed by queue-storage-capable runtimes.
178    pub storage_canonical_live_backlog: Gauge<i64>,
179    /// Current live runtime count per reported storage capability.
180    pub storage_live_runtime_capability: Gauge<i64>,
181    /// One-hot info gauge for the current storage transition state and engines.
182    pub storage_state: Gauge<i64>,
183    /// Maintenance ring rotation attempts, attributed to (ring, outcome,
184    /// blocker). For Rotated outcomes the `awa.ring.blocker` attribute is
185    /// "none"; for SkippedBusy it carries the per-ring blocker label
186    /// ("queue.ready_rows", "queue.claim_attempt_batches",
187    /// "queue.done_rows", "queue.tombstone_rows", "queue.ready_segments",
188    /// "queue.receipt_completion_batch_rows",
189    /// "queue.receipt_completion_tombstone_rows",
190    /// "queue.terminal_delta_rows", "lease.rows", "claim.rows",
191    /// "claim.closure_rows", "claim.closure_batch_rows"). One increment per
192    /// non-zero blocker means a single SkippedBusy with multiple populated
193    /// fields emits multiple events; that's intentional so dashboards can
194    /// attribute blame independently.
195    pub maintenance_rotate_attempts: Counter<u64>,
196    /// Magnitudes of the per-blocker row counts when a rotation is
197    /// SkippedBusy. Histogram so dashboards can show whether the ring is
198    /// pinned by handfuls of stragglers or by mountains of unfinished work.
199    pub maintenance_rotate_skipped_rows: Histogram<u64>,
200    /// Maintenance ring prune attempts, attributed to (ring, outcome, reason).
201    /// reason="none" for Pruned/Noop/Blocked; otherwise carries the
202    /// SkipReason discriminator (e.g. "queue.active_leases",
203    /// "queue.pending_ready", "claim.open").
204    pub maintenance_prune_attempts: Counter<u64>,
205    /// Magnitude of the reason count when prune returns SkippedActive.
206    pub maintenance_prune_skipped_rows: Histogram<u64>,
207    /// Per-branch wall-clock duration for the maintenance leader's main
208    /// `tokio::select!` loop, attributed by `awa.maintenance.branch`. Records
209    /// the time from the moment a select arm fires to the moment its body
210    /// returns — so each sample is the body's contribution to head-of-line
211    /// delay for every other branch on the same loop. Dashboards alert on
212    /// `histogram_quantile(0.99, ...)` per branch to see whether any one
213    /// branch is dominating select-loop time. Issue #242.
214    pub maintenance_branch_duration_seconds: Histogram<f64>,
215    /// Counter of "delayed tick" transitions per maintenance branch. One
216    /// increment is emitted when a branch fires after running longer than
217    /// its own tick interval in the previous iteration — i.e. the timer
218    /// was already overdue at the moment it fired. Emits on transition
219    /// (on-time -> delayed) only, not on every subsequent overrun tick,
220    /// so a sustained slow branch produces one event per "overrun episode"
221    /// rather than one per tick. Fleets alert on this rather than
222    /// scraping the matching `tracing::warn!` line. Issue #242.
223    pub maintenance_branch_overrun_total: Counter<u64>,
224    /// Current ring `current_slot` per ring, sampled from each rotate call.
225    /// The slot number itself isn't meaningful but the rate of advance is —
226    /// dashboards plot `rate(slot_changes)` to see whether rotation is
227    /// healthy or pinned.
228    pub ring_current_slot: Gauge<i64>,
229    /// Current ring `generation` per ring. Always-increasing; dashboards
230    /// show its derivative as "rotations per minute".
231    pub ring_generation: Gauge<i64>,
232}
233
234impl AwaMetrics {
235    /// Create metrics from an OpenTelemetry meter.
236    ///
237    /// Instrument names come from [`names`] so the public constants and the
238    /// registered instruments can't drift — a rename in `names::*` updates
239    /// the registration too.
240    pub fn new(meter: &Meter) -> Self {
241        Self {
242            jobs_inserted: meter
243                .u64_counter(names::JOB_INSERTED)
244                .with_description("Number of jobs inserted")
245                .with_unit("{job}")
246                .build(),
247            enqueue_batch_size: meter
248                .u64_histogram(names::ENQUEUE_BATCH_SIZE)
249                .with_description(
250                    "Direct queue-storage COPY enqueue: per-batch job count",
251                )
252                .with_unit("{job}")
253                .build(),
254            enqueue_duration_seconds: meter
255                .f64_histogram(names::ENQUEUE_DURATION)
256                .with_description(
257                    "Direct queue-storage COPY enqueue: per-batch wall-clock duration",
258                )
259                .with_unit("s")
260                .with_boundaries(WAIT_DURATION_BUCKETS_SECONDS.to_vec())
261                .build(),
262            jobs_completed: meter
263                .u64_counter(names::JOB_COMPLETED)
264                .with_description("Number of jobs completed successfully")
265                .with_unit("{job}")
266                .build(),
267            jobs_failed: meter
268                .u64_counter(names::JOB_FAILED)
269                .with_description("Number of jobs that failed terminally")
270                .with_unit("{job}")
271                .build(),
272            jobs_retried: meter
273                .u64_counter(names::JOB_RETRIED)
274                .with_description("Number of jobs marked retryable")
275                .with_unit("{job}")
276                .build(),
277            jobs_cancelled: meter
278                .u64_counter(names::JOB_CANCELLED)
279                .with_description("Number of jobs cancelled")
280                .with_unit("{job}")
281                .build(),
282            jobs_claimed: meter
283                .u64_counter(names::JOB_CLAIMED)
284                .with_description("Number of jobs claimed for execution")
285                .with_unit("{job}")
286                .build(),
287            claim_batches: meter
288                .u64_counter(names::DISPATCH_CLAIM_BATCHES)
289                .with_description("Number of dispatcher claim queries executed")
290                .with_unit("{batch}")
291                .build(),
292            dispatch_wakeups: meter
293                .u64_counter(names::DISPATCH_WAKEUPS)
294                .with_description("Number of dispatcher wake-ups by reason")
295                .with_unit("{wake}")
296                .build(),
297            dispatch_wake_to_claim_seconds: meter
298                .f64_histogram(names::DISPATCH_WAKE_TO_CLAIM_DURATION)
299                .with_description("Time from dispatcher wake-up to first claim attempt")
300                .with_unit("s")
301                .build(),
302            dispatch_capacity_available: meter
303                .u64_histogram(names::DISPATCH_CAPACITY_AVAILABLE)
304                .with_description("Number of permits available when a dispatcher wake is processed")
305                .with_unit("{permit}")
306                .build(),
307            dispatch_empty_claims: meter
308                .u64_counter(names::DISPATCH_EMPTY_CLAIMS)
309                .with_description("Number of dispatcher wakes that found no jobs despite available capacity")
310                .with_unit("{wake}")
311                .build(),
312            dispatch_unused_permits: meter
313                .u64_counter(names::DISPATCH_UNUSED_PERMITS)
314                .with_description("Number of pre-acquired permits released unused after claiming fewer jobs than capacity")
315                .with_unit("{permit}")
316                .build(),
317            dispatch_rate_limited: meter
318                .u64_counter(names::DISPATCH_RATE_LIMITED)
319                .with_description("Number of dispatcher wakes that could not claim because of rate limiting")
320                .with_unit("{wake}")
321                .build(),
322            claim_batch_size: meter
323                .u64_histogram(names::DISPATCH_CLAIM_BATCH_SIZE)
324                .with_description("Dispatcher claim batch size")
325                .with_unit("{job}")
326                .build(),
327            claim_duration_seconds: meter
328                .f64_histogram(names::DISPATCH_CLAIM_DURATION)
329                .with_description("Dispatcher claim query duration")
330                .with_unit("s")
331                .build(),
332            job_duration_seconds: meter
333                .f64_histogram(names::JOB_DURATION)
334                .with_description("Job execution duration")
335                .with_unit("s")
336                .build(),
337            completion_flushes: meter
338                .u64_counter(names::COMPLETION_FLUSHES)
339                .with_description("Number of completion batch flushes")
340                .with_unit("{batch}")
341                .build(),
342            completion_flush_batch_size: meter
343                .u64_histogram(names::COMPLETION_FLUSH_BATCH_SIZE)
344                .with_description("Completion batch flush size")
345                .with_unit("{job}")
346                .build(),
347            completion_flush_duration_seconds: meter
348                .f64_histogram(names::COMPLETION_FLUSH_DURATION)
349                .with_description("Completion batch flush duration")
350                .with_unit("s")
351                .build(),
352            promotion_batches: meter
353                .u64_counter(names::MAINTENANCE_PROMOTE_BATCHES)
354                .with_description("Number of scheduled/retryable promotion batches")
355                .with_unit("{batch}")
356                .build(),
357            promotion_batch_size: meter
358                .u64_histogram(names::MAINTENANCE_PROMOTE_BATCH_SIZE)
359                .with_description("Promotion batch size")
360                .with_unit("{job}")
361                .build(),
362            promotion_duration_seconds: meter
363                .f64_histogram(names::MAINTENANCE_PROMOTE_DURATION)
364                .with_description("Promotion batch duration")
365                .with_unit("s")
366                .build(),
367            jobs_in_flight: meter
368                .i64_up_down_counter(names::JOB_IN_FLIGHT)
369                .with_description("Current number of in-flight jobs")
370                .with_unit("{job}")
371                .build(),
372            heartbeat_batches: meter
373                .u64_counter(names::HEARTBEAT_BATCHES)
374                .with_description("Number of heartbeat batch updates sent")
375                .with_unit("{batch}")
376                .build(),
377            maintenance_rescues: meter
378                .u64_counter(names::MAINTENANCE_RESCUES)
379                .with_description("Number of jobs rescued by maintenance")
380                .with_unit("{job}")
381                .build(),
382            jobs_waiting_external: meter
383                .u64_counter(names::JOB_WAITING_EXTERNAL)
384                .with_description("Number of jobs parked for external callback")
385                .with_unit("{job}")
386                .build(),
387            queue_depth: meter
388                .i64_gauge(names::QUEUE_DEPTH)
389                .with_description("Current number of jobs per queue and state")
390                .with_unit("{job}")
391                .build(),
392            queue_lag_seconds: meter
393                .f64_gauge(names::QUEUE_LAG)
394                .with_description("Age of the oldest available job per queue")
395                .with_unit("s")
396                .build(),
397            wait_duration_seconds: meter
398                .f64_histogram(names::JOB_WAIT_DURATION)
399                .with_description("Time from job creation to claim")
400                .with_unit("s")
401                .with_boundaries(WAIT_DURATION_BUCKETS_SECONDS.to_vec())
402                .build(),
403            dlq_moved: meter
404                .u64_counter(names::JOB_DLQ_MOVED)
405                .with_description("Number of jobs moved into the Dead Letter Queue")
406                .with_unit("{job}")
407                .build(),
408            dlq_retried: meter
409                .u64_counter(names::JOB_DLQ_RETRIED)
410                .with_description("Number of jobs retried out of the Dead Letter Queue")
411                .with_unit("{job}")
412                .build(),
413            dlq_purged: meter
414                .u64_counter(names::JOB_DLQ_PURGED)
415                .with_description("Number of DLQ rows deleted")
416                .with_unit("{job}")
417                .build(),
418            dlq_depth: meter
419                .i64_gauge(names::JOB_DLQ_DEPTH)
420                .with_description("Current Dead Letter Queue depth per queue")
421                .with_unit("{job}")
422                .build(),
423            queue_info: meter
424                .i64_gauge(names::QUEUE_INFO)
425                .with_description(
426                    "Declared queue descriptors (always 1; use as a label-join target)",
427                )
428                .with_unit("{queue}")
429                .build(),
430            job_kind_info: meter
431                .i64_gauge(names::JOB_KIND_INFO)
432                .with_description(
433                    "Declared job-kind descriptors (always 1; use as a label-join target)",
434                )
435                .with_unit("{kind}")
436                .build(),
437            storage_transition_ready: meter
438                .i64_gauge(names::STORAGE_TRANSITION_READY)
439                .with_description("Storage transition readiness by action (1 = ready, 0 = blocked)")
440                .with_unit("{state}")
441                .build(),
442            storage_canonical_live_backlog: meter
443                .i64_gauge(names::STORAGE_CANONICAL_LIVE_BACKLOG)
444                .with_description("Current canonical live backlog during a storage transition")
445                .with_unit("{job}")
446                .build(),
447            storage_live_runtime_capability: meter
448                .i64_gauge(names::STORAGE_LIVE_RUNTIME_CAPABILITY)
449                .with_description("Current live runtime count by reported storage capability")
450                .with_unit("{runtime}")
451                .build(),
452            storage_state: meter
453                .i64_gauge(names::STORAGE_STATE)
454                .with_description(
455                    "Current storage transition state and engine combination (always 1)",
456                )
457                .with_unit("{state}")
458                .build(),
459            maintenance_rotate_attempts: meter
460                .u64_counter(names::MAINTENANCE_ROTATE_ATTEMPTS)
461                .with_description(
462                    "Ring rotation attempts by ring/outcome/blocker. Multiple increments per call when SkippedBusy has multiple non-zero blockers.",
463                )
464                .with_unit("{attempt}")
465                .build(),
466            maintenance_rotate_skipped_rows: meter
467                .u64_histogram(names::MAINTENANCE_ROTATE_SKIPPED_ROWS)
468                .with_description(
469                    "Row count for the blocker side of a SkippedBusy rotation",
470                )
471                .with_unit("{row}")
472                .build(),
473            maintenance_prune_attempts: meter
474                .u64_counter(names::MAINTENANCE_PRUNE_ATTEMPTS)
475                .with_description("Ring prune attempts by ring/outcome/reason")
476                .with_unit("{attempt}")
477                .build(),
478            maintenance_prune_skipped_rows: meter
479                .u64_histogram(names::MAINTENANCE_PRUNE_SKIPPED_ROWS)
480                .with_description("Magnitude of the reason count on a SkippedActive prune")
481                .with_unit("{row}")
482                .build(),
483            maintenance_branch_duration_seconds: meter
484                .f64_histogram(names::MAINTENANCE_BRANCH_DURATION)
485                .with_description(
486                    "Per-branch wall-clock duration of the maintenance leader's tokio::select! arms",
487                )
488                .with_unit("s")
489                .with_boundaries(WAIT_DURATION_BUCKETS_SECONDS.to_vec())
490                .build(),
491            maintenance_branch_overrun_total: meter
492                .u64_counter(names::MAINTENANCE_BRANCH_OVERRUN)
493                .with_description(
494                    "Maintenance branch overrun episodes: a branch fired after its previous run exceeded its tick interval",
495                )
496                .with_unit("{episode}")
497                .build(),
498            ring_current_slot: meter
499                .i64_gauge(names::RING_CURRENT_SLOT)
500                .with_description("Current slot index per ring (queue/lease/claim)")
501                .with_unit("{slot}")
502                .build(),
503            ring_generation: meter
504                .i64_gauge(names::RING_GENERATION)
505                .with_description("Current ring generation per ring; derivative is rotations/sec")
506                .with_unit("{generation}")
507                .build(),
508        }
509    }
510
511    /// Create metrics using the global OTel meter provider with meter name "awa".
512    pub fn from_global() -> Self {
513        let meter = opentelemetry::global::meter("awa");
514        Self::new(&meter)
515    }
516
517    /// Record a job completion with duration and attributes.
518    pub fn record_job_completed(&self, kind: &str, queue: &str, duration: Duration) {
519        let attrs = [
520            opentelemetry::KeyValue::new("awa.job.kind", kind.to_string()),
521            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
522        ];
523        self.jobs_completed.add(1, &attrs);
524        self.job_duration_seconds
525            .record(duration.as_secs_f64(), &attrs);
526    }
527
528    /// Record a job failure.
529    pub fn record_job_failed(&self, kind: &str, queue: &str, terminal: bool) {
530        let attrs = [
531            opentelemetry::KeyValue::new("awa.job.kind", kind.to_string()),
532            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
533            opentelemetry::KeyValue::new("awa.job.terminal", terminal),
534        ];
535        self.jobs_failed.add(1, &attrs);
536    }
537
538    /// Record a job retry.
539    pub fn record_job_retried(&self, kind: &str, queue: &str) {
540        let attrs = [
541            opentelemetry::KeyValue::new("awa.job.kind", kind.to_string()),
542            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
543        ];
544        self.jobs_retried.add(1, &attrs);
545    }
546
547    /// Record a producer batch through the direct queue-storage COPY
548    /// enqueue path (`QueueStorage::enqueue_params_copy` / Python
549    /// `Client.enqueue_many_copy`).
550    ///
551    /// `batch_size` is the row count that COPY actually wrote; `duration`
552    /// is the wall-clock time the producer spent in the COPY call (start
553    /// to commit), not including pre-batch row preparation.
554    ///
555    /// At `batch_size = 0` this is a no-op — empty batches are valid
556    /// callers and don't need a metric sample.
557    pub fn record_enqueue_batch(&self, queue: &str, batch_size: u64, duration: Duration) {
558        if batch_size == 0 {
559            return;
560        }
561        let attrs = [opentelemetry::KeyValue::new(
562            "awa.job.queue",
563            queue.to_string(),
564        )];
565        self.enqueue_batch_size.record(batch_size, &attrs);
566        self.enqueue_duration_seconds
567            .record(duration.as_secs_f64(), &attrs);
568    }
569
570    /// Record a job claimed from queue.
571    ///
572    /// Use this for the canonical engine, which has no per-shard
573    /// concept. The queue-storage engine uses
574    /// [`record_job_claimed_by_shard`][Self::record_job_claimed_by_shard]
575    /// so dashboards can read per-shard fairness directly.
576    pub fn record_job_claimed(&self, queue: &str, batch_size: u64) {
577        let attrs = [opentelemetry::KeyValue::new(
578            "awa.job.queue",
579            queue.to_string(),
580        )];
581        self.jobs_claimed.add(batch_size, &attrs);
582    }
583
584    /// Record a job claimed from queue, decorated with the enqueue shard.
585    ///
586    /// Used by the queue-storage path so dashboards can sum
587    /// `awa.job.claimed` by `awa.enqueue.shard` and confirm the claim
588    /// ordering is rotating across shards rather than starving the
589    /// higher-numbered ones. At `enqueue_shards > 1` this is the only
590    /// fairness signal that the operator gets from telemetry alone; at
591    /// `enqueue_shards = 1` the attribute is always `0` and the series
592    /// is identical to the un-decorated form.
593    ///
594    /// Call sites must not double-emit — invoke either this OR
595    /// `record_job_claimed`, never both for the same claim, or the
596    /// `awa.job.claimed` total will count each claim twice when
597    /// dashboards sum across all attribute combinations.
598    pub fn record_job_claimed_by_shard(&self, queue: &str, enqueue_shard: i16, batch_size: u64) {
599        if batch_size == 0 {
600            return;
601        }
602        let attrs = [
603            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
604            opentelemetry::KeyValue::new("awa.enqueue.shard", enqueue_shard as i64),
605        ];
606        self.jobs_claimed.add(batch_size, &attrs);
607    }
608
609    /// Record a dispatcher claim query batch and its latency.
610    pub fn record_claim_batch(&self, queue: &str, batch_size: u64, duration: Duration) {
611        let attrs = [opentelemetry::KeyValue::new(
612            "awa.job.queue",
613            queue.to_string(),
614        )];
615        self.claim_batches.add(1, &attrs);
616        self.claim_batch_size.record(batch_size, &attrs);
617        self.claim_duration_seconds
618            .record(duration.as_secs_f64(), &attrs);
619    }
620
621    /// Record a dispatcher wake-up reason.
622    pub fn record_dispatch_wake(&self, queue: &str, reason: &str) {
623        let attrs = [
624            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
625            opentelemetry::KeyValue::new("awa.dispatch.reason", reason.to_string()),
626        ];
627        self.dispatch_wakeups.add(1, &attrs);
628    }
629
630    /// Record time from wake-up to the first claim attempt.
631    pub fn record_dispatch_wake_to_claim(&self, queue: &str, reason: &str, duration: Duration) {
632        let attrs = [
633            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
634            opentelemetry::KeyValue::new("awa.dispatch.reason", reason.to_string()),
635        ];
636        self.dispatch_wake_to_claim_seconds
637            .record(duration.as_secs_f64(), &attrs);
638    }
639
640    /// Record how many permits were available on a dispatcher wake.
641    pub fn record_dispatch_capacity_available(&self, queue: &str, reason: &str, permits: u64) {
642        let attrs = [
643            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
644            opentelemetry::KeyValue::new("awa.dispatch.reason", reason.to_string()),
645        ];
646        self.dispatch_capacity_available.record(permits, &attrs);
647    }
648
649    /// Record a dispatcher wake that found no jobs.
650    pub fn record_dispatch_empty_claim(&self, queue: &str, reason: &str) {
651        let attrs = [
652            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
653            opentelemetry::KeyValue::new("awa.dispatch.reason", reason.to_string()),
654        ];
655        self.dispatch_empty_claims.add(1, &attrs);
656    }
657
658    /// Record permits released unused after a claim round.
659    pub fn record_dispatch_unused_permits(&self, queue: &str, count: u64) {
660        let attrs = [opentelemetry::KeyValue::new(
661            "awa.job.queue",
662            queue.to_string(),
663        )];
664        self.dispatch_unused_permits.add(count, &attrs);
665    }
666
667    /// Record a wake that could not claim because of rate limiting.
668    pub fn record_dispatch_rate_limited(&self, queue: &str, reason: &str) {
669        let attrs = [
670            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
671            opentelemetry::KeyValue::new("awa.dispatch.reason", reason.to_string()),
672        ];
673        self.dispatch_rate_limited.add(1, &attrs);
674    }
675
676    /// Record a completion batch flush.
677    pub fn record_completion_flush(&self, shard: usize, batch_size: u64, duration: Duration) {
678        let attrs = [opentelemetry::KeyValue::new(
679            "awa.completion.shard",
680            shard as i64,
681        )];
682        self.completion_flushes.add(1, &attrs);
683        self.completion_flush_batch_size.record(batch_size, &attrs);
684        self.completion_flush_duration_seconds
685            .record(duration.as_secs_f64(), &attrs);
686    }
687
688    /// Record a scheduled/retryable promotion batch.
689    pub fn record_promotion_batch(&self, state: &str, batch_size: u64, duration: Duration) {
690        let attrs = [opentelemetry::KeyValue::new(
691            "awa.job.state",
692            state.to_string(),
693        )];
694        self.promotion_batches.add(1, &attrs);
695        self.promotion_batch_size.record(batch_size, &attrs);
696        self.promotion_duration_seconds
697            .record(duration.as_secs_f64(), &attrs);
698    }
699
700    /// Record in-flight change.
701    pub fn record_in_flight_change(&self, queue: &str, delta: i64) {
702        let attrs = [opentelemetry::KeyValue::new(
703            "awa.job.queue",
704            queue.to_string(),
705        )];
706        self.jobs_in_flight.add(delta, &attrs);
707    }
708
709    /// Record queue depth for a specific state.
710    pub fn record_queue_depth(&self, queue: &str, state: &str, count: i64) {
711        let attrs = [
712            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
713            opentelemetry::KeyValue::new("awa.job.state", state.to_string()),
714        ];
715        self.queue_depth.record(count, &attrs);
716    }
717
718    /// Record queue lag (age of oldest available job).
719    pub fn record_queue_lag(&self, queue: &str, lag_seconds: f64) {
720        let attrs = [opentelemetry::KeyValue::new(
721            "awa.job.queue",
722            queue.to_string(),
723        )];
724        self.queue_lag_seconds.record(lag_seconds, &attrs);
725    }
726
727    /// Record job wait duration (time from creation to claim).
728    pub fn record_wait_duration(&self, queue: &str, seconds: f64) {
729        let attrs = [opentelemetry::KeyValue::new(
730            "awa.job.queue",
731            queue.to_string(),
732        )];
733        self.wait_duration_seconds.record(seconds, &attrs);
734    }
735
736    /// Record a job moved into the DLQ.
737    pub fn record_dlq_moved(&self, kind: &str, queue: &str, reason: &str) {
738        let attrs = [
739            opentelemetry::KeyValue::new("awa.job.kind", kind.to_string()),
740            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
741            opentelemetry::KeyValue::new("awa.dlq.reason", reason.to_string()),
742        ];
743        self.dlq_moved.add(1, &attrs);
744    }
745
746    /// Record a bulk admin move into the DLQ.
747    pub fn record_dlq_moved_bulk(
748        &self,
749        kind: Option<&str>,
750        queue: Option<&str>,
751        reason: &str,
752        count: u64,
753    ) {
754        if count == 0 {
755            return;
756        }
757
758        let mut attrs = vec![opentelemetry::KeyValue::new(
759            "awa.dlq.reason",
760            reason.to_string(),
761        )];
762        if let Some(kind) = kind {
763            attrs.push(opentelemetry::KeyValue::new(
764                "awa.job.kind",
765                kind.to_string(),
766            ));
767        }
768        if let Some(queue) = queue {
769            attrs.push(opentelemetry::KeyValue::new(
770                "awa.job.queue",
771                queue.to_string(),
772            ));
773        }
774        self.dlq_moved.add(count, &attrs);
775    }
776
777    /// Record jobs retried out of the DLQ.
778    pub fn record_dlq_retried(&self, queue: Option<&str>, count: u64) {
779        let attrs: Vec<opentelemetry::KeyValue> = queue
780            .map(|q| vec![opentelemetry::KeyValue::new("awa.job.queue", q.to_string())])
781            .unwrap_or_default();
782        self.dlq_retried.add(count, &attrs);
783    }
784
785    /// Record DLQ rows purged.
786    pub fn record_dlq_purged(&self, queue: Option<&str>, count: u64) {
787        let attrs: Vec<opentelemetry::KeyValue> = queue
788            .map(|q| vec![opentelemetry::KeyValue::new("awa.job.queue", q.to_string())])
789            .unwrap_or_default();
790        self.dlq_purged.add(count, &attrs);
791    }
792
793    /// Record current DLQ depth for a queue.
794    pub fn record_dlq_depth(&self, queue: &str, count: i64) {
795        let attrs = [opentelemetry::KeyValue::new(
796            "awa.job.queue",
797            queue.to_string(),
798        )];
799        self.dlq_depth.record(count, &attrs);
800    }
801
802    /// Emit the info gauge for a declared queue descriptor. Called once per
803    /// descriptor on every runtime snapshot tick — constant value of 1 with
804    /// the descriptor fields as attributes. Optional fields that are `None`
805    /// are elided so we don't produce `display_name=""` series.
806    pub fn record_queue_info(
807        &self,
808        queue: &str,
809        display_name: Option<&str>,
810        description: Option<&str>,
811        owner: Option<&str>,
812        docs_url: Option<&str>,
813        tags: &[String],
814    ) {
815        let mut attrs = vec![opentelemetry::KeyValue::new(
816            "awa.job.queue",
817            queue.to_string(),
818        )];
819        if let Some(v) = display_name {
820            attrs.push(opentelemetry::KeyValue::new(
821                "awa.queue.display_name",
822                v.to_string(),
823            ));
824        }
825        if let Some(v) = description {
826            attrs.push(opentelemetry::KeyValue::new(
827                "awa.queue.description",
828                v.to_string(),
829            ));
830        }
831        if let Some(v) = owner {
832            attrs.push(opentelemetry::KeyValue::new(
833                "awa.queue.owner",
834                v.to_string(),
835            ));
836        }
837        if let Some(v) = docs_url {
838            attrs.push(opentelemetry::KeyValue::new(
839                "awa.queue.docs_url",
840                v.to_string(),
841            ));
842        }
843        if !tags.is_empty() {
844            attrs.push(opentelemetry::KeyValue::new(
845                "awa.queue.tags",
846                tags.join(","),
847            ));
848        }
849        self.queue_info.record(1, &attrs);
850    }
851
852    /// Emit the info gauge for a declared job-kind descriptor. Same shape
853    /// as [`record_queue_info`][Self::record_queue_info].
854    pub fn record_job_kind_info(
855        &self,
856        kind: &str,
857        display_name: Option<&str>,
858        description: Option<&str>,
859        owner: Option<&str>,
860        docs_url: Option<&str>,
861        tags: &[String],
862    ) {
863        let mut attrs = vec![opentelemetry::KeyValue::new(
864            "awa.job.kind",
865            kind.to_string(),
866        )];
867        if let Some(v) = display_name {
868            attrs.push(opentelemetry::KeyValue::new(
869                "awa.job_kind.display_name",
870                v.to_string(),
871            ));
872        }
873        if let Some(v) = description {
874            attrs.push(opentelemetry::KeyValue::new(
875                "awa.job_kind.description",
876                v.to_string(),
877            ));
878        }
879        if let Some(v) = owner {
880            attrs.push(opentelemetry::KeyValue::new(
881                "awa.job_kind.owner",
882                v.to_string(),
883            ));
884        }
885        if let Some(v) = docs_url {
886            attrs.push(opentelemetry::KeyValue::new(
887                "awa.job_kind.docs_url",
888                v.to_string(),
889            ));
890        }
891        if !tags.is_empty() {
892            attrs.push(opentelemetry::KeyValue::new(
893                "awa.job_kind.tags",
894                tags.join(","),
895            ));
896        }
897        self.job_kind_info.record(1, &attrs);
898    }
899
900    /// Record whether a storage transition action is currently ready.
901    pub fn record_storage_transition_ready(&self, action: &str, ready: bool) {
902        let attrs = [opentelemetry::KeyValue::new(
903            "awa.storage.action",
904            action.to_string(),
905        )];
906        self.storage_transition_ready
907            .record(if ready { 1 } else { 0 }, &attrs);
908    }
909
910    /// Record canonical live backlog for the storage transition.
911    pub fn record_storage_canonical_live_backlog(&self, count: i64) {
912        self.storage_canonical_live_backlog.record(count, &[]);
913    }
914
915    /// Record the number of live runtimes reporting a given storage capability.
916    pub fn record_storage_live_runtime_capability(&self, capability: &str, count: i64) {
917        let attrs = [opentelemetry::KeyValue::new(
918            "awa.storage.capability",
919            capability.to_string(),
920        )];
921        self.storage_live_runtime_capability.record(count, &attrs);
922    }
923
924    /// Emit the current storage transition state as a one-hot info gauge.
925    pub fn record_storage_state(&self, status: &StorageStatus) {
926        let mut attrs = vec![
927            opentelemetry::KeyValue::new("awa.storage.state", status.state.clone()),
928            opentelemetry::KeyValue::new(
929                "awa.storage.current_engine",
930                status.current_engine.clone(),
931            ),
932            opentelemetry::KeyValue::new("awa.storage.active_engine", status.active_engine.clone()),
933        ];
934        if let Some(prepared_engine) = &status.prepared_engine {
935            attrs.push(opentelemetry::KeyValue::new(
936                "awa.storage.prepared_engine",
937                prepared_engine.clone(),
938            ));
939        }
940        self.storage_state.record(1, &attrs);
941    }
942
943    /// Record a ring rotation outcome.
944    ///
945    /// `ring` is one of "queue" / "lease" / "claim" — set by the caller based on
946    /// which rotate fn returned the outcome. For SkippedBusy, every non-zero
947    /// blocker count emits its own counter increment plus a histogram sample,
948    /// so a queue rotate skipped on (ready=42, done=17) produces two events
949    /// with different `awa.ring.blocker` labels. This makes
950    /// `sum by (awa.ring.blocker) (rate(...))` work cleanly in Grafana.
951    pub fn record_rotate_outcome(&self, ring: &'static str, outcome: &awa_model::RotateOutcome) {
952        match outcome {
953            awa_model::RotateOutcome::Rotated { slot, generation } => {
954                let attrs = [
955                    opentelemetry::KeyValue::new("awa.ring", ring),
956                    opentelemetry::KeyValue::new("awa.ring.outcome", "rotated"),
957                    opentelemetry::KeyValue::new("awa.ring.blocker", "none"),
958                ];
959                self.maintenance_rotate_attempts.add(1, &attrs);
960                let slot_attrs = [opentelemetry::KeyValue::new("awa.ring", ring)];
961                self.ring_current_slot.record(*slot as i64, &slot_attrs);
962                self.ring_generation.record(*generation, &slot_attrs);
963            }
964            awa_model::RotateOutcome::SkippedBusy { slot: _, busy } => {
965                let blockers: &[(&str, i64)] = &[
966                    ("queue.ready_rows", busy.queue_ready),
967                    (
968                        "queue.claim_attempt_batches",
969                        busy.queue_claim_attempt_batches,
970                    ),
971                    ("queue.done_rows", busy.queue_done),
972                    ("queue.tombstone_rows", busy.queue_tombstones),
973                    ("queue.ready_segments", busy.queue_ready_segments),
974                    (
975                        "queue.receipt_completion_batch_rows",
976                        busy.queue_receipt_completion_batches,
977                    ),
978                    (
979                        "queue.receipt_completion_tombstone_rows",
980                        busy.queue_receipt_completion_tombstones,
981                    ),
982                    ("queue.terminal_delta_rows", busy.queue_terminal_deltas),
983                    ("lease.rows", busy.leases),
984                    ("claim.rows", busy.claims),
985                    ("claim.closure_rows", busy.closures),
986                    ("claim.closure_batch_rows", busy.closure_batches),
987                ];
988                let mut emitted_any = false;
989                for (label, count) in blockers {
990                    if *count > 0 {
991                        emitted_any = true;
992                        let attrs = [
993                            opentelemetry::KeyValue::new("awa.ring", ring),
994                            opentelemetry::KeyValue::new("awa.ring.outcome", "skipped_busy"),
995                            opentelemetry::KeyValue::new("awa.ring.blocker", *label),
996                        ];
997                        self.maintenance_rotate_attempts.add(1, &attrs);
998                        self.maintenance_rotate_skipped_rows
999                            .record(*count as u64, &attrs);
1000                    }
1001                }
1002                // Lost-CAS path can return SkippedBusy with all-zero counts
1003                // (the row counts we sampled before were stale by the time
1004                // we lost the race). Emit a single attempt event so the
1005                // counter still reflects the call.
1006                if !emitted_any {
1007                    let attrs = [
1008                        opentelemetry::KeyValue::new("awa.ring", ring),
1009                        opentelemetry::KeyValue::new("awa.ring.outcome", "skipped_busy"),
1010                        opentelemetry::KeyValue::new("awa.ring.blocker", "lost_cas"),
1011                    ];
1012                    self.maintenance_rotate_attempts.add(1, &attrs);
1013                }
1014            }
1015        }
1016    }
1017
1018    /// Record the wall-clock duration of one maintenance `tokio::select!`
1019    /// arm. `branch` is a static name (e.g. `"promote_scheduled"`,
1020    /// `"rescue_stale_heartbeats"`) so the attribute set stays bounded.
1021    /// Issue #242.
1022    pub fn record_maintenance_branch_duration(&self, branch: &'static str, duration: Duration) {
1023        let attrs = [opentelemetry::KeyValue::new(
1024            "awa.maintenance.branch",
1025            branch,
1026        )];
1027        self.maintenance_branch_duration_seconds
1028            .record(duration.as_secs_f64(), &attrs);
1029    }
1030
1031    /// Record one maintenance branch overrun episode — a transition from
1032    /// "on-time" to "delayed" for `branch`. Increments
1033    /// `awa.maintenance.branch.overrun` (Prometheus:
1034    /// `awa_maintenance_branch_overrun_total{branch="<name>"}`). Issue #242.
1035    pub fn record_maintenance_branch_overrun(&self, branch: &'static str) {
1036        let attrs = [opentelemetry::KeyValue::new(
1037            "awa.maintenance.branch",
1038            branch,
1039        )];
1040        self.maintenance_branch_overrun_total.add(1, &attrs);
1041    }
1042
1043    /// Record a ring prune outcome.
1044    pub fn record_prune_outcome(&self, ring: &'static str, outcome: &awa_model::PruneOutcome) {
1045        let (label, reason, count) = match outcome {
1046            awa_model::PruneOutcome::Noop => ("noop", "none", None),
1047            awa_model::PruneOutcome::Pruned { .. } => ("pruned", "none", None),
1048            awa_model::PruneOutcome::Blocked { .. } => ("blocked", "none", None),
1049            awa_model::PruneOutcome::SkippedActive { reason, count, .. } => {
1050                ("skipped_active", reason.as_str(), Some(*count))
1051            }
1052        };
1053        let attrs = [
1054            opentelemetry::KeyValue::new("awa.ring", ring),
1055            opentelemetry::KeyValue::new("awa.ring.outcome", label),
1056            opentelemetry::KeyValue::new("awa.ring.reason", reason),
1057        ];
1058        self.maintenance_prune_attempts.add(1, &attrs);
1059        if let Some(c) = count {
1060            self.maintenance_prune_skipped_rows.record(c as u64, &attrs);
1061        }
1062    }
1063}
1064
1065/// No-op metrics for when OTel is not configured.
1066impl Default for AwaMetrics {
1067    fn default() -> Self {
1068        Self::from_global()
1069    }
1070}
1071
1072#[cfg(test)]
1073mod tests {
1074    use super::*;
1075
1076    /// The info gauges are no-op under a default (global) meter provider —
1077    /// this just confirms the method signatures build and don't panic when
1078    /// called with a realistic attribute mix. End-to-end OTLP export is
1079    /// covered by the telemetry integration test.
1080    #[test]
1081    fn record_queue_info_does_not_panic_on_mixed_attrs() {
1082        let metrics = AwaMetrics::from_global();
1083        metrics.record_queue_info(
1084            "emails",
1085            Some("Outbound email"),
1086            Some("Transactional mail"),
1087            Some("growth@example.com"),
1088            Some("https://runbook/emails"),
1089            &["user-facing".to_string(), "critical".to_string()],
1090        );
1091        // With every optional field absent only the queue label is emitted.
1092        metrics.record_queue_info("minimal", None, None, None, None, &[]);
1093    }
1094
1095    #[test]
1096    fn record_job_kind_info_does_not_panic_on_mixed_attrs() {
1097        let metrics = AwaMetrics::from_global();
1098        metrics.record_job_kind_info(
1099            "send_email",
1100            Some("Send user email"),
1101            None,
1102            Some("growth@example.com"),
1103            None,
1104            &["outbound".to_string()],
1105        );
1106        metrics.record_job_kind_info("minimal", None, None, None, None, &[]);
1107    }
1108}
awa_metrics/lib.rs

awa_metrics/
lib.rs
