use crate::collectors::Collector;
use crate::collectors::util::{
get_default_database, get_excluded_databases, get_or_create_pool_for_db,
};
use anyhow::Result;
use futures::future::BoxFuture;
use prometheus::{IntGaugeVec, Opts, Registry};
use sqlx::{PgPool, Row};
use std::sync::Arc;
use tokio::task::JoinSet;
use tracing::{debug, error, info_span, instrument};
use tracing_futures::Instrument as _;
#[derive(Clone)]
pub struct StatUserTablesCollector {
seq_scan: IntGaugeVec,
seq_tup_read: IntGaugeVec,
idx_scan: IntGaugeVec,
idx_tup_fetch: IntGaugeVec,
n_tup_ins: IntGaugeVec,
n_tup_upd: IntGaugeVec,
n_tup_del: IntGaugeVec,
n_tup_hot_upd: IntGaugeVec,
n_live_tup: IntGaugeVec,
n_dead_tup: IntGaugeVec,
n_mod_since_analyze: IntGaugeVec,
last_vacuum: IntGaugeVec,
last_autovacuum: IntGaugeVec,
last_analyze: IntGaugeVec,
last_autoanalyze: IntGaugeVec,
vacuum_count: IntGaugeVec,
autovacuum_count: IntGaugeVec,
analyze_count: IntGaugeVec,
autoanalyze_count: IntGaugeVec,
index_size_bytes: IntGaugeVec,
table_size_bytes: IntGaugeVec,
}
impl Default for StatUserTablesCollector {
fn default() -> Self {
Self::new()
}
}
impl StatUserTablesCollector {
pub fn new() -> Self {
let labels = &["datname", "schemaname", "relname"];
let seq_scan = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_seq_scan",
"Number of sequential scans initiated on this table",
),
labels,
)
.expect("pg_stat_user_tables_seq_scan");
let seq_tup_read = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_seq_tup_read",
"Number of live rows fetched by sequential scans",
),
labels,
)
.expect("pg_stat_user_tables_seq_tup_read");
let idx_scan = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_idx_scan",
"Number of index scans initiated on this table",
),
labels,
)
.expect("pg_stat_user_tables_idx_scan");
let idx_tup_fetch = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_idx_tup_fetch",
"Number of live rows fetched by index scans",
),
labels,
)
.expect("pg_stat_user_tables_idx_tup_fetch");
let n_tup_ins = IntGaugeVec::new(
Opts::new("pg_stat_user_tables_n_tup_ins", "Number of rows inserted"),
labels,
)
.expect("pg_stat_user_tables_n_tup_ins");
let n_tup_upd = IntGaugeVec::new(
Opts::new("pg_stat_user_tables_n_tup_upd", "Number of rows updated"),
labels,
)
.expect("pg_stat_user_tables_n_tup_upd");
let n_tup_del = IntGaugeVec::new(
Opts::new("pg_stat_user_tables_n_tup_del", "Number of rows deleted"),
labels,
)
.expect("pg_stat_user_tables_n_tup_del");
let n_tup_hot_upd = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_n_tup_hot_upd",
"Number of rows HOT updated",
),
labels,
)
.expect("pg_stat_user_tables_n_tup_hot_upd");
let n_live_tup = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_n_live_tup",
"Estimated number of live rows",
),
labels,
)
.expect("pg_stat_user_tables_n_live_tup");
let n_dead_tup = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_n_dead_tup",
"Estimated number of dead rows",
),
labels,
)
.expect("pg_stat_user_tables_n_dead_tup");
let n_mod_since_analyze = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_n_mod_since_analyze",
"Estimated number of rows changed since last analyze",
),
labels,
)
.expect("pg_stat_user_tables_n_mod_since_analyze");
let last_vacuum = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_last_vacuum",
"Last manual vacuum time (epoch seconds)",
),
labels,
)
.expect("pg_stat_user_tables_last_vacuum");
let last_autovacuum = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_last_autovacuum",
"Last autovacuum time (epoch seconds)",
),
labels,
)
.expect("pg_stat_user_tables_last_autovacuum");
let last_analyze = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_last_analyze",
"Last manual analyze time (epoch seconds)",
),
labels,
)
.expect("pg_stat_user_tables_last_analyze");
let last_autoanalyze = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_last_autoanalyze",
"Last autoanalyze time (epoch seconds)",
),
labels,
)
.expect("pg_stat_user_tables_last_autoanalyze");
let vacuum_count = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_vacuum_count",
"Number of times manually vacuumed",
),
labels,
)
.expect("pg_stat_user_tables_vacuum_count");
let autovacuum_count = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_autovacuum_count",
"Number of times vacuumed by autovacuum",
),
labels,
)
.expect("pg_stat_user_tables_autovacuum_count");
let analyze_count = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_analyze_count",
"Number of times manually analyzed",
),
labels,
)
.expect("pg_stat_user_tables_analyze_count");
let autoanalyze_count = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_autoanalyze_count",
"Number of times analyzed by autovacuum",
),
labels,
)
.expect("pg_stat_user_tables_autoanalyze_count");
let index_size_bytes = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_index_size_bytes",
"Total disk space used by indexes on this table, in bytes",
),
labels,
)
.expect("pg_stat_user_tables_index_size_bytes");
let table_size_bytes = IntGaugeVec::new(
Opts::new(
"pg_stat_user_tables_table_size_bytes",
"Total disk space used by this table, in bytes",
),
labels,
)
.expect("pg_stat_user_tables_table_size_bytes");
Self {
seq_scan,
seq_tup_read,
idx_scan,
idx_tup_fetch,
n_tup_ins,
n_tup_upd,
n_tup_del,
n_tup_hot_upd,
n_live_tup,
n_dead_tup,
n_mod_since_analyze,
last_vacuum,
last_autovacuum,
last_analyze,
last_autoanalyze,
vacuum_count,
autovacuum_count,
analyze_count,
autoanalyze_count,
index_size_bytes,
table_size_bytes,
}
}
}
impl Collector for StatUserTablesCollector {
fn name(&self) -> &'static str {
"stat_user_tables"
}
fn register_metrics(&self, registry: &Registry) -> Result<()> {
registry.register(Box::new(self.seq_scan.clone()))?;
registry.register(Box::new(self.seq_tup_read.clone()))?;
registry.register(Box::new(self.idx_scan.clone()))?;
registry.register(Box::new(self.idx_tup_fetch.clone()))?;
registry.register(Box::new(self.n_tup_ins.clone()))?;
registry.register(Box::new(self.n_tup_upd.clone()))?;
registry.register(Box::new(self.n_tup_del.clone()))?;
registry.register(Box::new(self.n_tup_hot_upd.clone()))?;
registry.register(Box::new(self.n_live_tup.clone()))?;
registry.register(Box::new(self.n_dead_tup.clone()))?;
registry.register(Box::new(self.n_mod_since_analyze.clone()))?;
registry.register(Box::new(self.last_vacuum.clone()))?;
registry.register(Box::new(self.last_autovacuum.clone()))?;
registry.register(Box::new(self.last_analyze.clone()))?;
registry.register(Box::new(self.last_autoanalyze.clone()))?;
registry.register(Box::new(self.vacuum_count.clone()))?;
registry.register(Box::new(self.autovacuum_count.clone()))?;
registry.register(Box::new(self.analyze_count.clone()))?;
registry.register(Box::new(self.autoanalyze_count.clone()))?;
registry.register(Box::new(self.index_size_bytes.clone()))?;
registry.register(Box::new(self.table_size_bytes.clone()))?;
Ok(())
}
#[instrument(skip(self, pool), level = "info", err, fields(collector="stat_user_tables", otel.kind="internal"))]
fn collect<'a>(&'a self, pool: &'a PgPool) -> BoxFuture<'a, Result<()>> {
Box::pin(async move {
let excluded = get_excluded_databases().to_vec();
let db_list_span = info_span!(
"db.query",
otel.kind = "client",
db.system = "postgresql",
db.operation = "SELECT",
db.statement = "SELECT datname FROM pg_database WHERE datallowconn ...",
db.sql.table = "pg_database"
);
let dbs: Vec<String> = sqlx::query_scalar(
r#"
SELECT datname
FROM pg_database
WHERE datallowconn
AND NOT datistemplate
AND NOT (datname = ANY($1))
ORDER BY datname
"#,
)
.bind(&excluded)
.fetch_all(pool)
.instrument(db_list_span)
.await?;
let shared_pool = pool.clone();
let default_db = get_default_database().map(|s| s.to_string());
let this = Arc::new(self.clone());
let mut tasks = JoinSet::new();
for datname in dbs {
let this = Arc::clone(&this);
let shared_pool = shared_pool.clone();
let default_db = default_db.clone();
tasks.spawn(async move {
let use_shared = default_db.as_deref() == Some(datname.as_str());
let query_span = info_span!(
"db.query",
otel.kind = "client",
db.system = "postgresql",
db.operation = "SELECT",
db.statement = "SELECT ... FROM pg_stat_user_tables",
db.sql.table = "pg_stat_user_tables",
datname = %datname,
reuse_pool = use_shared
);
let rows_res = if use_shared {
sqlx::query(
r#"
SELECT
current_database() AS datname,
schemaname,
relname,
seq_scan::bigint,
seq_tup_read::bigint,
idx_scan::bigint,
idx_tup_fetch::bigint,
n_tup_ins::bigint,
n_tup_upd::bigint,
n_tup_del::bigint,
n_tup_hot_upd::bigint,
n_live_tup::bigint,
n_dead_tup::bigint,
n_mod_since_analyze::bigint,
COALESCE(EXTRACT(EPOCH FROM last_vacuum)::bigint, 0) AS last_vacuum_epoch,
COALESCE(EXTRACT(EPOCH FROM last_autovacuum)::bigint, 0) AS last_autovacuum_epoch,
COALESCE(EXTRACT(EPOCH FROM last_analyze)::bigint, 0) AS last_analyze_epoch,
COALESCE(EXTRACT(EPOCH FROM last_autoanalyze)::bigint, 0) AS last_autoanalyze_epoch,
vacuum_count::bigint,
autovacuum_count::bigint,
analyze_count::bigint,
autoanalyze_count::bigint,
pg_indexes_size(relid)::bigint AS index_size_bytes,
pg_table_size(relid)::bigint AS table_size_bytes
FROM pg_stat_user_tables
"#,
)
.fetch_all(&shared_pool)
.instrument(query_span)
.await
} else {
match get_or_create_pool_for_db(&datname).await {
Ok(per_db_pool) => {
sqlx::query(
r#"
SELECT
current_database() AS datname,
schemaname,
relname,
seq_scan::bigint,
seq_tup_read::bigint,
idx_scan::bigint,
idx_tup_fetch::bigint,
n_tup_ins::bigint,
n_tup_upd::bigint,
n_tup_del::bigint,
n_tup_hot_upd::bigint,
n_live_tup::bigint,
n_dead_tup::bigint,
n_mod_since_analyze::bigint,
COALESCE(EXTRACT(EPOCH FROM last_vacuum)::bigint, 0) AS last_vacuum_epoch,
COALESCE(EXTRACT(EPOCH FROM last_autovacuum)::bigint, 0) AS last_autovacuum_epoch,
COALESCE(EXTRACT(EPOCH FROM last_analyze)::bigint, 0) AS last_analyze_epoch,
COALESCE(EXTRACT(EPOCH FROM last_autoanalyze)::bigint, 0) AS last_autoanalyze_epoch,
vacuum_count::bigint,
autovacuum_count::bigint,
analyze_count::bigint,
autoanalyze_count::bigint,
pg_indexes_size(relid)::bigint AS index_size_bytes,
pg_table_size(relid)::bigint AS table_size_bytes
FROM pg_stat_user_tables
"#,
)
.fetch_all(&per_db_pool)
.instrument(query_span)
.await
}
Err(e) => {
error!(%datname, error=?e, "stat_user_tables: pool init failed");
return Ok::<(), anyhow::Error>(());
}
}
};
let rows = match rows_res {
Ok(r) => r,
Err(e) => {
error!(%datname, error=?e, "stat_user_tables: query failed");
return Ok(());
}
};
for row in rows {
let dat: String = row.try_get::<Option<String>, _>("datname")?.unwrap_or_else(|| "[unknown]".to_string());
let schema: String = row.try_get("schemaname")?;
let table: String = row.try_get("relname")?;
let labels = [&dat, &schema, &table];
this.seq_scan.with_label_values(&labels).set(row.try_get::<i64, _>("seq_scan").unwrap_or(0));
this.seq_tup_read.with_label_values(&labels).set(row.try_get::<i64, _>("seq_tup_read").unwrap_or(0));
this.idx_scan.with_label_values(&labels).set(row.try_get::<i64, _>("idx_scan").unwrap_or(0));
this.idx_tup_fetch.with_label_values(&labels).set(row.try_get::<i64, _>("idx_tup_fetch").unwrap_or(0));
this.n_tup_ins.with_label_values(&labels).set(row.try_get::<i64, _>("n_tup_ins").unwrap_or(0));
this.n_tup_upd.with_label_values(&labels).set(row.try_get::<i64, _>("n_tup_upd").unwrap_or(0));
this.n_tup_del.with_label_values(&labels).set(row.try_get::<i64, _>("n_tup_del").unwrap_or(0));
this.n_tup_hot_upd.with_label_values(&labels).set(row.try_get::<i64, _>("n_tup_hot_upd").unwrap_or(0));
this.n_live_tup.with_label_values(&labels).set(row.try_get::<i64, _>("n_live_tup").unwrap_or(0));
this.n_dead_tup.with_label_values(&labels).set(row.try_get::<i64, _>("n_dead_tup").unwrap_or(0));
this.n_mod_since_analyze.with_label_values(&labels).set(row.try_get::<i64, _>("n_mod_since_analyze").unwrap_or(0));
this.last_vacuum.with_label_values(&labels).set(row.try_get::<i64, _>("last_vacuum_epoch").unwrap_or(0));
this.last_autovacuum.with_label_values(&labels).set(row.try_get::<i64, _>("last_autovacuum_epoch").unwrap_or(0));
this.last_analyze.with_label_values(&labels).set(row.try_get::<i64, _>("last_analyze_epoch").unwrap_or(0));
this.last_autoanalyze.with_label_values(&labels).set(row.try_get::<i64, _>("last_autoanalyze_epoch").unwrap_or(0));
this.vacuum_count.with_label_values(&labels).set(row.try_get::<i64, _>("vacuum_count").unwrap_or(0));
this.autovacuum_count.with_label_values(&labels).set(row.try_get::<i64, _>("autovacuum_count").unwrap_or(0));
this.analyze_count.with_label_values(&labels).set(row.try_get::<i64, _>("analyze_count").unwrap_or(0));
this.autoanalyze_count.with_label_values(&labels).set(row.try_get::<i64, _>("autoanalyze_count").unwrap_or(0));
this.index_size_bytes.with_label_values(&labels).set(row.try_get::<i64, _>("index_size_bytes").unwrap_or(0));
this.table_size_bytes.with_label_values(&labels).set(row.try_get::<i64, _>("table_size_bytes").unwrap_or(0));
debug!(datname=%dat, schema=%schema, table=%table, "updated pg_stat_user_tables metrics");
}
Ok::<(), anyhow::Error>(())
});
}
while let Some(res) = tasks.join_next().await {
if let Err(e) = res {
error!(error=?e, "stat_user_tables: task join error");
} else if let Ok(Err(e)) = res {
error!(error=?e, "stat_user_tables: task returned error");
}
}
Ok(())
})
}
}