jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Instrument database pool roles (#7356)

Browse Source 
## What

Name the relay's four Postgres pool roles—writer, reader, audit, and
search—and export one fixed-cardinality utilization metric contract for
all of them. Existing writer and reader gauges remain unchanged.

## Why

Audit and search already use separate pools, but they are missing from
pool pressure telemetry. Operators cannot see all process-level
connection demand before setting a deployment-wide connection budget.

## How

A closed `DbPoolRole` vocabulary now labels
`buzz_db_pool_connections{pool_role,state}` and
`buzz_db_pool_configured{pool_role}`. The relay retains cheap pool
clones for statistics only; ownership, capacities, timeouts, query
routing, and failure behavior do not change. The docs also replace the
stale async search-index description with the generated-column behavior
used today.

This simplifies four ad hoc observability states into one bounded model
without adding a pool manager or changing service boundaries.

## Risk

Low to moderate. This changes relay metrics and pool construction
plumbing, but not SQL execution or routing. Legacy metric names remain
available for current dashboards.

## Testing

- Push gate at `3f5468a0463bdfc5879bcdd3878fa06b42ca84c2`: all 14 Rust
test lanes passed; desktop Tauri checks passed.
- Blox: `cargo test --workspace --all-targets` with repository-declared
Postgres services.
- Blox: `cargo clippy --workspace --all-targets -- -D warnings`.
- Blox: `cargo fmt --all -- --check`.

## Bigger picture

Aggregate deployment budget enforcement remains deferred. This PR
exposes the per-role facts that a later deployment-policy change can
consume.

Generated with Claude Code

---------

Signed-off-by: tornquist <tornquist@squareup.com>
Co-authored-by: Codex <noreply@openai.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Luke Tornquist
2026-09-16 11:37:12 -04:00
committed by GitHub
parent 6d1f488d22
commit 213092074c
7 changed files with 470 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

288
crates/buzz-relay/src/metrics.rs
View File

@@ -255,6 +255,81 @@ pub(crate) fn describe_db_pool_metrics() {
"buzz_db_pool_waiters",
"Current tracked-operation database pool checkout attempts in progress by valid pool role and operation"
);
metrics::describe_gauge!(
"buzz_db_pool_connections",
"Current Postgres pool connections by physical pool role and bounded utilization state"
);
metrics::describe_gauge!(
"buzz_db_pool_max_connections",
"Maximum Postgres pool connections by physical pool role"
);
metrics::describe_gauge!(
"buzz_db_pool_configured",
"Whether a physical Postgres pool role is configured, where 1 is configured and 0 is not configured"
);
}
/// Named utilization snapshots for every physical Postgres pool role.
pub struct DbPoolMetricsInput {
/// Primary writer pool utilization.
pub writer: buzz_db::DbPoolStats,
/// Optional read-replica pool utilization.
pub reader: Option<buzz_db::DbPoolStats>,
/// Optional audit pool utilization.
pub audit: Option<buzz_db::DbPoolStats>,
/// Search pool utilization.
pub search: buzz_db::DbPoolStats,
}
/// Record fixed-cardinality utilization for every physical Postgres pool role.
///
/// Optional roles emit zero-valued utilization when absent so the unified
/// scrape always contains four roles by two states, four maximum-capacity
/// series, and four configured series. The original writer and reader gauges
/// are also emitted unchanged for dashboard compatibility.
pub fn record_db_pool_metrics(input: DbPoolMetricsInput) {
let DbPoolMetricsInput {
writer,
reader,
audit,
search,
} = input;
let pools = [Some(writer), reader, audit, Some(search)];
for (role, stats) in buzz_db::DbPoolRole::ALL.into_iter().zip(pools) {
let role = role.as_str();
let configured = stats.is_some();
let stats = stats.unwrap_or(buzz_db::DbPoolStats {
size: 0,
idle: 0,
max: 0,
});
metrics::gauge!("buzz_db_pool_configured", "pool_role" => role).set(if configured {
1.0
} else {
0.0
});
for (state, value) in [("idle", stats.idle), ("active", stats.active())] {
metrics::gauge!(
"buzz_db_pool_connections",
"pool_role" => role,
"state" => state,
)
.set(value as f64);
}
metrics::gauge!("buzz_db_pool_max_connections", "pool_role" => role).set(stats.max as f64);
}
metrics::gauge!("buzz_db_pool_size").set(writer.size as f64);
metrics::gauge!("buzz_db_pool_idle").set(writer.idle as f64);
metrics::gauge!("buzz_db_pool_active").set(writer.active() as f64);
metrics::gauge!("buzz_db_pool_max").set(writer.max as f64);
if let Some(reader) = reader {
metrics::gauge!("buzz_db_read_pool_size").set(reader.size as f64);
metrics::gauge!("buzz_db_read_pool_idle").set(reader.idle as f64);
metrics::gauge!("buzz_db_read_pool_active").set(reader.active() as f64);
metrics::gauge!("buzz_db_read_pool_max").set(reader.max as f64);
}
}
#[cfg(test)]
@@ -344,6 +419,219 @@ mod contract_tests {
.unwrap_or_default()
}
fn sample_value<'a>(scrape: &'a str, metric: &str, labels: &[(&str, &str)]) -> &'a str {
scrape
.lines()
.find(|line| {
line.starts_with(metric)
&& labels
.iter()
.all(|(key, value)| line.contains(&format!(r#"{key}="{value}""#)))
})
.and_then(|line| line.rsplit_once(' ').map(|(_, value)| value))
.unwrap_or_else(|| panic!("missing {metric} with {labels:?} in scrape:\n{scrape}"))
}
#[test]
fn production_pool_sampler_has_fixed_roles_utilization_and_compatibility_metrics() {
let (recorder, handle) = super::readiness_test_recorder();
metrics::with_local_recorder(&recorder, || {
super::describe_db_pool_metrics();
super::record_db_pool_metrics(super::DbPoolMetricsInput {
writer: buzz_db::DbPoolStats {
size: 12,
idle: 5,
max: 20,
},
reader: None,
audit: None,
search: buzz_db::DbPoolStats {
size: 4,
idle: 1,
max: 10,
},
});
});
let scrape = handle.render();
let connections = scrape
.lines()
.filter(|line| line.starts_with("buzz_db_pool_connections{"))
.collect::<Vec<_>>();
let max_connections = scrape
.lines()
.filter(|line| line.starts_with("buzz_db_pool_max_connections{"))
.collect::<Vec<_>>();
let configured = scrape
.lines()
.filter(|line| line.starts_with("buzz_db_pool_configured{"))
.collect::<Vec<_>>();
assert_eq!(connections.len(), 8, "unexpected scrape:\n{scrape}");
assert_eq!(max_connections.len(), 4, "unexpected scrape:\n{scrape}");
assert_eq!(configured.len(), 4, "unexpected scrape:\n{scrape}");
assert_eq!(
connections.len() + max_connections.len() + configured.len(),
16,
"physical pool contract must stay fixed at 16 raw series:\n{scrape}"
);
let roles = connections
.iter()
.filter_map(|line| {
["writer", "reader", "audit", "search"]
.into_iter()
.find(|role| line.contains(&format!(r#"pool_role="{role}""#)))
})
.collect::<BTreeSet<_>>();
assert_eq!(
roles,
BTreeSet::from(["writer", "reader", "audit", "search"])
);
assert!(connections
.iter()
.all(|line| label_keys(line) == BTreeSet::from(["pool_role", "state"])));
assert!(max_connections
.iter()
.all(|line| label_keys(line) == BTreeSet::from(["pool_role"])));
assert!(connections.iter().all(|line| {
line.contains(r#"state="idle""#) || line.contains(r#"state="active""#)
}));
assert!(!connections
.iter()
.any(|line| line.contains(r#"state="size""#) || line.contains(r#"state="max""#)));
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_connections{",
&[("pool_role", "writer"), ("state", "active")]
),
"7"
);
for state in ["idle", "active"] {
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_connections{",
&[("pool_role", "reader"), ("state", state)]
),
"0"
);
}
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_max_connections{",
&[("pool_role", "writer")]
),
"20"
);
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_max_connections{",
&[("pool_role", "reader")]
),
"0"
);
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_configured{",
&[("pool_role", "reader")]
),
"0"
);
assert_eq!(sample_value(&scrape, "buzz_db_pool_size ", &[]), "12");
assert_eq!(sample_value(&scrape, "buzz_db_pool_idle ", &[]), "5");
assert_eq!(sample_value(&scrape, "buzz_db_pool_active ", &[]), "7");
assert_eq!(sample_value(&scrape, "buzz_db_pool_max ", &[]), "20");
assert!(!scrape.contains("buzz_db_read_pool_size"));
}
#[test]
fn production_pool_sampler_reports_configured_reader_and_legacy_reader_family() {
let (recorder, handle) = super::readiness_test_recorder();
metrics::with_local_recorder(&recorder, || {
super::record_db_pool_metrics(super::DbPoolMetricsInput {
writer: buzz_db::DbPoolStats {
size: 1,
idle: 1,
max: 50,
},
reader: Some(buzz_db::DbPoolStats {
size: 9,
idle: 2,
max: 15,
}),
audit: None,
search: buzz_db::DbPoolStats {
size: 1,
idle: 1,
max: 10,
},
});
});
let scrape = handle.render();
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_configured{",
&[("pool_role", "reader")]
),
"1"
);
assert_eq!(sample_value(&scrape, "buzz_db_read_pool_size ", &[]), "9");
assert_eq!(sample_value(&scrape, "buzz_db_read_pool_idle ", &[]), "2");
assert_eq!(sample_value(&scrape, "buzz_db_read_pool_active ", &[]), "7");
assert_eq!(sample_value(&scrape, "buzz_db_read_pool_max ", &[]), "15");
}
#[tokio::test]
async fn production_pool_sampler_preserves_audit_and_search_pool_provenance() {
let audit_pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(5)
.connect_lazy("postgres://localhost/buzz")
.expect("construct lazy audit pool");
let search_pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(10)
.connect_lazy("postgres://localhost/buzz")
.expect("construct lazy search pool");
let (recorder, handle) = super::readiness_test_recorder();
metrics::with_local_recorder(&recorder, || {
super::record_db_pool_metrics(super::DbPoolMetricsInput {
writer: buzz_db::DbPoolStats {
size: 1,
idle: 1,
max: 50,
},
reader: None,
audit: Some(buzz_db::DbPoolStats::from_pool(&audit_pool)),
search: buzz_db::DbPoolStats::from_pool(&search_pool),
});
});
let scrape = handle.render();
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_max_connections{",
&[("pool_role", "audit")]
),
"5"
);
assert_eq!(
sample_value(
&scrape,
"buzz_db_pool_max_connections{",
&[("pool_role", "search")]
),
"10"
);
}
#[test]
fn production_builder_exports_frozen_db_pool_contract_and_187_series_budget() {
let (recorder, handle) = super::readiness_test_recorder();