diff --git a/compute_tools/src/compute.rs b/compute_tools/src/compute.rs index c7b4bdd240..b861bd99fc 100644 --- a/compute_tools/src/compute.rs +++ b/compute_tools/src/compute.rs @@ -11,7 +11,7 @@ use std::{env, fs}; use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use compute_api::privilege::Privilege; -use compute_api::responses::{ComputeConfig, ComputeCtlConfig, ComputeMetrics, ComputeStatus}; +use compute_api::responses::{ActivityKind, ComputeConfig, ComputeCtlConfig, ComputeMetrics, ComputeStatus}; use compute_api::spec::{ ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PgIdent, }; @@ -132,6 +132,10 @@ pub struct ComputeState { /// Timestamp of the last Postgres activity. It could be `None` if /// compute wasn't used since start. pub last_active: Option>, + /// Timestamp of the last client's activity. Unlike `last_time` it doesn't take in account + /// baclkground activity: autovacuum, LR,... + pub last_active_query: Option>, + pub last_activity_kind: Option, pub error: Option, /// Compute spec. This can be received from the CLI or - more likely - @@ -159,6 +163,8 @@ impl ComputeState { start_time: Utc::now(), status: ComputeStatus::Empty, last_active: None, + last_active_query: None, + last_activity_kind: None, error: None, pspec: None, startup_span: None, @@ -1688,13 +1694,22 @@ impl ComputeNode { } /// Update the `last_active` in the shared state, but ensure that it's a more recent one. - pub fn update_last_active(&self, last_active: Option>) { + pub fn update_last_active( + &self, + last_active: Option>, + activity_kind: ActivityKind, + ) { let mut state = self.state.lock().unwrap(); // NB: `Some()` is always greater than `None`. if last_active > state.last_active { state.last_active = last_active; debug!("set the last compute activity time to: {:?}", last_active); } + if activity_kind == ActivityKind::Query && last_active > state.last_active_query { + state.last_active_query = last_active; + debug!("set the last user's activity time to: {:?}", last_active); + } + state.last_activity_kind = Some(activity_kind); } // Look for core dumps and collect backtraces. diff --git a/compute_tools/src/http/routes/mod.rs b/compute_tools/src/http/routes/mod.rs index a67be7fd5a..8220558735 100644 --- a/compute_tools/src/http/routes/mod.rs +++ b/compute_tools/src/http/routes/mod.rs @@ -30,6 +30,8 @@ impl From<&ComputeState> for ComputeStatusResponse { .map(|pspec| pspec.timeline_id.to_string()), status: state.status, last_active: state.last_active, + last_active_query: state.last_active_query, + last_activity_kind: state.last_activity_kind, error: state.error.clone(), } } diff --git a/compute_tools/src/monitor.rs b/compute_tools/src/monitor.rs index 83318538cd..0f27fde656 100644 --- a/compute_tools/src/monitor.rs +++ b/compute_tools/src/monitor.rs @@ -3,7 +3,7 @@ use std::thread; use std::time::Duration; use chrono::{DateTime, Utc}; -use compute_api::responses::ComputeStatus; +use compute_api::responses::{ActivityKind, ComputeStatus}; use compute_api::spec::ComputeFeature; use postgres::{Client, NoTls}; use tracing::{debug, error, info, warn}; @@ -91,7 +91,7 @@ fn watch_compute_activity(compute: &ComputeNode) { if detected_activity { // Update the last active time and continue, we don't need to // check backends state change. - compute.update_last_active(Some(Utc::now())); + compute.update_last_active(Some(Utc::now()), ActivityKind::Query); continue; } } @@ -109,7 +109,7 @@ fn watch_compute_activity(compute: &ComputeNode) { // This helps us to discover new sessions, that did nothing yet. match get_backends_state_change(cli) { Ok(last_active) => { - compute.update_last_active(last_active); + compute.update_last_active(last_active, ActivityKind::Query); } Err(e) => { error!("could not get backends state change: {}", e); @@ -125,7 +125,10 @@ fn watch_compute_activity(compute: &ComputeNode) { Ok(r) => match r.try_get::<&str, i64>("count") { Ok(num_ws) => { if num_ws > 0 { - compute.update_last_active(Some(Utc::now())); + compute.update_last_active( + Some(Utc::now()), + ActivityKind::LogicalReplication, + ); continue; } } @@ -150,7 +153,10 @@ fn watch_compute_activity(compute: &ComputeNode) { Ok(row) => match row.try_get::<&str, i64>("count") { Ok(num_subscribers) => { if num_subscribers > 0 { - compute.update_last_active(Some(Utc::now())); + compute.update_last_active( + Some(Utc::now()), + ActivityKind::LogicalReplication, + ); continue; } } @@ -175,7 +181,8 @@ fn watch_compute_activity(compute: &ComputeNode) { Ok(r) => match r.try_get::<&str, i64>("count") { Ok(num_workers) => { if num_workers > 0 { - compute.update_last_active(Some(Utc::now())); + compute + .update_last_active(Some(Utc::now()), ActivityKind::Autovacuum); continue; } } diff --git a/libs/compute_api/src/responses.rs b/libs/compute_api/src/responses.rs index b7d6b7ca34..47bb910b22 100644 --- a/libs/compute_api/src/responses.rs +++ b/libs/compute_api/src/responses.rs @@ -56,9 +56,22 @@ pub struct ComputeStatusResponse { pub status: ComputeStatus, #[serde(serialize_with = "rfc3339_serialize")] pub last_active: Option>, + pub last_active_query: Option>, + pub last_activity_kind: Option, pub error: Option, } +#[derive(Serialize, Clone, Copy, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ActivityKind { + // Client's query is executed + Query, + // Logical replication is active (subscription or publication) + LogicalReplication, + // Autovacuum is active + Autovacuum, +} + #[derive(Serialize, Clone, Copy, Debug, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum ComputeStatus {