1use std::{sync::OnceLock, time::Duration};
2
3use opentelemetry::{
4 KeyValue, global,
5 metrics::{Counter, Gauge, Histogram},
6};
7
8struct KernelMetrics {
9 tasks: Counter<u64>,
10 lease_renewals: Counter<u64>,
11 queue_latency: Histogram<f64>,
12 execution_duration: Histogram<f64>,
13 schedule_occurrences: Counter<u64>,
14 schedule_skipped_occurrences: Counter<u64>,
15 schedule_lag: Histogram<f64>,
16 schedule_materialization_duration: Histogram<f64>,
17 queue_ready_tasks: Gauge<u64>,
18 workers_live: Gauge<u64>,
19 worker_heartbeats: Counter<u64>,
20 queue_unroutable_tasks: Gauge<u64>,
21 worker_configured_concurrency: Gauge<u64>,
22 worker_effective_concurrency: Gauge<u64>,
23 worker_active_handlers: Gauge<u64>,
24 worker_available_slots: Gauge<u64>,
25 worker_event_loop_lag: Gauge<f64>,
26 worker_lease_renewal_age: Gauge<f64>,
27 worker_admission_limit: Gauge<u64>,
28}
29
30fn metrics() -> &'static KernelMetrics {
31 static METRICS: OnceLock<KernelMetrics> = OnceLock::new();
32 METRICS.get_or_init(|| {
33 let meter = global::meter("pgtask");
34 KernelMetrics {
35 tasks: meter
36 .u64_counter("pgtask.tasks")
37 .with_description("Task state transitions")
38 .build(),
39 lease_renewals: meter
40 .u64_counter("pgtask.lease.renewals")
41 .with_description("Lease renewal outcomes")
42 .build(),
43 queue_latency: meter
44 .f64_histogram("pgtask.queue.latency")
45 .with_description("Time from task creation until execution starts")
46 .with_unit("s")
47 .build(),
48 execution_duration: meter
49 .f64_histogram("pgtask.execution.duration")
50 .with_description("Task handler execution duration")
51 .with_unit("s")
52 .build(),
53 schedule_occurrences: meter
54 .u64_counter("pgtask.schedule.occurrences")
55 .with_description("Schedule occurrences materialized as tasks")
56 .build(),
57 schedule_skipped_occurrences: meter
58 .u64_counter("pgtask.schedule.skipped_occurrences")
59 .with_description("Due schedule occurrences discarded by the misfire policy")
60 .build(),
61 schedule_lag: meter
62 .f64_histogram("pgtask.schedule.lag")
63 .with_description("Time from a logical schedule occurrence until materialization")
64 .with_unit("s")
65 .build(),
66 schedule_materialization_duration: meter
67 .f64_histogram("pgtask.schedule.materialization.duration")
68 .with_description("Schedule materialization transaction duration")
69 .with_unit("s")
70 .build(),
71 workers_live: meter
72 .u64_gauge("pgtask.workers.live")
73 .with_description("Workers the database still considers live for this queue")
74 .with_unit("{worker}")
75 .build(),
76 worker_heartbeats: meter
77 .u64_counter("pgtask.worker.heartbeats")
78 .with_description("Worker heartbeat attempts by outcome")
79 .build(),
80 queue_ready_tasks: meter
81 .u64_gauge("pgtask.queue.ready.tasks")
82 .with_description("Due tasks supported by this worker process")
83 .with_unit("{task}")
84 .build(),
85 queue_unroutable_tasks: meter
86 .u64_gauge("pgtask.queue.unroutable.tasks")
87 .with_description("Due tasks with no live capable worker")
88 .with_unit("{task}")
89 .build(),
90 worker_configured_concurrency: meter
91 .u64_gauge("pgtask.worker.concurrency.configured")
92 .with_description("Configured maximum active task handlers")
93 .with_unit("{handler}")
94 .build(),
95 worker_effective_concurrency: meter
96 .u64_gauge("pgtask.worker.concurrency.effective")
97 .with_description("Current admission limit for active task handlers")
98 .with_unit("{handler}")
99 .build(),
100 worker_active_handlers: meter
101 .u64_gauge("pgtask.worker.handlers.active")
102 .with_description("Active task handlers")
103 .with_unit("{handler}")
104 .build(),
105 worker_available_slots: meter
106 .u64_gauge("pgtask.worker.slots.available")
107 .with_description("Task handler slots available under the effective admission limit")
108 .with_unit("{handler}")
109 .build(),
110 worker_event_loop_lag: meter
111 .f64_gauge("pgtask.worker.event_loop.lag")
112 .with_description("Delay beyond the expected worker runtime health-sampling deadline")
113 .with_unit("s")
114 .build(),
115 worker_lease_renewal_age: meter
116 .f64_gauge("pgtask.worker.lease.renewal.age")
117 .with_description("Age of the oldest active task lease renewal")
118 .with_unit("s")
119 .build(),
120 worker_admission_limit: meter
121 .u64_gauge("pgtask.worker.admission.limit")
122 .with_description("Proposed or applied worker admission limit changes")
123 .with_unit("{handler}")
124 .build(),
125 }
126 })
127}
128
129fn record_task_transition(state: &'static str, queue_name: &str, task_name: Option<&str>, count: u64) {
130 let mut attributes = vec![
131 KeyValue::new("pgtask.task.state", state),
132 KeyValue::new("pgtask.queue.name", queue_name.to_owned()),
133 ];
134 if let Some(task_name) = task_name {
135 attributes.push(KeyValue::new("pgtask.task.name", task_name.to_owned()));
136 }
137 metrics().tasks.add(count, &attributes);
138}
139
140pub fn record_enqueued(queue_name: &str, task_name: &str, count: u64) {
141 record_task_transition("enqueued", queue_name, Some(task_name), count);
142}
143
144pub fn record_claimed(queue_name: &str, task_name: &str) {
145 record_task_transition("claimed", queue_name, Some(task_name), 1);
146}
147
148pub fn record_cancelled(queue_name: &str, task_name: &str) {
149 record_task_transition("cancelled", queue_name, Some(task_name), 1);
150}
151
152pub fn record_succeeded(queue_name: &str, task_name: &str) {
153 record_task_transition("succeeded", queue_name, Some(task_name), 1);
154}
155
156pub fn record_failed(queue_name: &str, task_name: &str) {
157 record_task_transition("failed", queue_name, Some(task_name), 1);
158}
159
160pub fn record_retried(queue_name: &str, task_name: &str) {
161 record_task_transition("retried", queue_name, Some(task_name), 1);
162}
163
164pub fn record_recovered(queue_name: &str, count: u64) {
165 record_task_transition("recovered", queue_name, None, count);
166}
167
168pub fn record_lease_lost(queue_name: &str, task_name: &str) {
169 record_task_transition("lease_lost", queue_name, Some(task_name), 1);
170}
171
172pub fn record_renewed(queue_name: &str, task_name: &str, renewed: bool) {
173 metrics().lease_renewals.add(
174 1,
175 &[
176 KeyValue::new("pgtask.queue.name", queue_name.to_owned()),
177 KeyValue::new("pgtask.task.name", task_name.to_owned()),
178 KeyValue::new("pgtask.lease.renewed", renewed),
179 ],
180 );
181}
182
183pub fn record_queue_latency(queue_name: &str, task_name: &str, duration: Duration) {
184 metrics().queue_latency.record(
185 duration.as_secs_f64(),
186 &[
187 KeyValue::new("pgtask.queue.name", queue_name.to_owned()),
188 KeyValue::new("pgtask.task.name", task_name.to_owned()),
189 ],
190 );
191}
192
193pub fn record_queue_demand(queue_name: &str, capable_tasks: u64, unroutable_tasks: u64) {
194 let attributes = [KeyValue::new("pgtask.queue.name", queue_name.to_owned())];
195 metrics().queue_ready_tasks.record(capable_tasks, &attributes);
196 metrics().queue_unroutable_tasks.record(unroutable_tasks, &attributes);
197}
198
199pub fn record_heartbeat(queue_name: &str, outcome: &'static str) {
201 metrics().worker_heartbeats.add(
202 1,
203 &[
204 KeyValue::new("pgtask.queue.name", queue_name.to_owned()),
205 KeyValue::new("pgtask.heartbeat.outcome", outcome),
206 ],
207 );
208}
209
210pub fn record_live_workers(queue_name: &str, live: u64) {
211 metrics()
212 .workers_live
213 .record(live, &[KeyValue::new("pgtask.queue.name", queue_name.to_owned())]);
214}
215
216pub fn record_execution(queue_name: &str, task_name: &str, outcome: &'static str, duration: Duration) {
217 metrics().execution_duration.record(
218 duration.as_secs_f64(),
219 &[
220 KeyValue::new("pgtask.queue.name", queue_name.to_owned()),
221 KeyValue::new("pgtask.task.name", task_name.to_owned()),
222 KeyValue::new("pgtask.execution.outcome", outcome),
223 ],
224 );
225}
226
227pub fn record_schedule_occurrences(
228 queue_name: &str,
229 task_name: &str,
230 kind: &'static str,
231 count: u64,
232 skipped: u64,
233 lag: Duration,
234) {
235 let attributes = [
236 KeyValue::new("pgtask.queue.name", queue_name.to_owned()),
237 KeyValue::new("pgtask.task.name", task_name.to_owned()),
238 KeyValue::new("pgtask.schedule.kind", kind),
239 ];
240 metrics().schedule_occurrences.add(count, &attributes);
241 metrics().schedule_skipped_occurrences.add(skipped, &attributes);
242 metrics().schedule_lag.record(lag.as_secs_f64(), &attributes);
243}
244
245pub fn record_schedule_materialization(duration: Duration) {
246 metrics()
247 .schedule_materialization_duration
248 .record(duration.as_secs_f64(), &[]);
249}
250
251pub fn record_worker_capacity(queue_name: &str, configured: u16, effective: u16, active: usize) {
252 let attributes = [KeyValue::new("pgtask.queue.name", queue_name.to_owned())];
253 let active = u64::try_from(active).unwrap_or(u64::MAX);
254 metrics()
255 .worker_configured_concurrency
256 .record(u64::from(configured), &attributes);
257 metrics()
258 .worker_effective_concurrency
259 .record(u64::from(effective), &attributes);
260 metrics().worker_active_handlers.record(active, &attributes);
261 metrics()
262 .worker_available_slots
263 .record(u64::from(effective).saturating_sub(active), &attributes);
264}
265
266pub fn record_worker_event_loop_lag(queue_name: &str, duration: Duration) {
267 metrics().worker_event_loop_lag.record(
268 duration.as_secs_f64(),
269 &[KeyValue::new("pgtask.queue.name", queue_name.to_owned())],
270 );
271}
272
273pub fn record_worker_lease_renewal_age(queue_name: &str, duration: Duration) {
274 metrics().worker_lease_renewal_age.record(
275 duration.as_secs_f64(),
276 &[KeyValue::new("pgtask.queue.name", queue_name.to_owned())],
277 );
278}
279
280pub fn record_worker_admission_limit(queue_name: &str, decision: &'static str, reason: &'static str, limit: u16) {
281 metrics().worker_admission_limit.record(
282 u64::from(limit),
283 &[
284 KeyValue::new("pgtask.queue.name", queue_name.to_owned()),
285 KeyValue::new("pgtask.admission.decision", decision),
286 KeyValue::new("pgtask.admission.reason", reason),
287 ],
288 );
289}