systemprompt_api/services/server/
ops_metrics.rs

1//! Operational signals for running many replicas: gateway phases and streams,
2//! pool acquire wait, scheduler jobs, projection backlog and dropped log rows.
3//!
4//! sqlx hands the pool straight to every query, so there is no central
5//! `acquire` to time. `db_pool_acquire_seconds{pool}` is instead a probe: every
6//! five seconds the sampler acquires one connection and records how long the
7//! pool made it wait, which is the wait a request arriving at that moment saw.
8//!
9//! `projection_backlog_depth{projection}` is sampled only on a node that runs
10//! the scheduler, since that is where the backlog drains.
11//! `log_entries_dropped_total` mirrors the database log sink's in-process drop
12//! counter.
13//!
14//! Copyright (c) systemprompt.io — Business Source License 1.1.
15//! See <https://systemprompt.io> for licensing details.
16
17use std::collections::HashSet;
18use std::sync::Arc;
19use std::time::{Duration, Instant};
20
21use metrics_exporter_prometheus::{BuildError, Matcher, PrometheusBuilder};
22use sqlx::PgPool;
23use systemprompt_config::ConfigFingerprint;
24use systemprompt_database::Database;
25use systemprompt_runtime::AppContext;
26use systemprompt_scheduler::services::job_metrics::{
27    JOB_DURATION_BUCKETS, SCHEDULER_JOB_DURATION_SECONDS, SCHEDULER_JOB_RUNS_TOTAL,
28};
29
30pub(super) const DB_POOL_ACQUIRE_SECONDS: &str = "db_pool_acquire_seconds";
31pub(super) const PROJECTION_BACKLOG_DEPTH: &str = "projection_backlog_depth";
32pub(super) const LOG_ENTRIES_DROPPED_TOTAL: &str = "log_entries_dropped_total";
33
34const ACQUIRE_BUCKETS: [f64; 12] = [
35    0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
36];
37const ACQUIRE_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
38const SAMPLE_INTERVAL: Duration = Duration::from_secs(5);
39const BACKLOG_INTERVAL: Duration = Duration::from_secs(30);
40
41pub(super) fn with_buckets(builder: PrometheusBuilder) -> Result<PrometheusBuilder, BuildError> {
42    builder
43        .set_buckets_for_metric(
44            Matcher::Full(systemprompt_gateway::GATEWAY_PHASE_SECONDS.to_owned()),
45            &systemprompt_gateway::PHASE_BUCKETS,
46        )?
47        .set_buckets_for_metric(
48            Matcher::Full(DB_POOL_ACQUIRE_SECONDS.to_owned()),
49            &ACQUIRE_BUCKETS,
50        )?
51        .set_buckets_for_metric(
52            Matcher::Full(SCHEDULER_JOB_DURATION_SECONDS.to_owned()),
53            &JOB_DURATION_BUCKETS,
54        )
55}
56
57pub(super) fn describe() {
58    metrics::describe_histogram!(
59        systemprompt_gateway::GATEWAY_PHASE_SECONDS,
60        metrics::Unit::Seconds,
61        "Gateway request time by phase (admission, upstream_first_byte, stream, finalize)"
62    );
63    metrics::describe_gauge!(
64        systemprompt_gateway::GATEWAY_STREAMS_IN_FLIGHT,
65        "Streamed gateway response bodies still being written by this process"
66    );
67    metrics::describe_histogram!(
68        DB_POOL_ACQUIRE_SECONDS,
69        metrics::Unit::Seconds,
70        "Time a probe waited to acquire a Postgres connection from the pool"
71    );
72    metrics::describe_counter!(
73        SCHEDULER_JOB_RUNS_TOTAL,
74        "Scheduled job runs by job and status (success, idle, error)"
75    );
76    metrics::describe_histogram!(
77        SCHEDULER_JOB_DURATION_SECONDS,
78        metrics::Unit::Seconds,
79        "Scheduled job run duration by job"
80    );
81    metrics::describe_gauge!(
82        PROJECTION_BACKLOG_DEPTH,
83        "Keys queued for a projection refresh and not yet drained"
84    );
85    metrics::describe_counter!(
86        LOG_ENTRIES_DROPPED_TOTAL,
87        "Log entries the database sink dropped because its queue was full"
88    );
89    metrics::describe_counter!(
90        systemprompt_events::CACHE_INVALIDATIONS_RECEIVED_TOTAL,
91        "Cross-replica cache invalidations received, by cache domain"
92    );
93}
94
95pub(super) fn spawn_samplers(ctx: &AppContext, runs_scheduler: bool) {
96    let mut sampler = OpsSampler::new(ctx.db_pool(), runs_scheduler);
97    ctx.background_tasks()
98        .spawn_cancellable("ops_metrics", |cancel| async move {
99            let mut tick = tokio::time::interval(SAMPLE_INTERVAL);
100            tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
101            loop {
102                tokio::select! {
103                    () = cancel.cancelled() => break,
104                    _ = tick.tick() => {},
105                }
106                sampler.sample().await;
107            }
108        });
109}
110
111#[derive(Debug)]
112pub struct OpsSampler {
113    write: Arc<PgPool>,
114    read: Option<Arc<PgPool>>,
115    backlog_pool: Option<Arc<PgPool>>,
116    backlog_due: Instant,
117    seen_projections: HashSet<String>,
118    config_info: Option<ConfigFingerprint>,
119}
120
121impl OpsSampler {
122    #[must_use]
123    pub fn new(db: &Database, runs_scheduler: bool) -> Self {
124        let write = db.write_pool();
125        let read = db.pool();
126        let read = (!Arc::ptr_eq(&read, &write)).then_some(read);
127        let backlog_pool = runs_scheduler.then(|| Arc::clone(&write));
128        Self {
129            write,
130            read,
131            backlog_pool,
132            backlog_due: Instant::now(),
133            seen_projections: HashSet::new(),
134            config_info: None,
135        }
136    }
137
138    pub async fn sample(&mut self) {
139        probe_acquire("write", &self.write).await;
140        if let Some(read) = self.read.as_ref() {
141            probe_acquire("read", read).await;
142        }
143        super::config_fingerprint::publish(&mut self.config_info);
144        metrics::counter!(LOG_ENTRIES_DROPPED_TOTAL)
145            .absolute(systemprompt_logging::dropped_log_entries());
146        if let Some(pool) = self.backlog_pool.as_ref()
147            && Instant::now() >= self.backlog_due
148        {
149            self.backlog_due = Instant::now() + BACKLOG_INTERVAL;
150            sample_backlog(pool, &mut self.seen_projections).await;
151        }
152    }
153}
154
155async fn probe_acquire(label: &'static str, pool: &PgPool) {
156    let started = Instant::now();
157    match tokio::time::timeout(ACQUIRE_PROBE_TIMEOUT, pool.acquire()).await {
158        Ok(Ok(connection)) => {
159            let waited = started.elapsed();
160            drop(connection);
161            metrics::histogram!(DB_POOL_ACQUIRE_SECONDS, "pool" => label)
162                .record(waited.as_secs_f64());
163        },
164        Ok(Err(error)) => {
165            tracing::warn!(pool = label, error = %error, "pool acquire probe failed");
166        },
167        Err(_) => {
168            metrics::histogram!(DB_POOL_ACQUIRE_SECONDS, "pool" => label)
169                .record(ACQUIRE_PROBE_TIMEOUT.as_secs_f64());
170        },
171    }
172}
173
174async fn sample_backlog(pool: &PgPool, seen: &mut HashSet<String>) {
175    match systemprompt_database::projection::depths(pool).await {
176        Ok(depths) => {
177            for drained in seen
178                .iter()
179                .filter(|name| !depths.iter().any(|(p, _)| p == *name))
180            {
181                metrics::gauge!(PROJECTION_BACKLOG_DEPTH, "projection" => drained.clone()).set(0.0);
182            }
183            for (projection, depth) in depths {
184                seen.insert(projection.clone());
185                #[expect(
186                    clippy::cast_precision_loss,
187                    reason = "a backlog depth is far below 2^52"
188                )]
189                let depth = depth as f64;
190                metrics::gauge!(PROJECTION_BACKLOG_DEPTH, "projection" => projection).set(depth);
191            }
192        },
193        Err(error) => {
194            tracing::warn!(error = %error, "projection backlog sample failed");
195        },
196    }
197}
systemprompt_api/services/server/ops_metrics.rs

systemprompt_api/services/server/
ops_metrics.rs
