pgsync_core/
metrics.rs

1//! Process-wide counters, cheap to update from any task and snapshot-able as JSON.
2
3use crate::lsn::Lsn;
4use crate::state::JobState;
5use serde::Serialize;
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::time::Instant;
8
9#[derive(Default)]
10pub struct Metrics {
11    pub events_received: AtomicU64,
12    pub bytes_received: AtomicU64,
13    pub txns_applied: AtomicU64,
14    pub txns_skipped: AtomicU64,
15    pub rows_applied: AtomicU64,
16    pub rows_copied: AtomicU64,
17    pub bytes_copied: AtomicU64,
18    pub feedback_sent: AtomicU64,
19    pub reconnects: AtomicU64,
20    pub last_received_lsn: AtomicU64,
21    pub server_wal_end: AtomicU64,
22    pub committed_lsn: AtomicU64,
23    pub acked_lsn: AtomicU64,
24    pub inflight_items: AtomicU64,
25    pub inflight_bytes: AtomicU64,
26    /// Unix microseconds of the last target commit (0 = none yet).
27    pub last_apply_unix_micros: AtomicU64,
28    state: AtomicU64,
29    connection: AtomicU64,
30    started: Option<Instant>,
31}
32
33/// State of the source replication connection as seen by the engine.
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
35#[serde(rename_all = "snake_case")]
36pub enum ConnectionState {
37    /// No stream opened yet, or the job is not running.
38    Disconnected,
39    Connected,
40    /// A transient failure is being retried with backoff.
41    Reconnecting,
42}
43
44const CONNECTIONS: [ConnectionState; 3] = [
45    ConnectionState::Disconnected,
46    ConnectionState::Connected,
47    ConnectionState::Reconnecting,
48];
49
50#[derive(Debug, Clone, Serialize)]
51pub struct MetricsSnapshot {
52    pub state: JobState,
53    pub uptime_secs: u64,
54    pub events_received: u64,
55    pub bytes_received: u64,
56    pub txns_applied: u64,
57    pub txns_skipped: u64,
58    pub rows_applied: u64,
59    pub rows_copied: u64,
60    pub bytes_copied: u64,
61    pub feedback_sent: u64,
62    pub reconnects: u64,
63    pub last_received_lsn: Lsn,
64    pub server_wal_end: Lsn,
65    pub committed_lsn: Lsn,
66    pub acked_lsn: Lsn,
67    /// `server_wal_end - acked_lsn`, bytes: WAL the server has produced that this job has
68    /// not yet durably covered (applied, or acknowledged as carrying nothing for it).
69    pub apply_lag_bytes: u64,
70    pub inflight_items: u64,
71    pub inflight_bytes: u64,
72    pub rss_bytes: Option<u64>,
73    pub last_apply_unix_micros: Option<u64>,
74    pub connection: ConnectionState,
75}
76
77const STATES: [JobState; 7] = [
78    JobState::Initializing,
79    JobState::Snapshotting,
80    JobState::CatchingUp,
81    JobState::Streaming,
82    JobState::Paused,
83    JobState::Error,
84    JobState::ResyncRequired,
85];
86
87impl Metrics {
88    pub fn new() -> Self {
89        Metrics {
90            started: Some(Instant::now()),
91            ..Default::default()
92        }
93    }
94
95    pub fn set_state(&self, s: JobState) {
96        let idx = STATES.iter().position(|x| *x == s).unwrap_or(0) as u64;
97        self.state.store(idx, Ordering::Relaxed);
98    }
99
100    pub fn state(&self) -> JobState {
101        STATES[self.state.load(Ordering::Relaxed) as usize % STATES.len()]
102    }
103
104    pub fn set_connection(&self, c: ConnectionState) {
105        let idx = CONNECTIONS.iter().position(|x| *x == c).unwrap_or(0) as u64;
106        self.connection.store(idx, Ordering::Relaxed);
107    }
108
109    pub fn connection(&self) -> ConnectionState {
110        CONNECTIONS[self.connection.load(Ordering::Relaxed) as usize % CONNECTIONS.len()]
111    }
112
113    pub fn mark_applied_now(&self) {
114        let micros = std::time::SystemTime::now()
115            .duration_since(std::time::UNIX_EPOCH)
116            .map(|d| d.as_micros() as u64)
117            .unwrap_or(0);
118        self.last_apply_unix_micros.store(micros, Ordering::Relaxed);
119    }
120
121    pub fn add(&self, counter: &AtomicU64, n: u64) {
122        counter.fetch_add(n, Ordering::Relaxed);
123    }
124
125    pub fn set_lsn(&self, slot: &AtomicU64, lsn: Lsn) {
126        slot.fetch_max(lsn.0, Ordering::Relaxed);
127    }
128
129    pub fn snapshot(&self) -> MetricsSnapshot {
130        let g = |a: &AtomicU64| a.load(Ordering::Relaxed);
131        let wal_end = g(&self.server_wal_end);
132        let committed = g(&self.committed_lsn);
133        let acked = g(&self.acked_lsn);
134        MetricsSnapshot {
135            state: self.state(),
136            uptime_secs: self.started.map(|s| s.elapsed().as_secs()).unwrap_or(0),
137            events_received: g(&self.events_received),
138            bytes_received: g(&self.bytes_received),
139            txns_applied: g(&self.txns_applied),
140            txns_skipped: g(&self.txns_skipped),
141            rows_applied: g(&self.rows_applied),
142            rows_copied: g(&self.rows_copied),
143            bytes_copied: g(&self.bytes_copied),
144            feedback_sent: g(&self.feedback_sent),
145            reconnects: g(&self.reconnects),
146            last_received_lsn: Lsn(g(&self.last_received_lsn)),
147            server_wal_end: Lsn(wal_end),
148            committed_lsn: Lsn(committed),
149            acked_lsn: Lsn(acked),
150            apply_lag_bytes: wal_end.saturating_sub(acked.max(committed)),
151            inflight_items: g(&self.inflight_items),
152            inflight_bytes: g(&self.inflight_bytes),
153            rss_bytes: process_rss_bytes(),
154            last_apply_unix_micros: match g(&self.last_apply_unix_micros) {
155                0 => None,
156                v => Some(v),
157            },
158            connection: self.connection(),
159        }
160    }
161}
162
163/// Resident set size of this process, if the platform exposes it cheaply.
164pub fn process_rss_bytes() -> Option<u64> {
165    #[cfg(target_os = "linux")]
166    {
167        let statm = std::fs::read_to_string("/proc/self/statm").ok()?;
168        let pages: u64 = statm.split_whitespace().nth(1)?.parse().ok()?;
169        return Some(pages * 4096);
170    }
171    #[cfg(target_os = "macos")]
172    {
173        // `ps` is cheap enough for periodic status output and avoids an FFI dependency.
174        let out = std::process::Command::new("ps")
175            .args(["-o", "rss=", "-p", &std::process::id().to_string()])
176            .output()
177            .ok()?;
178        let kb: u64 = String::from_utf8_lossy(&out.stdout).trim().parse().ok()?;
179        return Some(kb * 1024);
180    }
181    #[allow(unreachable_code)]
182    None
183}
184
185#[cfg(test)]
186mod tests {
187    use super::*;
188
189    #[test]
190    fn snapshot_reflects_counters_and_lag() {
191        let m = Metrics::new();
192        m.set_state(JobState::Streaming);
193        m.add(&m.txns_applied, 3);
194        m.set_lsn(&m.server_wal_end, Lsn(1000));
195        m.set_lsn(&m.committed_lsn, Lsn(400));
196        m.set_lsn(&m.committed_lsn, Lsn(300)); // never regresses
197        let s = m.snapshot();
198        assert_eq!(s.state, JobState::Streaming);
199        assert_eq!(s.txns_applied, 3);
200        assert_eq!(s.committed_lsn, Lsn(400));
201        assert_eq!(s.apply_lag_bytes, 600);
202        // Keepalive-acknowledged WAL (nothing for this publication) is not lag.
203        m.set_lsn(&m.acked_lsn, Lsn(900));
204        assert_eq!(m.snapshot().apply_lag_bytes, 100);
205        let json = serde_json::to_value(&s).unwrap();
206        assert_eq!(json["state"], "STREAMING");
207    }
208
209    #[test]
210    fn rss_is_reported_on_supported_platforms() {
211        if cfg!(any(target_os = "linux", target_os = "macos")) {
212            assert!(process_rss_bytes().unwrap() > 0);
213        }
214    }
215}
pgsync_core/metrics.rs

pgsync_core/
metrics.rs
