rupg_func/
activity.rs

1//! The table of the sessions, from `backend_status.c` and `pgstatfuncs.c`: the rows of `pg_stat_get_activity`, which the views `pg_stat_activity`, `pg_stat_ssl` and `pg_stat_gssapi` read, the functions `pg_stat_get_backend_*` of one session, `pg_stat_get_db_numbackends`, and the addresses of the connection of `inet_client_addr` and its sibling functions.
2//!
3//! The session gives the table as [`Session::backends`]. The first call in a transaction takes a snapshot of the table, and the later calls of the transaction read the same snapshot, as `pgstat_read_current_status` does. `pg_stat_clear_snapshot` drops it.
4//!
5//! A row of another role shows less: the columns that `HAS_PGSTAT_PERMISSIONS` guards are null, and the text of the statement is `<insufficient privilege>`. `backend_xid` and `backend_xmin` show for each row, as in PostgreSQL. rupg has no parallel workers, no GSSAPI and no query IDs yet, so `leader_pid` and `query_id` are null, and the GSSAPI columns show a session with no GSSAPI.
6
7use std::net::SocketAddr;
8
9use rupg_common::Result;
10use rupg_types::{Value, inet_in, numeric_in};
11
12use crate::{Call, Kernel, Session, acl, bad_value, type_error};
13
14/// `GetBackendTypeDesc(B_BACKEND)`. Each session of rupg is a client backend.
15const CLIENT_BACKEND: &str = "client backend";
16
17/// The text of the functions of one session for a number that has no session.
18const NOT_AVAILABLE: &str = "<backend information not available>";
19
20/// The text of the statement of a session that the current user cannot see.
21const INSUFFICIENT: &str = "<insufficient privilege>";
22
23/// The state of a session, `BackendState` of `backend_status.h`.
24#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
25pub enum BackendState {
26    /// The session started, and did not send `ReadyForQuery` yet.
27    #[default]
28    Starting,
29    /// The session waits for a statement.
30    Idle,
31    /// The session runs a statement.
32    Running,
33    /// The session waits for a statement in a transaction block.
34    IdleInTransaction,
35    /// The session waits for the end of a failed transaction block.
36    IdleInTransactionAborted,
37    /// `track_activities` is off, so the session does not report its state.
38    Disabled,
39}
40
41impl BackendState {
42    /// The text of the column `state`.
43    fn text(self) -> &'static str {
44        match self {
45            BackendState::Starting => "starting",
46            BackendState::Idle => "idle",
47            BackendState::Running => "active",
48            BackendState::IdleInTransaction => "idle in transaction",
49            BackendState::IdleInTransactionAborted => "idle in transaction (aborted)",
50            BackendState::Disabled => "disabled",
51        }
52    }
53}
54
55/// An end of the connection of a session.
56#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
57pub enum Client {
58    /// The address is not known: the columns of the address are null.
59    #[default]
60    Unknown,
61    /// A Unix socket. It has no address, and `pg_stat_activity` shows the port -1.
62    Local,
63    /// A TCP connection.
64    Inet(SocketAddr),
65}
66
67impl Client {
68    /// The address as `inet`, for a TCP connection only. PostgreSQL makes the value from the text of `getnameinfo`, which is the text of the address with no zone.
69    fn addr(self) -> Result<Value> {
70        let Client::Inet(addr) = self else { return Ok(Value::Null) };
71        inet_in(&addr.ip().to_string(), false).map(Value::Inet).map_err(type_error)
72    }
73
74    /// The port, for a TCP connection only.
75    fn port(self) -> Value {
76        match self {
77            Client::Inet(addr) => Value::Int4(i32::from(addr.port())),
78            _ => Value::Null,
79        }
80    }
81
82    /// The port as `pg_stat_activity` shows it: -1 for a Unix socket.
83    fn shown_port(self) -> Value {
84        match self {
85            Client::Local => Value::Int4(-1),
86            _ => self.port(),
87        }
88    }
89}
90
91/// The TLS of a session, `PgBackendSSLStatus`. An empty text of the client certificate is null, as for a session with no client certificate.
92#[derive(Clone, Debug, Default, PartialEq, Eq)]
93pub struct Ssl {
94    /// The protocol, as `SSL_get_version` gives it, for example `TLSv1.3`.
95    pub version: String,
96    /// The cipher suite, by its name in OpenSSL.
97    pub cipher: String,
98    /// The bits of the key of the cipher.
99    pub bits: i32,
100    /// The subject of the client certificate, in the form of `X509_NAME_to_cstring`.
101    pub client_dn: String,
102    /// The serial number of the client certificate, in decimal.
103    pub client_serial: String,
104    /// The issuer of the client certificate, in the form of `X509_NAME_to_cstring`.
105    pub issuer_dn: String,
106}
107
108/// A row of the table of the sessions, `PgBackendStatus`.
109#[derive(Clone, Debug, Default)]
110pub struct Backend {
111    /// The number of the session in the table, `ProcNumber`, which `pg_stat_get_backend_idset` gives.
112    pub number: i32,
113    pub pid: i32,
114    /// The OID of the database.
115    pub database: u32,
116    /// The OID of the role of the session.
117    pub role: u32,
118    pub application: String,
119    pub state: BackendState,
120    /// The text of the last statement, as at most `track_activity_query_size - 1` bytes.
121    pub query: String,
122    /// The times in microseconds since 2000-01-01 UTC. `None` is null.
123    pub backend_start: i64,
124    pub transaction_start: Option<i64>,
125    pub query_start: Option<i64>,
126    pub state_change: Option<i64>,
127    /// True when the session waits for a message of the client, the wait event `ClientRead`.
128    pub reading: bool,
129    /// The type and the name of the wait event of a wait in a statement, such as `Timeout` and `PgSleep` in `pg_sleep`.
130    pub waiting: Option<(&'static str, &'static str)>,
131    pub client: Client,
132    /// The TLS of the session, or `None` for a session with no TLS.
133    pub ssl: Option<Ssl>,
134    /// `backend_xid` and `backend_xmin`: the transaction ID of the session and the `xmin` of its oldest snapshot, with 0 for none.
135    pub xid: u32,
136    pub xmin: u32,
137}
138
139/// `HAS_PGSTAT_PERMISSIONS`: the current user has the privileges of `pg_read_all_stats` or of the role of the session.
140fn permitted(session: &dyn Session, role: u32) -> bool {
141    acl::has_pgstat_permissions(session, role)
142}
143
144/// A time of a row, or null.
145fn time(at: Option<i64>) -> Value {
146    at.map_or(Value::Null, Value::TimestampTz)
147}
148
149/// A text value, or null.
150fn text(value: Option<&str>) -> Value {
151    value.map_or(Value::Null, Value::text)
152}
153
154/// A text of the TLS status that is null when it is empty.
155fn filled(value: &str) -> Value {
156    if value.is_empty() { Value::Null } else { Value::text(value) }
157}
158
159/// An OID that is null when it is 0.
160fn valid_oid(oid: u32) -> Value {
161    if oid == 0 { Value::Null } else { Value::Oid(oid) }
162}
163
164/// The wait event of a row as its type and its name.
165fn wait_event(backend: &Backend) -> (Option<&'static str>, Option<&'static str>) {
166    match backend.waiting {
167        Some((kind, name)) => (Some(kind), Some(name)),
168        None if backend.reading => (Some("Client"), Some("ClientRead")),
169        None => (None, None),
170    }
171}
172
173/// The 31 columns of `pg_stat_get_activity` for a row.
174fn activity_row(session: &dyn Session, backend: &Backend) -> Result<Vec<Value>> {
175    let mut row = vec![Value::Null; 31];
176    row[0] = valid_oid(backend.database);
177    row[1] = Value::Int4(backend.pid);
178    row[2] = valid_oid(backend.role);
179    row[3] = Value::text(&backend.application);
180    row[15] = valid_oid(backend.xid);
181    row[16] = valid_oid(backend.xmin);
182    if !permitted(session, backend.role) {
183        row[5] = Value::text(INSUFFICIENT);
184        return Ok(row);
185    }
186    row[4] = Value::text(backend.state.text());
187    row[5] = Value::text(&backend.query);
188    let (kind, event) = wait_event(backend);
189    row[6] = text(kind);
190    row[7] = text(event);
191    row[8] = time(backend.transaction_start);
192    row[9] = time(backend.query_start);
193    row[10] = Value::TimestampTz(backend.backend_start);
194    row[11] = time(backend.state_change);
195    row[12] = backend.client.addr()?;
196    row[14] = backend.client.shown_port();
197    row[17] = Value::text(CLIENT_BACKEND);
198    row[18] = Value::Bool(backend.ssl.is_some());
199    if let Some(ssl) = &backend.ssl {
200        row[19] = Value::text(&ssl.version);
201        row[20] = Value::text(&ssl.cipher);
202        row[21] = Value::Int4(ssl.bits);
203        row[22] = filled(&ssl.client_dn);
204        if !ssl.client_serial.is_empty() {
205            row[23] = Value::Numeric(numeric_in(&ssl.client_serial, -1).map_err(type_error)?);
206        }
207        row[24] = filled(&ssl.issuer_dn);
208    }
209    row[25] = Value::Bool(false);
210    row[27] = Value::Bool(false);
211    row[28] = Value::Bool(false);
212    Ok(row)
213}
214
215/// `pg_stat_get_activity`: a row for each session, or for the session with the process ID of the argument. A null argument gives all the sessions.
216pub(crate) fn activity(call: &Call<'_>, args: &[Value]) -> Result<Vec<Vec<Value>>> {
217    let pid = match args.first() {
218        Some(Value::Int4(pid)) => Some(*pid),
219        Some(Value::Null) | None => None,
220        Some(_) => return Err(bad_value()),
221    };
222    let backends = call.session.backends();
223    let mut rows = Vec::new();
224    for backend in backends.iter().filter(|b| pid.is_none_or(|pid| b.pid == pid)) {
225        rows.push(activity_row(call.session, backend)?);
226        if pid.is_some() {
227            break;
228        }
229    }
230    Ok(rows)
231}
232
233/// `pg_stat_get_backend_idset`: the number of each session.
234pub(crate) fn backend_idset(call: &Call<'_>, _: &[Value]) -> Result<Vec<Vec<Value>>> {
235    Ok(call.session.backends().iter().map(|b| vec![Value::Int4(b.number)]).collect())
236}
237
238/// `pgstat_get_beentry_by_proc_number`: the row with the number of the argument, or `None`.
239fn backend<T>(
240    call: &Call<'_>,
241    args: &[Value],
242    read: impl FnOnce(&Backend) -> T,
243) -> Result<Option<T>> {
244    let Some(Value::Int4(number)) = args.first() else { return Err(bad_value()) };
245    let backends = call.session.backends();
246    Ok(backends.iter().find(|b| b.number == *number).map(read))
247}
248
249/// A value of a row that only a permitted user sees, or null.
250fn guarded(
251    call: &Call<'_>,
252    args: &[Value],
253    read: impl FnOnce(&Backend) -> Result<Value>,
254) -> Result<Value> {
255    match backend(call, args, |b| permitted(call.session, b.role).then(|| read(b)))? {
256        Some(Some(value)) => value,
257        _ => Ok(Value::Null),
258    }
259}
260
261fn backend_pid(call: &Call<'_>, args: &[Value]) -> Result<Value> {
262    Ok(backend(call, args, |b| Value::Int4(b.pid))?.unwrap_or(Value::Null))
263}
264
265fn backend_dbid(call: &Call<'_>, args: &[Value]) -> Result<Value> {
266    Ok(backend(call, args, |b| Value::Oid(b.database))?.unwrap_or(Value::Null))
267}
268
269fn backend_userid(call: &Call<'_>, args: &[Value]) -> Result<Value> {
270    Ok(backend(call, args, |b| Value::Oid(b.role))?.unwrap_or(Value::Null))
271}
272
273/// `pg_stat_get_backend_activity`: the text of the statement, or a text that tells why it is not there.
274fn backend_activity(call: &Call<'_>, args: &[Value]) -> Result<Value> {
275    let found = backend(call, args, |b| {
276        if !permitted(call.session, b.role) {
277            INSUFFICIENT.to_owned()
278        } else if b.query.is_empty() {
279            "<command string not enabled>".to_owned()
280        } else {
281            b.query.clone()
282        }
283    })?;
284    Ok(Value::text(found.as_deref().unwrap_or(NOT_AVAILABLE)))
285}
286
287/// `pg_stat_get_backend_wait_event_type` and `pg_stat_get_backend_wait_event`.
288fn backend_wait(call: &Call<'_>, args: &[Value], name: bool) -> Result<Value> {
289    let found = backend(call, args, |b| {
290        if !permitted(call.session, b.role) {
291            return Some(INSUFFICIENT);
292        }
293        let (kind, event) = wait_event(b);
294        if name { event } else { kind }
295    })?;
296    Ok(text(found.unwrap_or(Some(NOT_AVAILABLE))))
297}
298
299fn backend_wait_event_type(call: &Call<'_>, args: &[Value]) -> Result<Value> {
300    backend_wait(call, args, false)
301}
302
303fn backend_wait_event(call: &Call<'_>, args: &[Value]) -> Result<Value> {
304    backend_wait(call, args, true)
305}
306
307fn backend_activity_start(call: &Call<'_>, args: &[Value]) -> Result<Value> {
308    guarded(call, args, |b| Ok(time(b.query_start)))
309}
310
311fn backend_xact_start(call: &Call<'_>, args: &[Value]) -> Result<Value> {
312    guarded(call, args, |b| Ok(time(b.transaction_start)))
313}
314
315fn backend_start(call: &Call<'_>, args: &[Value]) -> Result<Value> {
316    guarded(call, args, |b| Ok(Value::TimestampTz(b.backend_start)))
317}
318
319fn backend_client_addr(call: &Call<'_>, args: &[Value]) -> Result<Value> {
320    guarded(call, args, |b| b.client.addr())
321}
322
323fn backend_client_port(call: &Call<'_>, args: &[Value]) -> Result<Value> {
324    guarded(call, args, |b| Ok(b.client.shown_port()))
325}
326
327/// `pg_stat_get_db_numbackends`: the number of sessions in the database.
328fn db_numbackends(call: &Call<'_>, args: &[Value]) -> Result<Value> {
329    let database = args.first().and_then(Value::as_oid).ok_or_else(bad_value)?;
330    let count = call.session.backends().iter().filter(|b| b.database == database).count();
331    Ok(Value::Int4(i32::try_from(count).unwrap_or(i32::MAX)))
332}
333
334/// `pg_stat_clear_snapshot`: the next read of the table takes a new snapshot. The result is `void`, whose value the engine keeps as an empty text.
335fn clear_snapshot(call: &Call<'_>, _: &[Value]) -> Result<Value> {
336    call.session.clear_backends();
337    Ok(Value::text(""))
338}
339
340fn client_addr(call: &Call<'_>, _: &[Value]) -> Result<Value> {
341    call.session.addresses().0.addr()
342}
343
344fn client_port(call: &Call<'_>, _: &[Value]) -> Result<Value> {
345    Ok(call.session.addresses().0.port())
346}
347
348fn server_addr(call: &Call<'_>, _: &[Value]) -> Result<Value> {
349    call.session.addresses().1.addr()
350}
351
352fn server_port(call: &Call<'_>, _: &[Value]) -> Result<Value> {
353    Ok(call.session.addresses().1.port())
354}
355
356/// The kernel of a function of this module by its `prosrc`.
357pub(crate) fn by_src(src: &str) -> Option<Kernel> {
358    Some(match src {
359        "pg_stat_get_backend_pid" => backend_pid,
360        "pg_stat_get_backend_dbid" => backend_dbid,
361        "pg_stat_get_backend_userid" => backend_userid,
362        "pg_stat_get_backend_activity" => backend_activity,
363        "pg_stat_get_backend_wait_event_type" => backend_wait_event_type,
364        "pg_stat_get_backend_wait_event" => backend_wait_event,
365        "pg_stat_get_backend_activity_start" => backend_activity_start,
366        "pg_stat_get_backend_xact_start" => backend_xact_start,
367        "pg_stat_get_backend_start" => backend_start,
368        "pg_stat_get_backend_client_addr" => backend_client_addr,
369        "pg_stat_get_backend_client_port" => backend_client_port,
370        "pg_stat_get_db_numbackends" => db_numbackends,
371        "pg_stat_clear_snapshot" => clear_snapshot,
372        "inet_client_addr" => client_addr,
373        "inet_client_port" => client_port,
374        "inet_server_addr" => server_addr,
375        "inet_server_port" => server_port,
376        _ => return None,
377    })
378}
rupg_func/activity.rs

rupg_func/
activity.rs
