diff --git a/.schema/pgdog.schema.json b/.schema/pgdog.schema.json index a334988f3..2dfcfdcc0 100644 --- a/.schema/pgdog.schema.json +++ b/.schema/pgdog.schema.json @@ -92,6 +92,20 @@ "query_parser_engine": "pg_query_protobuf", "query_size_limit": null, "query_size_limit_action": "warn", + "query_time_buckets": [ + 0.1, + 0.3, + 1.0, + 3.0, + 10.0, + 30.0, + 100.0, + 300.0, + 1000.0, + 3000.0, + 10000.0, + 30000.0 + ], "query_timeout": 9223372036854775807, "read_write_split": "include_primary", "read_write_strategy": "conservative", @@ -175,7 +189,8 @@ "endpoint": null, "headers": {}, "namespace": null, - "push_interval": 0 + "push_interval": 10000, + "temporality_preference": "Cumulative" } }, "plugins": { @@ -1085,6 +1100,28 @@ "$ref": "#/$defs/QuerySizeLimitAction", "default": "warn" }, + "query_time_buckets": { + "description": "Upper bounds, in milliseconds, of the `query_time_seconds` histogram buckets.\n\nEach bound emits one time series per pool, so prefer a short ladder that\nbrackets the latencies worth alerting on. Values are sorted and\ndeduplicated, and an implicit `+Inf` bucket is always appended. At most\n20 bounds are accepted and every one must be finite and greater than\nzero; a ladder PgDog cannot use is refused at startup rather than\nrepaired, so the exported buckets always match what was configured.\n\n**Note:** This setting cannot be changed at runtime. Restart PgDog after changing it.\n\n**Note:** A malformed `PGDOG_QUERY_TIME_BUCKETS` falls back to the default\nladder instead of being refused, matching every other environment variable.\n\n_Default:_ `[0.1, 0.3, 1, 3, 10, 30, 100, 300, 1000, 3000, 10000, 30000]`\n\nEnv: `PGDOG_QUERY_TIME_BUCKETS` (comma-separated milliseconds)\n\n", + "type": "array", + "default": [ + 0.1, + 0.3, + 1.0, + 3.0, + 10.0, + 30.0, + 100.0, + 300.0, + 1000.0, + 3000.0, + 10000.0, + 30000.0 + ], + "items": { + "type": "number", + "format": "double" + } + }, "query_timeout": { "description": "Maximum amount of time to wait for a Postgres query to finish executing.\n\n", "type": "integer", @@ -1598,10 +1635,42 @@ "format": "uint64", "default": 10000, "minimum": 0 + }, + "temporality_preference": { + "description": "Describes how the exported metric points should be described.\n\nSee https://opentelemetry.io/docs/specs/otel/metrics/data-model/#metric-points\n\n_Default:_ `Cumulative`, or `Delta` when `datadog_api_key` is set.\n\nEnv: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE`", + "anyOf": [ + { + "$ref": "#/$defs/OtelTemporalityPreference" + }, + { + "type": "null" + } + ], + "default": "Cumulative" } }, "additionalProperties": false }, + "OtelTemporalityPreference": { + "description": "Aggregation temporality used when exporting OTLP metric points.", + "oneOf": [ + { + "description": "Points report the value accumulated since the exporter started.", + "type": "string", + "const": "Cumulative" + }, + { + "description": "Points report the change since the last export.", + "type": "string", + "const": "Delta" + }, + { + "description": "Delta for sums, cumulative for histograms; minimizes exporter memory.", + "type": "string", + "const": "LowMemory" + } + ] + }, "PassthroughAuth": { "description": "toggle automatic creation of connection pools given the user name, database and password.\n\nSee [passthrough authentication](https://docs.pgdog.dev/features/authentication/#passthrough-authentication).\n\n", "oneOf": [ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 23f413c36..8c8b2bf30 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -24,3 +24,4 @@ Contributions are welcome. If you see a bug, feel free to submit a PR with a fix 1. Please format your code with `cargo fmt`. 2. If you're feeling generous, `cargo clippy` as well. 3. Please write and include tests. This is production software used in one of the most important areas of the stack. +4. If changes have been made to configuration schemas, run `cargo run -p pgdog-jsonschema` diff --git a/Cargo.lock b/Cargo.lock index 7717242b0..173e16ff6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3398,6 +3398,7 @@ dependencies = [ "semver", "serde", "serde_json", + "serde_repr", "sha1", "smallvec", "socket2 0.5.10", @@ -4607,6 +4608,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_repr" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d3b1629de253c70a0508c3899572da79ca359fdab27c7920ff00406df418906" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "serde_spanned" version = "0.6.9" @@ -5115,6 +5127,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" diff --git a/example.pgdog.toml b/example.pgdog.toml index d110dd659..41347ed71 100644 --- a/example.pgdog.toml +++ b/example.pgdog.toml @@ -152,6 +152,18 @@ openmetrics_port = 9090 # # Default: none openmetrics_namespace = "pgdog_" +# Upper bounds, in milliseconds, of the "query_time_seconds" histogram buckets. +# +# Each bound emits one time series per pool, so prefer a short ladder that +# brackets the latencies worth alerting on. Read once at startup: changing +# this requires a restart. +# +# At most 20 bounds, each finite and greater than zero. A ladder PgDog cannot +# use is refused at startup rather than repaired, so the buckets you scrape are +# always the ones you configured here. +# +# Default: [0.1, 0.3, 1, 3, 10, 30, 100, 300, 1000, 3000, 10000, 30000] +query_time_buckets = [0.1, 0.3, 1.0, 3.0, 10.0, 30.0, 100.0, 300.0, 1000.0, 3000.0, 10000.0, 30000.0] # Log output format. # # Default: text diff --git a/pgdog-config/src/core.rs b/pgdog-config/src/core.rs index 491cd9742..0c1fe7d77 100644 --- a/pgdog-config/src/core.rs +++ b/pgdog-config/src/core.rs @@ -6,6 +6,7 @@ use std::fs::read_to_string; use std::path::{Path, PathBuf}; use tracing::{error, info, warn}; +use crate::otel_temporality::OtelTemporalityPreference; use crate::sharding::ShardedSchema; use crate::util::random_string; use crate::{ @@ -120,6 +121,7 @@ impl ConfigAndUsers { self.config.check(); self.users.check(&self.config); self.validate_server_auth()?; + self.warn_if_data_dog_cumulative(); Ok(()) } @@ -149,6 +151,33 @@ impl ConfigAndUsers { Ok(()) } + fn warn_if_data_dog_cumulative(&self) { + match ( + &self.config.otel.datadog_api_key, + &self.config.otel.temporality_preference, + ) { + (Some(_datadog_present), Some(OtelTemporalityPreference::Cumulative)) + if std::env::var("IGNORE_DATADOG_CUMULATIVE_WARNING") + .ok() + .as_deref() + != Some("1") => + { + warn!( + "Sending Cumulative OTLP sums/histograms to Datadog is stateful and lossy: \ + all points on a timeseries must reach the same Agent/exporter (constraining \ + how you scale collectors), the first point of a new series may be dropped \ + (causing gaps on restart), and histogram min/max may be missing or \ + approximated. See \ + https://docs.datadoghq.com/opentelemetry/guide/otlp_delta_temporality/?tab=python#implications-of-using-cumulative-aggregation-temporality. \ + Set IGNORE_DATADOG_CUMULATIVE_WARNING=1 to silence." + ); + } + _ => { + // valid + } + } + } + /// Prepared statements are enabled. pub fn prepared_statements(&self) -> PreparedStatements { // Disable prepared statements automatically in session mode @@ -284,6 +313,7 @@ pub struct Config { /// /// #[serde(default)] + #[schemars(default = "Otel::schema_default")] pub otel: Otel, /// HashiCorp Vault settings, required for users configured with `server_auth = "vault"`. diff --git a/pgdog-config/src/general.rs b/pgdog-config/src/general.rs index 5630ef054..9db06496c 100644 --- a/pgdog-config/src/general.rs +++ b/pgdog-config/src/general.rs @@ -344,6 +344,29 @@ pub struct General { /// pub openmetrics_namespace: Option, + /// Upper bounds, in milliseconds, of the `query_time_seconds` histogram buckets. + /// + /// Each bound emits one time series per pool, so prefer a short ladder that + /// brackets the latencies worth alerting on. Values are sorted and + /// deduplicated, and an implicit `+Inf` bucket is always appended. At most + /// 20 bounds are accepted and every one must be finite and greater than + /// zero; a ladder PgDog cannot use is refused at startup rather than + /// repaired, so the exported buckets always match what was configured. + /// + /// **Note:** This setting cannot be changed at runtime. Restart PgDog after changing it. + /// + /// **Note:** A malformed `PGDOG_QUERY_TIME_BUCKETS` falls back to the default + /// ladder instead of being refused, matching every other environment variable. + /// + /// _Default:_ `[0.1, 0.3, 1, 3, 10, 30, 100, 300, 1000, 3000, 10000, 30000]` + /// + /// Env: `PGDOG_QUERY_TIME_BUCKETS` (comma-separated milliseconds) + /// + /// + #[serde(default = "General::query_time_buckets")] + #[schemars(default = "General::schema_default_query_time_buckets")] + pub query_time_buckets: Vec, + /// Enables support for prepared statements. /// /// _Default:_ `extended` @@ -881,6 +904,7 @@ impl Default for General { query_size_limit_action: Self::query_size_limit_action(), openmetrics_port: Self::openmetrics_port(), openmetrics_namespace: Self::openmetrics_namespace(), + query_time_buckets: Self::query_time_buckets(), prepared_statements: Self::prepared_statements(), query_parser_enabled: Self::query_parser_enabled(), query_parser: QueryParserLevel::default(), @@ -954,6 +978,11 @@ impl Default for General { } impl General { + /// Default upper bounds of the `query_time_seconds` histogram, in milliseconds. + pub const DEFAULT_QUERY_TIME_BUCKETS: [f64; 12] = [ + 0.1, 0.3, 1.0, 3.0, 10.0, 30.0, 100.0, 300.0, 1_000.0, 3_000.0, 10_000.0, 30_000.0, + ]; + fn env_or_default(env_var: &str, default: T) -> T { env::var(env_var) .ok() @@ -1312,6 +1341,31 @@ impl General { Self::env_option_string("PGDOG_OPENMETRICS_NAMESPACE") } + /// Default `query_time_seconds` bucket bounds, in milliseconds. + /// + /// Exponential from 100µs to 30s. A malformed env var degrades to the + /// defaults instead of failing startup; it degrades as a whole, so an + /// operator never gets a ladder made of the values that happened to parse. + pub fn query_time_buckets() -> Vec { + let Some(raw) = Self::env_option_string("PGDOG_QUERY_TIME_BUCKETS") else { + return Self::DEFAULT_QUERY_TIME_BUCKETS.to_vec(); + }; + + raw.split(',') + .map(|value| value.trim().parse::()) + .collect::, _>>() + .ok() + .filter(|buckets| !buckets.is_empty()) + .unwrap_or_else(|| Self::DEFAULT_QUERY_TIME_BUCKETS.to_vec()) + } + + /// Schema-only default, so the generated schema documents the shipped + /// bounds rather than whatever `PGDOG_QUERY_TIME_BUCKETS` happened to be + /// set to when the schema was generated. + fn schema_default_query_time_buckets() -> Vec { + Self::DEFAULT_QUERY_TIME_BUCKETS.to_vec() + } + fn default_dns_ttl() -> Option { Self::env_option("PGDOG_DNS_TTL") } @@ -1928,6 +1982,55 @@ mod tests { assert_eq!(General::openmetrics_namespace(), None); } + #[test] + fn test_query_time_buckets() { + let _guard = remove_env_var("PGDOG_QUERY_TIME_BUCKETS"); + + let general = General::default(); + assert_eq!( + general.query_time_buckets, + General::DEFAULT_QUERY_TIME_BUCKETS.to_vec() + ); + + let general: General = toml::from_str("query_time_buckets = [1.0, 10.0, 100.0]").unwrap(); + assert_eq!(general.query_time_buckets, vec![1.0, 10.0, 100.0]); + + // Omitting the setting keeps the defaults. + let general: General = toml::from_str("").unwrap(); + assert_eq!( + general.query_time_buckets, + General::DEFAULT_QUERY_TIME_BUCKETS.to_vec() + ); + } + + #[test] + fn test_env_query_time_buckets() { + let _guard = set_env_var("PGDOG_QUERY_TIME_BUCKETS", "1, 5,25"); + assert_eq!(General::query_time_buckets(), vec![1.0, 5.0, 25.0]); + + // Nothing parseable falls back to the defaults rather than + // disabling bucketing. + let _guard = set_env_var("PGDOG_QUERY_TIME_BUCKETS", "nonsense"); + assert_eq!( + General::query_time_buckets(), + General::DEFAULT_QUERY_TIME_BUCKETS.to_vec() + ); + + // One bad value discards the whole ladder: keeping the rest would build + // a histogram the operator never asked for. + let _guard = set_env_var("PGDOG_QUERY_TIME_BUCKETS", "1,abc,3"); + assert_eq!( + General::query_time_buckets(), + General::DEFAULT_QUERY_TIME_BUCKETS.to_vec() + ); + + let _guard = remove_env_var("PGDOG_QUERY_TIME_BUCKETS"); + assert_eq!( + General::query_time_buckets(), + General::DEFAULT_QUERY_TIME_BUCKETS.to_vec() + ); + } + #[test] fn test_env_invalid_enum_values() { let _guard = set_env_var("PGDOG_POOLER_MODE", "invalid_mode"); diff --git a/pgdog-config/src/lib.rs b/pgdog-config/src/lib.rs index b70ecb55d..d56925d92 100644 --- a/pgdog-config/src/lib.rs +++ b/pgdog-config/src/lib.rs @@ -8,6 +8,7 @@ pub mod general; pub mod memory; pub mod networking; pub mod otel; +pub mod otel_temporality; pub mod overrides; pub mod pooling; pub mod replication; diff --git a/pgdog-config/src/otel.rs b/pgdog-config/src/otel.rs index 6ba7ed51b..573e1f6d4 100644 --- a/pgdog-config/src/otel.rs +++ b/pgdog-config/src/otel.rs @@ -4,6 +4,10 @@ use std::env; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use crate::otel_temporality::OtelTemporalityPreference; + +const DEFAULT_PUSH_INTERVAL: u64 = 10_000; + /// OpenTelemetry push exporter settings. /// /// When `endpoint` is set, PgDog periodically POSTs OTLP JSON metrics @@ -60,10 +64,31 @@ pub struct Otel { /// /// Env: `OTEL_METRIC_EXPORT_INTERVAL` #[serde(default = "Otel::push_interval")] + #[schemars(default = "Otel::schema_default_push_interval")] pub push_interval: u64, + + /// Describes how the exported metric points should be described. + /// + /// See https://opentelemetry.io/docs/specs/otel/metrics/data-model/#metric-points + /// + /// _Default:_ `Cumulative`, or `Delta` when `datadog_api_key` is set. + /// + /// Env: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` + #[serde(default = "Otel::temporality_preference")] + #[schemars(default = "Otel::schema_default_temporality_preference")] + pub temporality_preference: Option, } impl Otel { + pub fn effective_temporality_preference(&self) -> OtelTemporalityPreference { + self.temporality_preference + .unwrap_or(if self.datadog_api_key.is_some() { + OtelTemporalityPreference::Delta + } else { + OtelTemporalityPreference::Cumulative + }) + } + fn env_option_string(env_var: &str) -> Option { env::var(env_var).ok().filter(|s| !s.is_empty()) } @@ -97,7 +122,32 @@ impl Otel { env::var("OTEL_METRIC_EXPORT_INTERVAL") .ok() .and_then(|v| v.parse().ok()) - .unwrap_or(10_000) + .unwrap_or(DEFAULT_PUSH_INTERVAL) + } + + fn temporality_preference() -> Option { + env::var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE") + .ok() + .and_then(|v| v.parse().ok()) + } + + fn schema_default_push_interval() -> u64 { + DEFAULT_PUSH_INTERVAL + } + + fn schema_default_temporality_preference() -> Option { + Some(OtelTemporalityPreference::Cumulative) + } + + /// Schema-only default for the whole `Otel` object, used so the top-level + /// `default` block in the generated JSON schema matches the per-field + /// documented defaults instead of the raw derived `Default` (0 / null). + pub fn schema_default() -> Self { + Self { + push_interval: Self::schema_default_push_interval(), + temporality_preference: Self::schema_default_temporality_preference(), + ..Self::default() + } } } @@ -132,7 +182,33 @@ mod test { assert!(otel.headers.is_empty()); assert!(otel.endpoint.is_none()); assert!(otel.datadog_api_key.is_none()); - assert_eq!(otel.push_interval, 10_000); + assert_eq!(otel.push_interval, DEFAULT_PUSH_INTERVAL); + assert!(otel.temporality_preference.is_none()); + } + + #[test] + fn endpoint_toml_wins_over_env() { + let _guard = set_env_var("OTEL_EXPORTER_OTLP_ENDPOINT", "https://env.example/v1"); + let toml = r#"endpoint = "https://toml.example/v1""#; + let otel: Otel = toml::from_str(toml).expect("parse"); + assert_eq!(otel.endpoint.as_deref(), Some("https://toml.example/v1")); + } + + #[test] + fn push_interval_env_used_when_toml_absent() { + let _guard = set_env_var("OTEL_METRIC_EXPORT_INTERVAL", "7500"); + let otel: Otel = toml::from_str("").expect("parse"); + assert_eq!(otel.push_interval, 7500); + } + + #[test] + fn temporality_preference_env_parsed() { + let _guard = set_env_var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE", "Delta"); + let otel: Otel = toml::from_str("").expect("parse"); + assert_eq!( + otel.temporality_preference, + Some(OtelTemporalityPreference::Delta) + ); } #[test] @@ -143,6 +219,7 @@ mod test { namespace = "pgdog_" datadog_api_key = "my-key" push_interval = 5000 + temporality_preference = "Delta" [otel.headers] Authorization = "Bearer token" @@ -156,12 +233,37 @@ mod test { assert_eq!(config.otel.namespace.as_deref(), Some("pgdog_")); assert_eq!(config.otel.datadog_api_key.as_deref(), Some("my-key")); assert_eq!(config.otel.push_interval, 5000); + assert_eq!( + config.otel.temporality_preference, + Some(OtelTemporalityPreference::Delta) + ); assert_eq!( config.otel.headers.get("Authorization").unwrap(), "Bearer token" ); } + #[test] + fn effective_temporality_defaults_to_delta_with_datadog_key() { + let mut otel = Otel::default(); + assert_eq!( + otel.effective_temporality_preference(), + OtelTemporalityPreference::Cumulative + ); + + otel.datadog_api_key = Some("abc".into()); + assert_eq!( + otel.effective_temporality_preference(), + OtelTemporalityPreference::Delta + ); + + otel.temporality_preference = Some(OtelTemporalityPreference::Cumulative); + assert_eq!( + otel.effective_temporality_preference(), + OtelTemporalityPreference::Cumulative + ); + } + #[test] fn namespace_from_env() { let _guard = set_env_var("PGDOG_OTEL_NAMESPACE", "pgdog_"); diff --git a/pgdog-config/src/otel_temporality.rs b/pgdog-config/src/otel_temporality.rs new file mode 100644 index 000000000..c2533e5dd --- /dev/null +++ b/pgdog-config/src/otel_temporality.rs @@ -0,0 +1,107 @@ +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Aggregation temporality used when exporting OTLP metric points. +// Note: Derive FromStr is case insensitive, matching OTEL behavior, though serde Deserialize +// see https://docs.rs/derive_more/latest/derive_more/derive.FromStr.html#empty-enums +#[derive( + derive_more::FromStr, Debug, Clone, Copy, PartialEq, Eq, Default, JsonSchema, Serialize, +)] +pub enum OtelTemporalityPreference { + /// Points report the value accumulated since the exporter started. + #[default] + Cumulative, + + /// Points report the change since the last export. + Delta, + /// Delta for sums, cumulative for histograms; minimizes exporter memory. + LowMemory, +} + +// Use case insensitive deserialization to match env var behavior +impl<'de> Deserialize<'de> for OtelTemporalityPreference { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + use std::str::FromStr; + + let s = String::deserialize(deserializer)?; + + // this from_str is case insensitive + Self::from_str(&s.to_ascii_lowercase()).map_err(|_| { + serde::de::Error::unknown_variant(&s, &["Cumulative", "Delta", "LowMemory"]) + }) + } +} + +#[cfg(test)] +mod test { + use super::*; + use std::str::FromStr; + + #[test] + fn default_is_cumulative() { + assert_eq!( + OtelTemporalityPreference::default(), + OtelTemporalityPreference::Cumulative, + ); + } + + #[test] + fn from_str_is_case_insensitive() { + let cases = [ + ("cumulative", OtelTemporalityPreference::Cumulative), + ("CUMULATIVE", OtelTemporalityPreference::Cumulative), + ("Cumulative", OtelTemporalityPreference::Cumulative), + ("delta", OtelTemporalityPreference::Delta), + ("DELTA", OtelTemporalityPreference::Delta), + ("Delta", OtelTemporalityPreference::Delta), + ("lowmemory", OtelTemporalityPreference::LowMemory), + ("LOWMEMORY", OtelTemporalityPreference::LowMemory), + ("LowMemory", OtelTemporalityPreference::LowMemory), + ]; + + for (input, expected) in cases { + assert_eq!( + OtelTemporalityPreference::from_str(input).unwrap(), + expected, + "input {input:?}", + ); + } + } + + #[test] + fn from_str_rejects_unknown_variant() { + assert!(OtelTemporalityPreference::from_str("nope").is_err()); + assert!(OtelTemporalityPreference::from_str("").is_err()); + } + + #[derive(Debug, Deserialize)] + struct Wrap { + t: OtelTemporalityPreference, + } + + #[test] + fn deserialize_is_case_insensitive() { + for (raw, expected) in [ + ("delta", OtelTemporalityPreference::Delta), + ("DELTA", OtelTemporalityPreference::Delta), + ("LowMemory", OtelTemporalityPreference::LowMemory), + ("lowmemory", OtelTemporalityPreference::LowMemory), + ("Cumulative", OtelTemporalityPreference::Cumulative), + ] { + let toml = format!("t = \"{raw}\""); + let w: Wrap = toml::from_str(&toml).expect("deserialize"); + assert_eq!(w.t, expected, "input {raw:?}"); + } + } + + #[test] + fn deserialize_rejects_unknown_variant() { + let err = toml::from_str::("t = \"histogram\"").unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains("histogram"), "message was: {msg}"); + assert!(msg.contains("Cumulative"), "message was: {msg}"); + } +} diff --git a/pgdog-stats/src/histogram.rs b/pgdog-stats/src/histogram.rs new file mode 100644 index 000000000..7f068f608 --- /dev/null +++ b/pgdog-stats/src/histogram.rs @@ -0,0 +1,515 @@ +//! Fixed-bucket histogram for latency distributions. +//! +//! Bucket bounds are process-wide: they are latched once at startup from the +//! configuration and never change while PgDog runs. That keeps [`Histogram`] +//! `Copy` and free of any per-instance bound storage, so histograms can be +//! summed element-wise without checking that their bounds agree. + +use std::{ops::AddAssign, sync::OnceLock, time::Duration}; + +use pgdog_config::General; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Maximum number of explicit bucket bounds. +/// +/// Each bound costs one series per pool on the OpenMetrics endpoint, so the +/// limit keeps cardinality bounded no matter what the configuration asks for. +pub const MAX_BUCKETS: usize = 20; + +/// Default bucket bounds, in milliseconds. +/// +/// Exponential from 100µs to 30s, which covers everything from an index lookup +/// on a warm buffer cache to a query about to hit `statement_timeout`. +/// +/// This belongs to the histogram rather than to the configuration, but owning +/// the default here would force `pgdog-config` to depend back on `pgdog-stats`, +/// which already depends on `pgdog-config` — so the default lives in +/// `pgdog-config` and this crate aliases it. +pub const DEFAULT_BOUNDS_MS: [f64; 12] = General::DEFAULT_QUERY_TIME_BUCKETS; + +static BOUNDS: OnceLock = OnceLock::new(); + +/// The process-wide bounds, plus how they got there. +struct LatchedBounds { + bounds: Bounds, + /// Set when [`set_bounds`] filled the latch, clear when a [`bounds`] read + /// fell back to the defaults. The two need different remedies. + configured: bool, +} + +impl LatchedBounds { + /// What a conflicting [`set_bounds`] call reports, given that these bounds + /// are already latched. A configured latch is replaced by a restart; a + /// read-defaults latch is not, because restarting just repeats the same + /// read-before-config ordering. + fn conflict(&self) -> Latch { + if self.configured { + Latch::AlreadySet(self.bounds) + } else { + Latch::DefaultedByRead(self.bounds) + } + } +} + +/// What a [`set_bounds`] call did. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Latch { + /// The bounds are now in force for the life of the process. + Set, + /// An earlier [`set_bounds`] already latched the enclosed bounds, which + /// stay in force. Applying different ones requires a restart. + AlreadySet(Bounds), + /// A [`bounds`] read latched the defaults before any [`set_bounds`] ran, so + /// the enclosed defaults stay in force. Unlike [`Latch::AlreadySet`] a + /// restart doesn't help: the read has to move after configuration load. + DefaultedByRead(Bounds), +} + +/// Latch the process-wide bucket bounds. +/// +/// Already-recorded histograms are indexed by position, so re-bucketing at +/// runtime would reinterpret every existing sample. The first value latched +/// wins for the life of the process. +pub fn set_bounds(bounds: Bounds) -> Latch { + match BOUNDS.set(LatchedBounds { + bounds, + configured: true, + }) { + Ok(()) => Latch::Set, + Err(_) => BOUNDS + .get() + .expect("a failed set means the latch is full") + .conflict(), + } +} + +/// Process-wide bucket bounds, defaulting to [`DEFAULT_BOUNDS_MS`]. +pub fn bounds() -> &'static Bounds { + &BOUNDS + .get_or_init(|| LatchedBounds { + bounds: Bounds::default(), + configured: false, + }) + .bounds +} + +/// Ascending upper bounds of histogram buckets. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Bounds { + bounds: [Duration; MAX_BUCKETS], + len: usize, +} + +impl Default for Bounds { + fn default() -> Self { + Self::defaults() + } +} + +/// Why a configured ladder can't be used. +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum BoundsError { + /// A value isn't a positive, finite number of milliseconds, or is too + /// large to be a [`Duration`]. + Invalid(f64), + /// More bounds than [`MAX_BUCKETS`] allows. + TooMany(usize), + /// No bounds at all. A histogram with no explicit bounds files every + /// sample under `+Inf`, which is worse than having no histogram. + Empty, +} + +impl std::fmt::Display for BoundsError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Invalid(value) => write!( + f, + "bound {value} is not a positive, finite number of milliseconds" + ), + Self::TooMany(got) => write!(f, "has {got} bounds, at most {MAX_BUCKETS} are allowed"), + Self::Empty => write!(f, "is empty; remove the setting to use the defaults"), + } + } +} + +impl std::error::Error for BoundsError {} + +impl Bounds { + /// Build bounds from millisecond values, or say why they can't be used. + /// + /// A ladder is taken whole or not at all: every value must be finite, + /// greater than zero, and small enough for a [`Duration`], and there must + /// be at least one and no more than [`MAX_BUCKETS`] of them. Dropping the + /// values that don't qualify would leave an operator running a histogram + /// whose buckets aren't the ones they wrote, with nothing in the exported + /// metrics to say which went missing. + /// + /// Sorting and deduplication are applied silently, since neither changes + /// which bounds the operator asked for. + pub fn try_from_millis(millis: &[f64]) -> Result { + if millis.is_empty() { + return Err(BoundsError::Empty); + } + + if millis.len() > MAX_BUCKETS { + return Err(BoundsError::TooMany(millis.len())); + } + + let mut values = millis + .iter() + .copied() + .map(|ms| { + if !ms.is_finite() || ms <= 0.0 { + return Err(BoundsError::Invalid(ms)); + } + // Anything past Duration::MAX is unusable for the same reason + // a negative bound is: it can't name a latency. + Duration::try_from_secs_f64(ms / 1_000.0).map_err(|_| BoundsError::Invalid(ms)) + }) + .collect::, _>>()?; + + values.sort_unstable(); + // Deduplicate on the exported value, not on the Duration: past roughly + // 10^7 seconds an f64 can't resolve nanoseconds, and two bounds that + // render to the same `le` label fail the whole Prometheus scrape. + values.dedup_by_key(|value| value.as_secs_f64()); + + let mut bounds = [Duration::ZERO; MAX_BUCKETS]; + bounds[..values.len()].copy_from_slice(&values); + + Ok(Self { + bounds, + len: values.len(), + }) + } + + /// The built-in bounds, which are always usable. + fn defaults() -> Self { + Self::try_from_millis(&DEFAULT_BOUNDS_MS).expect("DEFAULT_BOUNDS_MS is a valid ladder") + } + + /// Upper bounds, ascending. + pub fn as_slice(&self) -> &[Duration] { + &self.bounds[..self.len] + } + + /// Number of explicit bounds, excluding the implicit `+Inf` bucket. + pub fn len(&self) -> usize { + self.len + } + + /// No explicit bounds are configured. + pub fn is_empty(&self) -> bool { + self.len == 0 + } + + /// Upper bounds in seconds, the base unit used when exporting metrics. + pub fn seconds(&self) -> Vec { + self.as_slice().iter().map(Duration::as_secs_f64).collect() + } + + /// Bucket a sample belongs to. `len()` is the implicit `+Inf` bucket. + /// + /// Buckets are inclusive of their upper bound, matching the `le` semantics + /// of OpenMetrics. + fn index_of(&self, sample: Duration) -> usize { + self.as_slice().partition_point(|bound| *bound < sample) + } +} + +/// Cumulative distribution of duration samples. +/// +/// Counts are per-bucket rather than cumulative, so merging two histograms is +/// an element-wise add. Samples above the last bound land in the trailing +/// `+Inf` bucket. +/// +/// [`Histogram::observe_with`] is what keeps the buckets, the sum and the count +/// agreeing with each other; nothing enforces that across `Deserialize`, which +/// exists only so the stats structs embedding this one can derive it. A decoded +/// value can hold a count that disagrees with its buckets. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema)] +pub struct Histogram { + /// Observations per bucket. Index [`MAX_BUCKETS`] is the `+Inf` bucket, so + /// the slot a sample lands in doesn't depend on how many bounds are + /// configured. + buckets: [u64; MAX_BUCKETS + 1], + /// Sum of all observed samples. + sum: Duration, + /// Number of observed samples. + count: u64, +} + +impl Histogram { + /// Record a sample using the process-wide [`bounds`]. + #[inline] + pub fn observe(&mut self, sample: Duration) { + self.observe_with(sample, bounds()); + } + + /// Record a sample against explicit bounds. + #[inline] + pub fn observe_with(&mut self, sample: Duration, bounds: &Bounds) { + let index = bounds.index_of(sample); + // Samples past the last bound go to the dedicated overflow slot. + let index = if index < bounds.len() { + index + } else { + MAX_BUCKETS + }; + + self.buckets[index] = self.buckets[index].saturating_add(1); + self.sum = self.sum.saturating_add(sample); + self.count = self.count.saturating_add(1); + } + + /// Number of observed samples. + pub fn count(&self) -> u64 { + self.count + } + + /// Sum of all observed samples. + pub fn sum(&self) -> Duration { + self.sum + } + + /// No samples have been observed. + pub fn is_empty(&self) -> bool { + self.count == 0 + } + + /// Per-bucket counts for `bounds`, with the `+Inf` bucket last. + /// + /// The returned length is always `bounds.len() + 1`. + pub fn buckets(&self, bounds: &Bounds) -> Vec { + let mut counts = Vec::with_capacity(bounds.len() + 1); + counts.extend_from_slice(&self.buckets[..bounds.len()]); + counts.push(self.buckets[MAX_BUCKETS]); + // `_bucket` and `_count` are exported from different fields; a sample + // filed under a bound this call doesn't cover would only show up as a + // scrape that doesn't add up. + debug_assert_eq!(counts.iter().sum::(), self.count); + counts + } +} + +impl AddAssign for Histogram { + fn add_assign(&mut self, rhs: Self) { + for (bucket, rhs) in self.buckets.iter_mut().zip(rhs.buckets.iter()) { + *bucket = bucket.saturating_add(*rhs); + } + + self.sum = self.sum.saturating_add(rhs.sum); + self.count = self.count.saturating_add(rhs.count); + } +} + +#[cfg(test)] +mod test { + use super::*; + + fn test_bounds() -> Bounds { + Bounds::try_from_millis(&[1.0, 10.0, 100.0]).expect("valid ladder") + } + + #[test] + fn bounds_are_sorted_and_deduplicated() { + // Neither changes which bounds were asked for, so both are silent. + let bounds = Bounds::try_from_millis(&[10.0, 1.0, 10.0, 100.0]).expect("valid ladder"); + + assert_eq!(bounds.len(), 3); + assert_eq!( + bounds.as_slice(), + [ + Duration::from_millis(1), + Duration::from_millis(10), + Duration::from_millis(100), + ] + ); + } + + #[test] + fn the_default_ladder_is_usable() { + // `Bounds::defaults` panics if this ever stops holding, so state the + // invariant here rather than degrading at runtime to a bound-less + // histogram that files every sample under +Inf. + assert!(Bounds::try_from_millis(&DEFAULT_BOUNDS_MS).is_ok()); + assert_eq!(Bounds::default().len(), DEFAULT_BOUNDS_MS.len()); + } + + #[test] + fn a_conflict_reports_how_the_latch_was_filled() { + let bounds = test_bounds(); + + // Latched by an earlier `set_bounds`: a restart applies the new ladder. + assert_eq!( + LatchedBounds { + bounds, + configured: true + } + .conflict(), + Latch::AlreadySet(bounds), + ); + + // Latched by a `bounds()` read: a restart repeats the same ordering, so + // the warning has to point somewhere other than "restart". + assert_eq!( + LatchedBounds { + bounds, + configured: false + } + .conflict(), + Latch::DefaultedByRead(bounds), + ); + } + + #[test] + fn a_single_bad_bound_rejects_the_whole_ladder() { + // The point of the error: an operator who typo'd one value gets told + // so, rather than silently running a ladder missing that bucket. + for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY, -1.0, 0.0] { + match Bounds::try_from_millis(&[1.0, bad, 100.0]) { + // Compared by shape rather than with assert_eq!: NaN is not + // equal to itself, so a derived PartialEq can never match it. + Err(BoundsError::Invalid(value)) => assert!( + value == bad || (value.is_nan() && bad.is_nan()), + "rejected {bad} but reported {value}" + ), + other => panic!("{bad} should have been rejected, got {other:?}"), + } + } + } + + #[test] + fn a_bound_too_large_for_a_duration_is_rejected() { + // 1e30 ms exceeds Duration::MAX: an error, not a panic and not a drop. + assert_eq!( + Bounds::try_from_millis(&[1e30, 5.0]), + Err(BoundsError::Invalid(1e30)) + ); + } + + #[test] + fn more_bounds_than_the_maximum_are_rejected() { + let millis = (1..=(MAX_BUCKETS as u64 + 10)) + .map(|ms| ms as f64) + .collect::>(); + + assert_eq!( + Bounds::try_from_millis(&millis), + Err(BoundsError::TooMany(MAX_BUCKETS + 10)) + ); + + // Exactly at the cap is fine. + let millis = (1..=MAX_BUCKETS as u64) + .map(|ms| ms as f64) + .collect::>(); + assert_eq!( + Bounds::try_from_millis(&millis).map(|b| b.len()), + Ok(MAX_BUCKETS) + ); + } + + #[test] + fn an_empty_ladder_is_rejected() { + // Silently substituting the defaults would leave the operator running + // buckets they explicitly asked not to have. + assert_eq!(Bounds::try_from_millis(&[]), Err(BoundsError::Empty)); + } + + #[test] + fn bounds_errors_name_the_offending_value() { + // These render into the startup error an operator has to act on. + assert_eq!( + BoundsError::Invalid(-1.0).to_string(), + "bound -1 is not a positive, finite number of milliseconds" + ); + assert_eq!( + BoundsError::TooMany(25).to_string(), + format!("has 25 bounds, at most {MAX_BUCKETS} are allowed") + ); + assert_eq!( + BoundsError::Empty.to_string(), + "is empty; remove the setting to use the defaults" + ); + } + + #[test] + fn bounds_are_inclusive_of_upper_bound() { + let bounds = test_bounds(); + + // Exactly on a bound belongs to that bound's bucket, not the next. + assert_eq!(bounds.index_of(Duration::from_millis(1)), 0); + assert_eq!(bounds.index_of(Duration::from_micros(999)), 0); + assert_eq!(bounds.index_of(Duration::from_micros(1001)), 1); + assert_eq!(bounds.index_of(Duration::from_millis(100)), 2); + // Past the last bound: the +Inf bucket. + assert_eq!(bounds.index_of(Duration::from_millis(101)), 3); + } + + #[test] + fn observe_counts_sum_and_buckets() { + let bounds = test_bounds(); + let mut histogram = Histogram::default(); + + histogram.observe_with(Duration::from_micros(500), &bounds); + histogram.observe_with(Duration::from_millis(5), &bounds); + histogram.observe_with(Duration::from_millis(50), &bounds); + histogram.observe_with(Duration::from_secs(1), &bounds); + + assert_eq!(histogram.count(), 4); + assert_eq!( + histogram.sum(), + Duration::from_micros(500) + Duration::from_millis(55) + Duration::from_secs(1) + ); + // One per bucket, including the +Inf overflow. + assert_eq!(histogram.buckets(&bounds), vec![1, 1, 1, 1]); + } + + #[test] + fn buckets_length_matches_bounds() { + let bounds = test_bounds(); + let histogram = Histogram::default(); + + assert_eq!(histogram.buckets(&bounds).len(), bounds.len() + 1); + assert!(histogram.is_empty()); + } + + #[test] + fn overflow_bucket_is_independent_of_bound_count() { + // The +Inf slot is fixed, so a histogram observed under one bound count + // still reports its overflow correctly. + let narrow = Bounds::try_from_millis(&[1.0]).expect("valid ladder"); + let mut histogram = Histogram::default(); + + histogram.observe_with(Duration::from_secs(10), &narrow); + + assert_eq!(histogram.buckets(&narrow), vec![0, 1]); + } + + #[test] + fn add_assign_merges_element_wise() { + let bounds = test_bounds(); + let mut a = Histogram::default(); + let mut b = Histogram::default(); + + a.observe_with(Duration::from_micros(500), &bounds); + a.observe_with(Duration::from_millis(5), &bounds); + b.observe_with(Duration::from_millis(5), &bounds); + b.observe_with(Duration::from_secs(1), &bounds); + + let mut merged = a; + merged += b; + + assert_eq!(merged.count(), 4); + assert_eq!(merged.buckets(&bounds), vec![1, 2, 0, 1]); + assert_eq!(merged.sum(), a.sum() + b.sum()); + } + + #[test] + fn seconds_converts_bounds() { + let bounds = Bounds::try_from_millis(&[0.1, 1.0, 1_000.0]).expect("valid ladder"); + + assert_eq!(bounds.seconds(), vec![0.0001, 0.001, 1.0]); + } +} diff --git a/pgdog-stats/src/lib.rs b/pgdog-stats/src/lib.rs index 6938332a5..5a5671c83 100644 --- a/pgdog-stats/src/lib.rs +++ b/pgdog-stats/src/lib.rs @@ -1,4 +1,5 @@ pub mod client; +pub mod histogram; pub mod memory; pub mod pool; pub mod replication; @@ -8,6 +9,7 @@ pub mod server; pub mod state; pub mod user; +pub use histogram::Histogram; pub use memory::*; pub use pool::*; pub use replication::*; diff --git a/pgdog-stats/src/pool.rs b/pgdog-stats/src/pool.rs index a993c8ec2..c934d7dec 100644 --- a/pgdog-stats/src/pool.rs +++ b/pgdog-stats/src/pool.rs @@ -7,7 +7,7 @@ use pgdog_config::{PoolerMode, PreparedStatements, Role, pooling::ConnectionReco use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use crate::{LsnStats, ReplicaLag}; +use crate::{Histogram, LsnStats, ReplicaLag, server::Counts as ServerCounts}; /// Pool statistics. /// @@ -195,9 +195,21 @@ pub struct Stats { last_counts: Counts, // Average counts. pub averages: Counts, + /// Distribution of individual query durations, cumulative for the lifetime + /// of the pool. Unlike [`Stats::averages`], this is never recalculated. + pub query_time_histogram: Histogram, } impl Stats { + /// Fold a checked-in server's counts and latency samples into the totals. + /// + /// Counts add; the histogram merges element-wise. Both are cumulative, so + /// the caller is responsible for handing over each sample exactly once. + pub fn check_in(&mut self, counts: ServerCounts, histogram: Histogram) { + self.counts = self.counts + counts; + self.query_time_histogram += histogram; + } + /// Calculate averages. pub fn calc_averages(&mut self, time: Duration) { let secs = time.as_secs() as usize; @@ -438,3 +450,103 @@ impl Default for Config { } } } + +#[cfg(test)] +mod test { + use super::*; + use crate::histogram::Bounds; + + /// Explicit bounds, so tests don't depend on the process-wide latch. + fn bounds() -> Bounds { + Bounds::try_from_millis(&[1.0, 10.0, 100.0]).expect("valid ladder") + } + + fn samples(millis: &[u64]) -> Histogram { + let bounds = bounds(); + let mut histogram = Histogram::default(); + for ms in millis { + histogram.observe_with(Duration::from_millis(*ms), &bounds); + } + histogram + } + + fn queries(count: usize, query_time: Duration) -> ServerCounts { + ServerCounts { + queries: count, + query_time, + ..Default::default() + } + } + + #[test] + fn check_in_adds_counts_and_merges_samples() { + let mut stats = Stats::default(); + + stats.check_in(queries(2, Duration::from_millis(7)), samples(&[5, 2])); + + assert_eq!(stats.counts.query_count, 2); + assert_eq!(stats.counts.query_time, Duration::from_millis(7)); + assert_eq!(stats.query_time_histogram.count(), 2); + assert_eq!(stats.query_time_histogram.sum(), Duration::from_millis(7)); + } + + #[test] + fn check_in_accumulates_across_servers() { + let mut stats = Stats::default(); + + stats.check_in(queries(1, Duration::from_millis(5)), samples(&[5])); + stats.check_in( + queries(3, Duration::from_millis(60)), + samples(&[20, 20, 20]), + ); + + assert_eq!(stats.counts.query_count, 4); + assert_eq!(stats.query_time_histogram.count(), 4); + assert_eq!(stats.query_time_histogram.sum(), Duration::from_millis(65)); + } + + #[test] + fn check_in_preserves_the_bucket_distribution() { + let mut stats = Stats::default(); + + // Distinct counts per bucket, so a merge that mixed up slots is caught. + stats.check_in(ServerCounts::default(), samples(&[1])); + stats.check_in(ServerCounts::default(), samples(&[5, 5])); + stats.check_in(ServerCounts::default(), samples(&[50, 50, 50])); + stats.check_in(ServerCounts::default(), samples(&[500, 500, 500, 500])); + + // Per-bucket, not cumulative: <=1ms, <=10ms, <=100ms, +Inf. The + // cumulative `le` form is derived at export time. + assert_eq!( + stats.query_time_histogram.buckets(&bounds()), + vec![1, 2, 3, 4] + ); + assert_eq!(stats.query_time_histogram.count(), 10); + } + + #[test] + fn check_in_without_samples_leaves_the_histogram_alone() { + let mut stats = Stats::default(); + stats.check_in(queries(1, Duration::from_millis(5)), samples(&[5])); + + // A server can be checked in without having run a query. + stats.check_in(ServerCounts::default(), Histogram::default()); + + assert_eq!(stats.query_time_histogram.count(), 1); + assert_eq!(stats.query_time_histogram.sum(), Duration::from_millis(5)); + } + + #[test] + fn calc_averages_does_not_disturb_the_histogram() { + let mut stats = Stats::default(); + stats.check_in(queries(2, Duration::from_millis(10)), samples(&[5, 5])); + + stats.calc_averages(Duration::from_secs(1)); + + // The histogram is exported as a cumulative counter, so unlike + // `averages` it is never rescaled or reset. + assert_eq!(stats.query_time_histogram.count(), 2); + assert_eq!(stats.query_time_histogram.sum(), Duration::from_millis(10)); + assert_eq!(stats.averages.query_count, 2); + } +} diff --git a/pgdog-stats/src/server.rs b/pgdog-stats/src/server.rs index 0e4a14f48..33319eaf9 100644 --- a/pgdog-stats/src/server.rs +++ b/pgdog-stats/src/server.rs @@ -1,3 +1,4 @@ +use crate::Histogram; use crate::memory::MemoryStats; use crate::pool::Counts as PoolCounts; use crate::state::State; @@ -106,6 +107,9 @@ pub struct Stats { pub created_at_time: SystemTime, pub total: Counts, pub last_checkout: Counts, + /// Distribution of query durations since the last check-in. Drained into + /// the pool's histogram when the server is checked back in. + pub query_time_histogram: Histogram, pub pool_id: u64, pub memory: MemoryStats, pub last_sent: u8, @@ -119,6 +123,7 @@ impl Default for Stats { created_at_time: SystemTime::now(), total: Counts::default(), last_checkout: Counts::default(), + query_time_histogram: Histogram::default(), pool_id: 0, memory: MemoryStats::default(), last_sent: 0, diff --git a/pgdog/Cargo.toml b/pgdog/Cargo.toml index 6987a3456..5426f0299 100644 --- a/pgdog/Cargo.toml +++ b/pgdog/Cargo.toml @@ -29,6 +29,7 @@ bytes.workspace = true clap = { version = "4", features = ["derive"] } serde = { version = "1", features = ["derive"] } serde_json.workspace = true +serde_repr = "0.1" async-trait = "0.1" rand = "0.9.2" once_cell = "1" diff --git a/pgdog/src/backend/pool/inner.rs b/pgdog/src/backend/pool/inner.rs index 11e35f64f..55fa2e6a6 100644 --- a/pgdog/src/backend/pool/inner.rs +++ b/pgdog/src/backend/pool/inner.rs @@ -10,7 +10,7 @@ use crate::backend::{Server, stats::Counts as BackendCounts}; use crate::net::messages::{BackendKeyData, BackendPid, FrontendPid}; use pgdog_config::Role; -use pgdog_stats::RoleSpecificConfig; +use pgdog_stats::{Histogram, RoleSpecificConfig}; use tokio::time::Instant; use super::{Config, Error, Pool, Request, Stats, Taken, Waiter, lsn_monitor::ReplicaLag}; @@ -367,6 +367,7 @@ impl Inner { mut server: Box, now: Instant, stats: BackendCounts, + histogram: Histogram, moving: bool, ) -> Result { let mut result = CheckInResult { @@ -381,15 +382,17 @@ impl Inner { server.stats_mut().set_pool_id(moved.id()); server.stats().update(); server.replace_oids(&moved.inner().oids); - moved.lock().maybe_check_in(server, now, stats, true)?; + moved + .lock() + .maybe_check_in(server, now, stats, histogram, true)?; return Ok(result); } } self.taken.check_in(server.id())?; - // Update stats - self.stats.counts = self.stats.counts + stats; + // Update stats. + self.stats.check_in(stats, histogram); // Ban the pool from serving more clients. if server.error() { @@ -559,6 +562,41 @@ mod test { assert!(!inner.paused); } + /// `maybe_check_in` is glue; the merge itself is covered exhaustively by + /// `pgdog_stats::pool::Stats::check_in`. This only proves the wire is + /// connected, so a drained server histogram can't silently go nowhere. + #[test] + fn check_in_hands_the_server_histogram_to_the_pool() { + let mut inner = Inner { + online: true, + ..Default::default() + }; + let server = Box::new(Server::default()); + + inner + .taken + .take(FrontendPid::new(), server.id(), server.key().clone()); + + let mut histogram = Histogram::default(); + histogram.observe(Duration::from_millis(5)); + + inner + .maybe_check_in( + server, + Instant::now(), + BackendCounts::default(), + histogram, + false, + ) + .unwrap(); + + assert_eq!(inner.stats.query_time_histogram.count(), 1); + assert_eq!( + inner.stats.query_time_histogram.sum(), + Duration::from_millis(5) + ); + } + #[test] fn test_offline_pool_behavior() { let mut inner = Inner::default(); @@ -570,7 +608,13 @@ mod test { .take(FrontendPid::new(), server.id(), server.key().clone()); let result = inner - .maybe_check_in(server, Instant::now(), BackendCounts::default(), false) + .maybe_check_in( + server, + Instant::now(), + BackendCounts::default(), + Histogram::default(), + false, + ) .unwrap(); assert!(!result.server_error); @@ -593,7 +637,13 @@ mod test { .take(FrontendPid::new(), server.id(), server.key().clone()); inner - .maybe_check_in(server, Instant::now(), BackendCounts::default(), false) + .maybe_check_in( + server, + Instant::now(), + BackendCounts::default(), + Histogram::default(), + false, + ) .unwrap(); assert_eq!(inner.total(), 0); // pool paused, connection not added @@ -614,7 +664,13 @@ mod test { .take(FrontendPid::new(), server.id(), server.key().clone()); let result = inner - .maybe_check_in(server, Instant::now(), BackendCounts::default(), false) + .maybe_check_in( + server, + Instant::now(), + BackendCounts::default(), + Histogram::default(), + false, + ) .unwrap(); assert!(!result.server_error); @@ -638,7 +694,13 @@ mod test { assert_eq!(inner.checked_out(), 1); let result = inner - .maybe_check_in(server, Instant::now(), BackendCounts::default(), false) + .maybe_check_in( + server, + Instant::now(), + BackendCounts::default(), + Histogram::default(), + false, + ) .unwrap(); assert!(result.server_error); @@ -795,7 +857,13 @@ mod test { .take(FrontendPid::new(), server.id(), server.key().clone()); inner - .maybe_check_in(server, Instant::now(), BackendCounts::default(), false) + .maybe_check_in( + server, + Instant::now(), + BackendCounts::default(), + Histogram::default(), + false, + ) .unwrap(); assert_eq!(inner.idle(), 1); @@ -848,6 +916,7 @@ mod test { server, Instant::now() + Duration::from_secs(61), // Exceeds max age BackendCounts::default(), + Histogram::default(), false, ) .unwrap(); @@ -1221,14 +1290,26 @@ mod test { // Check in both connections let now = Instant::now(); inner - .maybe_check_in(conn1, now, BackendCounts::default(), false) + .maybe_check_in( + conn1, + now, + BackendCounts::default(), + Histogram::default(), + false, + ) .unwrap(); assert_eq!(inner.idle(), 1); assert_eq!(inner.checked_out(), 1); assert_eq!(inner.total(), 2); inner - .maybe_check_in(conn2, now, BackendCounts::default(), false) + .maybe_check_in( + conn2, + now, + BackendCounts::default(), + Histogram::default(), + false, + ) .unwrap(); assert_eq!(inner.idle(), 2); assert_eq!(inner.checked_out(), 0); diff --git a/pgdog/src/backend/pool/pool_impl.rs b/pgdog/src/backend/pool/pool_impl.rs index 376b3cfd3..a74883a4e 100644 --- a/pgdog/src/backend/pool/pool_impl.rs +++ b/pgdog/src/backend/pool/pool_impl.rs @@ -254,12 +254,12 @@ impl Pool { server.stats().last_used() }; - let counts = { + let (counts, histogram) = { let stats = server.stats_mut(); stats.clear_client_id(); - let counts = stats.reset_last_checkout(); + let drained = stats.reset_last_checkout(); stats.update(); - counts + drained }; // Check everything and maybe check the connection @@ -267,7 +267,10 @@ impl Pool { let CheckInResult { server_error, replenish, - } = { self.lock().maybe_check_in(server, now, counts, false)? }; + } = { + self.lock() + .maybe_check_in(server, now, counts, histogram, false)? + }; if server_error { error!( diff --git a/pgdog/src/backend/server.rs b/pgdog/src/backend/server.rs index 96d9d0e7f..4155cbc6b 100644 --- a/pgdog/src/backend/server.rs +++ b/pgdog/src/backend/server.rs @@ -2667,7 +2667,7 @@ pub mod test { assert_eq!(server.stats().total().transactions, i); } - let counts = server.stats_mut().reset_last_checkout(); + let (counts, _) = server.stats_mut().reset_last_checkout(); assert_eq!(counts.queries, 25); assert_eq!(counts.transactions, 25); @@ -2688,7 +2688,7 @@ pub mod test { assert_eq!(server.stats().total().transactions, 25 + i); } - let counts = server.stats_mut().reset_last_checkout(); + let (counts, _) = server.stats_mut().reset_last_checkout(); assert_eq!(counts.queries, 25 * 4); assert_eq!(counts.transactions, 25); assert_eq!(server.stats().total().queries, 25 + (25 * 4)); diff --git a/pgdog/src/backend/stats.rs b/pgdog/src/backend/stats.rs index 905b61cdf..4a76bb248 100644 --- a/pgdog/src/backend/stats.rs +++ b/pgdog/src/backend/stats.rs @@ -1,5 +1,6 @@ //! Keep track of server stats. +use pgdog_stats::Histogram; use std::ops::{Deref, DerefMut}; use std::sync::Arc; @@ -251,6 +252,9 @@ impl Stats { let duration = now.duration_since(query_timer); self.local.total.query_time += duration; self.local.last_checkout.query_time += duration; + // The histogram is handed to the pool on check-in separately from + // the ordinary server counts, so it is not duplicated in `total`. + self.local.query_time_histogram.observe(duration); } } @@ -330,10 +334,12 @@ impl Stats { } /// Reset last_checkout counts. - pub fn reset_last_checkout(&mut self) -> Counts { + pub fn reset_last_checkout(&mut self) -> (Counts, Histogram) { let counts = self.local.last_checkout; + let histogram = self.local.query_time_histogram; self.local.last_checkout = Counts::default(); - counts + self.local.query_time_histogram = Histogram::default(); + (counts, histogram) } // Fast accessor methods - read from local, no locking. @@ -387,6 +393,12 @@ impl Stats { self.local.last_checkout } + /// Get query latency samples taken since the last check-in (local, no lock). + #[inline] + pub fn query_time_histogram(&self) -> Histogram { + self.local.query_time_histogram + } + /// Clear client_id. #[inline] pub fn clear_client_id(&mut self) { @@ -399,3 +411,102 @@ impl Stats { self.sync_to_shared(); } } + +#[cfg(test)] +mod test { + use std::time::Duration; + + use super::*; + + /// A `Stats` handle with no network connection behind it. + /// + /// `query()` takes `now` as an argument and only mutates local counters, + /// so the whole latency-recording path is testable without a server. + /// `pid` must be unique per test: `connect` registers in a global map. + fn stats(pid: i32) -> Stats { + Stats::connect( + BackendPid::for_test(pid), + &Address::default(), + &Parameters::default(), + &ServerOptions::default(), + &Memory::default(), + ) + } + + #[test] + fn query_records_latency_into_server_histogram() { + let mut stats = stats(1); + let start = Instant::now(); + + stats.set_timers(start); + stats.query(start + Duration::from_millis(5), false); + + let histogram = stats.query_time_histogram(); + assert_eq!(histogram.count(), 1); + assert_eq!(histogram.sum(), Duration::from_millis(5)); + } + + #[test] + fn query_time_is_recorded_as_both_a_total_and_a_sample() { + let mut stats = stats(2); + let start = Instant::now(); + + stats.set_timers(start); + stats.query(start + Duration::from_millis(5), false); + + // One observation feeds two exports: the scalar `query_time` total and + // the histogram. The histogram is held separately from the counts and + // handed to the pool once, at check-in. + assert_eq!(stats.total().query_time, Duration::from_millis(5)); + assert_eq!(stats.query_time_histogram().count(), 1); + } + + #[test] + fn every_query_is_observed() { + let mut stats = stats(3); + let start = Instant::now(); + + for i in 1..=25u64 { + stats.set_timers(start); + stats.query(start + Duration::from_millis(i), false); + + assert_eq!(stats.last_checkout().queries, i as usize); + assert_eq!(stats.query_time_histogram().count(), i); + } + + // 1 + 2 + ... + 25 milliseconds. + assert_eq!( + stats.query_time_histogram().sum(), + Duration::from_millis(25 * 26 / 2) + ); + } + + #[test] + fn reset_last_checkout_hands_off_the_histogram() { + let mut stats = stats(4); + let start = Instant::now(); + + stats.set_timers(start); + stats.query(start + Duration::from_millis(5), false); + + // Check-in drains the samples separately from the ordinary counts... + let (counts, histogram) = stats.reset_last_checkout(); + assert_eq!(counts.queries, 1); + assert_eq!(histogram.count(), 1); + + // ...and leaves nothing behind, so the next checkout can't re-report them. + assert_eq!(stats.query_time_histogram().count(), 0); + assert_eq!(stats.query_time_histogram().sum(), Duration::ZERO); + } + + #[test] + fn query_without_a_timer_is_not_observed() { + let mut stats = stats(5); + + // No `set_timers`: nothing to measure, so nothing is recorded. + stats.query(Instant::now(), false); + + assert_eq!(stats.last_checkout().queries, 1); + assert_eq!(stats.query_time_histogram().count(), 0); + } +} diff --git a/pgdog/src/config/mod.rs b/pgdog/src/config/mod.rs index a7a298694..f90b9c259 100644 --- a/pgdog/src/config/mod.rs +++ b/pgdog/src/config/mod.rs @@ -47,6 +47,8 @@ use std::sync::Arc; use arc_swap::ArcSwap; use once_cell::sync::Lazy; +use pgdog_stats::histogram::{self, Bounds, Latch}; +use tracing::warn; static CONFIG: Lazy> = Lazy::new(|| ArcSwap::from_pointee(ConfigAndUsers::default())); @@ -72,10 +74,51 @@ pub fn set(mut config: ConfigAndUsers) -> Result { // And also moved outside the configuration to the place of table.load_centroids()?; } + set_histogram_bounds(&config)?; CONFIG.store(Arc::new(config.clone())); Ok(config) } +/// Latch the process-wide query latency histogram buckets. +/// +/// Bucket bounds are fixed for the life of the process: already-recorded +/// histograms are indexed by position, so re-bucketing at runtime would +/// silently reinterpret every existing sample. A reload that changes them is +/// ignored with a warning. +/// +/// A ladder PgDog can't use is a hard error rather than something to repair. +/// Dropping a bad bound and carrying on leaves an operator with a histogram +/// whose buckets don't match what they wrote, and nothing in the exported +/// metrics says which one went missing. Refusing at startup — or refusing the +/// reload, leaving the running configuration untouched — is the only outcome +/// they can act on. +/// +/// This covers `pgdog.toml` only. A malformed `PGDOG_QUERY_TIME_BUCKETS` falls +/// back to the defaults without complaint, because every environment variable +/// in `General` does: they are read through serde defaults, which have no way +/// to report a failure. +fn set_histogram_bounds(config: &ConfigAndUsers) -> Result<(), Error> { + let configured = Bounds::try_from_millis(&config.config.general.query_time_buckets) + .map_err(|err| Error::ParseError(format!("\"query_time_buckets\" {err}")))?; + + match histogram::set_bounds(configured) { + Latch::AlreadySet(current) if current != configured => { + warn!("\"query_time_buckets\" cannot be changed at runtime, restart PgDog to apply") + } + // The bounds were read before this ran, so the read latched the + // defaults. Restarting repeats the same ordering, so telling the + // operator to restart would send them round a loop. + Latch::DefaultedByRead(current) if current != configured => warn!( + "\"query_time_buckets\" was read before the configuration was loaded, so the default buckets are in force for the life of this process" + ), + // The configuration is in force, either because this call latched it or + // because an earlier load latched the same bounds. + Latch::Set | Latch::AlreadySet(_) | Latch::DefaultedByRead(_) => (), + } + + Ok(()) +} + /// Validate sharding key lookup queries with the SQL parser: /// the query must be syntactically valid, a single statement, and /// reference exactly one parameter, `$1`. `lookup_result = "shard"` @@ -579,3 +622,60 @@ hasher = "sha1" assert!(err.to_string().contains("$1")); } } + +#[cfg(test)] +mod histogram_bounds_tests { + use super::*; + + fn config_from(source: &str) -> ConfigAndUsers { + ConfigAndUsers { + config: toml::from_str(source).unwrap(), + ..Default::default() + } + } + + /// Every case here is rejected before `histogram::set_bounds` is reached, + /// so these never touch the process-wide latch and stay order-independent. + fn rejection(buckets: &str) -> String { + let config = config_from(&format!("[general]\nquery_time_buckets = {buckets}\n")); + set_histogram_bounds(&config) + .expect_err("should have been rejected") + .to_string() + } + + #[test] + fn test_rejects_a_negative_bound() { + let err = rejection("[1.0, -3.0, 4.0]"); + assert!(err.contains("query_time_buckets"), "{err}"); + assert!(err.contains("-3"), "{err}"); + } + + #[test] + fn test_rejects_a_zero_bound() { + assert!(rejection("[0.0, 1.0]").contains("0 is not a positive")); + } + + #[test] + fn test_rejects_an_empty_ladder() { + assert!(rejection("[]").contains("is empty")); + } + + #[test] + fn test_rejects_more_bounds_than_the_maximum() { + let many = (1..=30) + .map(|n| n.to_string()) + .collect::>() + .join(", "); + assert!(rejection(&format!("[{many}]")).contains("at most")); + } + + #[test] + fn test_error_names_the_setting_so_an_operator_can_find_it() { + // The whole point of failing at startup is an actionable message. + let err = rejection("[1.0, -3.0]"); + assert!( + err.starts_with("parse error: \"query_time_buckets\""), + "{err}" + ); + } +} diff --git a/pgdog/src/stats/listeners.rs b/pgdog/src/stats/listeners.rs index 89ea5a1b0..3a63522dd 100644 --- a/pgdog/src/stats/listeners.rs +++ b/pgdog/src/stats/listeners.rs @@ -1,4 +1,4 @@ -use crate::backend::pub_sub::listener; +use crate::{backend::pub_sub::listener, stats::OpenMetricType}; use super::{Measurement, Metric, OpenMetric}; @@ -40,19 +40,19 @@ impl Listeners { name: "pub_sub_listeners".into(), measurements: listeners, help: "Current number of clients listening on a pub/sub channel.".into(), - metric_type: "gauge".into(), + metric_type: OpenMetricType::Gauge, }), Metric::new(ListenerMetric { name: "pub_sub_listener_received".into(), measurements: received, help: "Total number of notifications received by pub/sub listeners.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }), Metric::new(ListenerMetric { name: "pub_sub_listener_dropped".into(), measurements: dropped, help: "Total number of notifications dropped by lagging pub/sub listeners.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }), ] } @@ -62,7 +62,7 @@ struct ListenerMetric { name: String, measurements: Vec, help: String, - metric_type: String, + metric_type: OpenMetricType, } impl OpenMetric for ListenerMetric { @@ -78,8 +78,8 @@ impl OpenMetric for ListenerMetric { Some(self.help.clone()) } - fn metric_type(&self) -> String { - self.metric_type.clone() + fn metric_type(&self) -> OpenMetricType { + self.metric_type } } @@ -100,8 +100,8 @@ mod tests { "pub_sub_listener_dropped", ] ); - assert_eq!(metrics[0].metric_type(), "gauge"); - assert_eq!(metrics[1].metric_type(), "counter"); - assert_eq!(metrics[2].metric_type(), "counter"); + assert_eq!(metrics[0].metric_type(), OpenMetricType::Gauge); + assert_eq!(metrics[1].metric_type(), OpenMetricType::Counter); + assert_eq!(metrics[2].metric_type(), OpenMetricType::Counter); } } diff --git a/pgdog/src/stats/lookup.rs b/pgdog/src/stats/lookup.rs index 7b1d1720d..d3963ec76 100644 --- a/pgdog/src/stats/lookup.rs +++ b/pgdog/src/stats/lookup.rs @@ -3,7 +3,7 @@ use std::sync::atomic::Ordering; -use crate::backend::databases::databases; +use crate::{backend::databases::databases, stats::OpenMetricType}; use super::{Measurement, Metric, OpenMetric}; @@ -14,13 +14,13 @@ pub struct LookupMetrics; struct Series { name: &'static str, help: &'static str, - metric_type: &'static str, + metric_type: OpenMetricType, measurements: Vec, global: u64, } impl Series { - fn new(name: &'static str, help: &'static str, metric_type: &'static str) -> Self { + fn new(name: &'static str, help: &'static str, metric_type: OpenMetricType) -> Self { Self { name, help, @@ -47,7 +47,7 @@ impl Series { name: self.name.into(), measurements: self.measurements, help: self.help.into(), - metric_type: self.metric_type.into(), + metric_type: self.metric_type, }) } } @@ -57,13 +57,13 @@ impl Series { struct TimeSeries { name: &'static str, help: &'static str, - metric_type: &'static str, + metric_type: OpenMetricType, measurements: Vec, global: u64, } impl TimeSeries { - fn new(name: &'static str, help: &'static str, metric_type: &'static str) -> Self { + fn new(name: &'static str, help: &'static str, metric_type: OpenMetricType) -> Self { Self { name, help, @@ -90,7 +90,7 @@ impl TimeSeries { name: self.name.into(), measurements: self.measurements, help: self.help.into(), - metric_type: self.metric_type.into(), + metric_type: self.metric_type, }) } } @@ -100,38 +100,38 @@ impl LookupMetrics { let mut hits = Series::new( "sharding_lookup_cache_hits", "Sharding key values translated from the lookup cache.", - "counter", + OpenMetricType::Counter, ); let mut misses = Series::new( "sharding_lookup_cache_misses", "Sharding key values that missed the lookup cache.", - "counter", + OpenMetricType::Counter, ); let mut evictions = Series::new( "sharding_lookup_cache_evictions", "Lookup cache entries evicted to stay within the memory bound.", - "counter", + OpenMetricType::Counter, ); let mut queries = Series::new( "sharding_lookup_queries", "Lookup queries run to resolve cache misses.", - "counter", + OpenMetricType::Counter, ); let mut query_time = TimeSeries::new( "sharding_lookup_query_time", "Total time spent running lookup queries, in milliseconds. \ Divided by sharding_lookup_queries, the average lookup latency.", - "counter", + OpenMetricType::Counter, ); let mut bytes = Series::new( "sharding_lookup_cache_bytes", "Approximate memory used by cached translations.", - "gauge", + OpenMetricType::Gauge, ); let mut entries = Series::new( "sharding_lookup_cache_entries", "Number of cached translations.", - "gauge", + OpenMetricType::Gauge, ); for (user, cluster) in databases().all() { @@ -169,7 +169,7 @@ struct LookupMetric { name: String, measurements: Vec, help: String, - metric_type: String, + metric_type: OpenMetricType, } impl OpenMetric for LookupMetric { @@ -185,8 +185,8 @@ impl OpenMetric for LookupMetric { Some(self.help.clone()) } - fn metric_type(&self) -> String { - self.metric_type.clone() + fn metric_type(&self) -> OpenMetricType { + self.metric_type } } diff --git a/pgdog/src/stats/mirror_stats.rs b/pgdog/src/stats/mirror_stats.rs index 63ede00ab..908ce0c2c 100644 --- a/pgdog/src/stats/mirror_stats.rs +++ b/pgdog/src/stats/mirror_stats.rs @@ -1,6 +1,6 @@ use crate::backend::databases::databases; -use super::{Measurement, Metric, OpenMetric}; +use super::{Measurement, Metric, OpenMetric, OpenMetricType}; pub struct MirrorStatsMetrics; @@ -96,35 +96,35 @@ impl MirrorStatsMetrics { name: "mirror_total_count".into(), measurements: total_count_measurements, help: "Total number of requests considered for mirroring.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, })); metrics.push(Metric::new(MirrorStatsMetric { name: "mirror_mirrored_count".into(), measurements: mirrored_count_measurements, help: "Total number of requests successfully mirrored.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, })); metrics.push(Metric::new(MirrorStatsMetric { name: "mirror_dropped_count".into(), measurements: dropped_count_measurements, help: "Total number of requests dropped due to exposure settings.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, })); metrics.push(Metric::new(MirrorStatsMetric { name: "mirror_error_count".into(), measurements: error_count_measurements, help: "Total number of mirror requests that encountered errors.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, })); metrics.push(Metric::new(MirrorStatsMetric { name: "mirror_queue_length".into(), measurements: queue_length_measurements, help: "Current number of transactions in the mirror queue.".into(), - metric_type: "gauge".into(), + metric_type: OpenMetricType::Gauge, })); metrics @@ -135,7 +135,7 @@ struct MirrorStatsMetric { name: String, measurements: Vec, help: String, - metric_type: String, + metric_type: OpenMetricType, } impl OpenMetric for MirrorStatsMetric { @@ -151,8 +151,8 @@ impl OpenMetric for MirrorStatsMetric { Some(self.help.clone()) } - fn metric_type(&self) -> String { - self.metric_type.clone() + fn metric_type(&self) -> OpenMetricType { + self.metric_type } } @@ -188,7 +188,7 @@ mod tests { }, ], help: "Total number of requests considered for mirroring.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }; let metric = Metric::new(metric); @@ -233,7 +233,7 @@ mod tests { name: "mirror_mirrored_count".into(), measurements, help: "Total number of requests successfully mirrored.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }; let metric = Metric::new(metric); @@ -254,7 +254,7 @@ mod tests { measurement: 10usize.into(), }], help: "Total number of requests considered for mirroring.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }; let mirrored = MirrorStatsMetric { @@ -264,7 +264,7 @@ mod tests { measurement: 5usize.into(), }], help: "Total number of requests successfully mirrored.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }; let dropped = MirrorStatsMetric { @@ -274,7 +274,7 @@ mod tests { measurement: 3usize.into(), }], help: "Total number of requests dropped due to exposure settings.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }; let error = MirrorStatsMetric { @@ -284,7 +284,7 @@ mod tests { measurement: 2usize.into(), }], help: "Total number of mirror requests that encountered errors.".into(), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }; let metrics = vec![ @@ -324,7 +324,7 @@ mod tests { measurement: value.into(), }], help: format!("Test metric for {}", name), - metric_type: "counter".into(), + metric_type: OpenMetricType::Counter, }; let metric = Metric::new(metric); @@ -356,7 +356,7 @@ mod tests { measurement: 5usize.into(), }], help: "Current number of transactions in the mirror queue.".into(), - metric_type: "gauge".into(), + metric_type: OpenMetricType::Gauge, }; let metric = Metric::new(metric); diff --git a/pgdog/src/stats/open_metric.rs b/pgdog/src/stats/open_metric.rs index b1707795e..24e09054f 100644 --- a/pgdog/src/stats/open_metric.rs +++ b/pgdog/src/stats/open_metric.rs @@ -1,31 +1,118 @@ //! Open metrics. -use std::ops::Deref; +use std::{ops::Deref, sync::Arc}; use crate::config::config; pub trait OpenMetric: Send + Sync { fn name(&self) -> String; + /// Metric measurement. fn measurements(&self) -> Vec; + /// Metric unit. fn unit(&self) -> Option { None } - fn metric_type(&self) -> String { - "gauge".into() + fn metric_type(&self) -> OpenMetricType { + OpenMetricType::Gauge } + fn help(&self) -> Option { None } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OpenMetricType { + Gauge, + Counter, + /// A distribution. Renders as several series (`_bucket`/`_sum`/`_count`) + /// rather than one, and carries a `MeasurementType::Histogram`. + Histogram, +} + +impl std::fmt::Display for OpenMetricType { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let s = match self { + OpenMetricType::Gauge => "gauge", + OpenMetricType::Counter => "counter", + OpenMetricType::Histogram => "histogram", + }; + f.write_str(s) + } +} + #[derive(Debug, Clone)] pub enum MeasurementType { Float(f64), Integer(i64), Millis(u128), + /// Distribution rendered as an OpenMetrics histogram. + /// + /// Boxed: the payload dwarfs the scalar variants, and one `Measurement` per + /// pool per metric family is built and deep-cloned on every scrape, so the + /// padding would be paid by every gauge and counter in the process. + Histogram(Box), +} + +/// A histogram observation set, in seconds. +/// +/// Bucket counts are per-bucket, as the source `pgdog_stats::Histogram` +/// reports them; the OpenMetrics renderer accumulates them on the way out via +/// `cumulative`. `bounds` excludes the implicit `+Inf` bucket, so +/// `buckets.len() == bounds.len() + 1`. +#[derive(Debug, Clone)] +pub struct HistogramMeasurement { + /// Upper bounds in seconds, ascending. Bounds are process-constant, so one + /// shared allocation backs every pool's measurement; cloning a measurement + /// only bumps the refcount. + pub bounds: Arc<[f64]>, + /// Per-bucket (non-cumulative) counts, with the `+Inf` bucket last. + pub buckets: Vec, + /// Sum of all observations, in seconds. + pub sum: f64, + /// Total number of observations. + pub count: u64, +} + +impl HistogramMeasurement { + /// Build a measurement from per-bucket (non-cumulative) counts. + pub fn new( + bounds: impl Into>, + per_bucket: impl Into>, + sum: f64, + count: u64, + ) -> Self { + Self { + bounds: bounds.into(), + buckets: per_bucket.into(), + sum, + count, + } + } +} + +/// Running sum of per-bucket counts. +/// +/// OpenMetrics `le` buckets are cumulative; OTLP's are not. Cumulative is a +/// property of the `le` wire format, not of the data, so the conversion lives +/// at the OpenMetrics boundary rather than in the measurement. +fn cumulative(per_bucket: &[u64]) -> Vec { + let mut buckets = Vec::with_capacity(per_bucket.len()); + let mut running = 0u64; + for bucket in per_bucket { + running = running.saturating_add(*bucket); + buckets.push(running); + } + buckets +} + +impl From for MeasurementType { + fn from(value: HistogramMeasurement) -> Self { + Self::Histogram(Box::new(value)) + } } impl From for MeasurementType { @@ -58,6 +145,23 @@ impl From for MeasurementType { } } +impl MeasurementType { + /// Whether a measurement of this shape can be exported under a metric + /// declared as `metric_type`. + /// + /// Histograms render and encode completely differently from scalars, so + /// the two must agree: a `Histogram` measurement only fits a `Histogram` + /// metric, and scalar measurements only fit `Gauge` or `Counter`. + fn matches(&self, metric_type: OpenMetricType) -> bool { + match metric_type { + OpenMetricType::Histogram => matches!(self, MeasurementType::Histogram(_)), + OpenMetricType::Gauge | OpenMetricType::Counter => { + !matches!(self, MeasurementType::Histogram(_)) + } + } + } +} + #[derive(Debug, Clone)] pub struct Measurement { pub labels: Vec<(String, String)>, @@ -66,26 +170,96 @@ pub struct Measurement { impl Measurement { pub fn render(&self, name: &str) -> String { - let labels = if self.labels.is_empty() { - "".into() - } else { - let labels = self - .labels - .iter() - .map(|(name, value)| format!("{}=\"{}\"", name, value)) - .collect::>(); - format!("{{{}}}", labels.join(",")) + let value = match &self.measurement { + // Histograms render as several lines, so they bypass the scalar format. + MeasurementType::Histogram(histogram) => return self.render_histogram(name, histogram), + MeasurementType::Float(f) => format!("{:.3}", f), + MeasurementType::Integer(i) => i.to_string(), + MeasurementType::Millis(i) => i.to_string(), }; - format!( - "{}{} {}", + + format!("{}{} {}", name, self.render_labels(&[]), value) + } + + /// Render the label set, appending `extra` labels after this + /// measurement's own. + fn render_labels(&self, extra: &[(&str, String)]) -> String { + if self.labels.is_empty() && extra.is_empty() { + return String::new(); + } + + let labels = self + .labels + .iter() + .map(|(name, value)| format!("{}=\"{}\"", name, value)) + .chain( + extra + .iter() + .map(|(name, value)| format!("{}=\"{}\"", name, value)), + ) + .collect::>(); + + format!("{{{}}}", labels.join(",")) + } + + /// Render `_bucket`, `_sum` and `_count` series for a histogram. + /// + /// Per the OpenMetrics spec, bucket counts are cumulative and the final + /// bucket must be `le="+Inf"`. + fn render_histogram(&self, name: &str, histogram: &HistogramMeasurement) -> String { + let mut lines = Vec::with_capacity(histogram.buckets.len() + 2); + let buckets = cumulative(&histogram.buckets); + + // `bounds` excludes +Inf, so zip stops one short and leaves the + // overflow bucket for the explicit +Inf line below. + for (bound, count) in histogram.bounds.iter().zip(buckets.iter()) { + lines.push(format!( + "{}_bucket{} {}", + name, + self.render_labels(&[("le", format_bound(*bound))]), + count + )); + } + + lines.push(format!( + "{}_bucket{} {}", name, - labels, - match self.measurement { - MeasurementType::Float(f) => format!("{:.3}", f), - MeasurementType::Integer(i) => i.to_string(), - MeasurementType::Millis(i) => i.to_string(), - } - ) + self.render_labels(&[("le", "+Inf".into())]), + histogram.count + )); + + lines.push(format!( + "{}_sum{} {:.6}", + name, + self.render_labels(&[]), + histogram.sum + )); + lines.push(format!( + "{}_count{} {}", + name, + self.render_labels(&[]), + histogram.count + )); + + lines.join("\n") + } +} + +/// Format a bucket bound without losing sub-millisecond precision. +/// +/// Bounds are seconds, so a default `{}` on `0.0001` would render as +/// `0.0001` but `1e-5` in scientific notation, which Prometheus rejects. +/// Nine decimals preserve `Duration`'s nanosecond resolution, so two +/// distinct bounds can never collapse to the same `le` label — duplicate +/// series fail the entire Prometheus scrape. +fn format_bound(bound: f64) -> String { + let formatted = format!("{:.9}", bound); + let trimmed = formatted.trim_end_matches('0').trim_end_matches('.'); + + if trimmed.is_empty() { + "0".into() + } else { + trimmed.to_owned() } } @@ -95,9 +269,23 @@ pub struct Metric { impl Metric { pub fn new(metric: impl OpenMetric + 'static) -> Self { - Self { - metric: Box::new(metric), - } + let metric: Box = Box::new(metric); + + // Exporters branch on `metric_type()` and then pattern-match each + // measurement, so a disagreement between the two fails silently: + // `# TYPE … gauge` followed by `_bucket` lines, a histogram exported + // as `0.0` over OTLP, or scalars dropped from a `Histogram` metric. + debug_assert!( + metric + .measurements() + .iter() + .all(|m| m.measurement.matches(metric.metric_type())), + "{:?} is typed {:?} but carries incompatible measurements", + metric.name(), + metric.metric_type(), + ); + + Self { metric } } } @@ -128,7 +316,11 @@ impl std::fmt::Display for Metric { } for measurement in self.measurements() { - writeln!(f, "{}{}", prefix, measurement.render(&name))?; + // A measurement can render as several lines (histograms), and each + // one needs the namespace prefix. + for line in measurement.render(&name).lines() { + writeln!(f, "{}{}", prefix, line)?; + } } Ok(()) } @@ -167,6 +359,48 @@ mod test { let render = Metric::new(TestMetric {}).to_string(); assert_eq!(render.lines().next().unwrap(), "# TYPE pgdog.test gauge"); assert_eq!(render.lines().last().unwrap(), "pgdog.test 5"); + + // A histogram renders as several lines, and every one needs the + // prefix. Asserted here rather than in its own test because the + // namespace is global state. + struct TestHistogram; + + impl OpenMetric for TestHistogram { + fn name(&self) -> String { + "query_time_seconds".into() + } + + fn metric_type(&self) -> OpenMetricType { + OpenMetricType::Histogram + } + + fn measurements(&self) -> Vec { + vec![Measurement { + labels: vec![], + measurement: test_histogram().into(), + }] + } + } + + let render = Metric::new(TestHistogram {}).to_string(); + assert_eq!( + render.lines().next().unwrap(), + "# TYPE pgdog.query_time_seconds histogram" + ); + for line in render.lines().filter(|line| !line.starts_with('#')) { + assert!( + line.starts_with("pgdog.query_time_seconds"), + "missing prefix: {}", + line + ); + } + assert_eq!( + render + .lines() + .filter(|line| line.contains("_bucket")) + .count(), + 4 + ); } #[test] @@ -193,4 +427,213 @@ mod test { let rendered = measurement.render("query_latency_seconds"); assert_eq!(rendered, "query_latency_seconds 1.235"); } + + fn test_histogram() -> HistogramMeasurement { + // Per-bucket counts 1/2/0, plus 1 in the +Inf bucket. + HistogramMeasurement::new(vec![0.001, 0.01, 0.1], [1, 2, 0, 1], 1.5, 4) + } + + #[test] + fn histogram_new_preserves_per_bucket_counts() { + let histogram = test_histogram(); + + // Stored verbatim, matching the source `pgdog_stats::Histogram`. + // Accumulating is the OpenMetrics renderer's job, not the + // measurement's — OTLP wants these counts as they are. + assert_eq!(histogram.buckets, vec![1, 2, 0, 1]); + assert_eq!(histogram.count, 4); + } + + #[test] + fn cumulative_accumulates_per_bucket_counts() { + assert_eq!(cumulative(&[1, 2, 0, 1]), vec![1, 3, 3, 4]); + assert_eq!(cumulative(&[]), Vec::::new()); + assert_eq!(cumulative(&[0, 0, 0]), vec![0, 0, 0]); + + // Never decreases, whatever the inputs. + let counts = cumulative(&[3, 0, 7, 0, 0, 11]); + assert!( + counts.windows(2).all(|pair| pair[0] <= pair[1]), + "must be monotonic: {:?}", + counts + ); + } + + #[test] + fn cumulative_saturates_instead_of_overflowing() { + assert_eq!(cumulative(&[u64::MAX, 1]), vec![u64::MAX, u64::MAX]); + } + + #[test] + fn last_cumulative_bucket_agrees_with_count() { + // The `+Inf` line renders `count`, not the last bucket, so the two + // reach the wire by different routes and must not disagree. + let histogram = test_histogram(); + let buckets = cumulative(&histogram.buckets); + + assert_eq!(buckets.last().copied(), Some(histogram.count)); + + let rendered = Measurement { + labels: vec![], + measurement: histogram.into(), + } + .render("query_time_seconds"); + + let value_of = |suffix: &str| -> u64 { + rendered + .lines() + .find(|line| line.starts_with(&format!("query_time_seconds{}", suffix))) + .and_then(|line| line.rsplit_once(' ')) + .expect("line") + .1 + .parse() + .expect("numeric") + }; + + assert_eq!(value_of("_bucket{le=\"+Inf\"}"), value_of("_count")); + } + + #[test] + fn histogram_render_emits_buckets_sum_and_count() { + let measurement = Measurement { + labels: vec![("database".into(), "app".into())], + measurement: test_histogram().into(), + }; + + let rendered = measurement.render("query_time_seconds"); + let lines: Vec<&str> = rendered.lines().collect(); + + assert_eq!( + lines, + vec![ + r#"query_time_seconds_bucket{database="app",le="0.001"} 1"#, + r#"query_time_seconds_bucket{database="app",le="0.01"} 3"#, + r#"query_time_seconds_bucket{database="app",le="0.1"} 3"#, + r#"query_time_seconds_bucket{database="app",le="+Inf"} 4"#, + r#"query_time_seconds_sum{database="app"} 1.500000"#, + r#"query_time_seconds_count{database="app"} 4"#, + ] + ); + } + + #[test] + fn histogram_render_without_labels() { + let measurement = Measurement { + labels: vec![], + measurement: test_histogram().into(), + }; + + let rendered = measurement.render("query_time_seconds"); + let lines: Vec<&str> = rendered.lines().collect(); + + assert_eq!(lines[0], r#"query_time_seconds_bucket{le="0.001"} 1"#); + assert_eq!(lines[3], r#"query_time_seconds_bucket{le="+Inf"} 4"#); + assert_eq!(lines[4], "query_time_seconds_sum 1.500000"); + assert_eq!(lines[5], "query_time_seconds_count 4"); + } + + #[test] + fn histogram_bucket_counts_are_monotonic() { + let measurement = Measurement { + labels: vec![], + measurement: test_histogram().into(), + }; + + let counts: Vec = measurement + .render("query_time_seconds") + .lines() + .filter(|line| line.contains("_bucket")) + .map(|line| { + line.rsplit_once(' ') + .expect("value") + .1 + .parse() + .expect("numeric") + }) + .collect(); + + assert!( + counts.windows(2).all(|pair| pair[0] <= pair[1]), + "bucket counts must be cumulative: {:?}", + counts + ); + assert_eq!(counts.last(), Some(&4)); + } + + #[test] + fn bound_formatting_avoids_scientific_notation() { + assert_eq!(format_bound(0.0001), "0.0001"); + assert_eq!(format_bound(0.001), "0.001"); + assert_eq!(format_bound(1.0), "1"); + assert_eq!(format_bound(30.0), "30"); + } + + #[test] + fn bound_formatting_distinguishes_sub_microsecond_bounds() { + use std::time::Duration; + + // Distinct Durations must never collapse to the same le label: + // duplicates fail the entire Prometheus scrape. + assert_eq!( + format_bound(Duration::from_nanos(100).as_secs_f64()), + "0.0000001" + ); + assert_ne!( + format_bound(Duration::from_nanos(100).as_secs_f64()), + format_bound(Duration::from_nanos(200).as_secs_f64()) + ); + assert_eq!(format_bound(1.1e-6), "0.0000011"); + assert_ne!(format_bound(1.1e-6), format_bound(1.2e-6)); + } + + // The disagreement check lives in a debug_assert!, so these only panic + // when debug assertions are enabled. + #[cfg(debug_assertions)] + #[test] + #[should_panic(expected = "carries incompatible measurements")] + fn metric_new_rejects_histogram_measurement_in_gauge_metric() { + struct BadMetric; + + impl OpenMetric for BadMetric { + fn name(&self) -> String { + "bad".into() + } + + // metric_type() defaults to Gauge. + fn measurements(&self) -> Vec { + vec![Measurement { + labels: vec![], + measurement: test_histogram().into(), + }] + } + } + + let _ = Metric::new(BadMetric); + } + + #[cfg(debug_assertions)] + #[test] + #[should_panic(expected = "carries incompatible measurements")] + fn metric_new_rejects_scalar_measurement_in_histogram_metric() { + struct BadMetric; + + impl OpenMetric for BadMetric { + fn name(&self) -> String { + "bad".into() + } + + fn metric_type(&self) -> OpenMetricType { + OpenMetricType::Histogram + } + + fn measurements(&self) -> Vec { + vec![Measurement { + labels: vec![], + measurement: MeasurementType::Integer(1), + }] + } + } + + let _ = Metric::new(BadMetric); + } } diff --git a/pgdog/src/stats/otel.rs b/pgdog/src/stats/otel.rs index 407c39a67..7839a8c91 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -5,15 +5,19 @@ use std::collections::HashMap; use std::env; +use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; use once_cell::sync::Lazy; use parking_lot::Mutex; +use pgdog_config::otel_temporality::OtelTemporalityPreference; use serde::Serialize; use crate::util::hostname; -use super::open_metric::{MeasurementType, Metric}; +use super::open_metric::{ + HistogramMeasurement, Measurement, MeasurementType, Metric, OpenMetricType, +}; static RESOURCE_ATTRIBUTES: Lazy> = Lazy::new(resource_attributes); @@ -27,9 +31,95 @@ struct CounterKey { labels: Vec<(String, String)>, } -/// Previous cumulative values for delta computation. -static PREV_COUNTERS: Lazy>> = - Lazy::new(|| Mutex::new(HashMap::new())); +/// Per-data-point counter bookkeeping: previous cumulative values (for delta +/// computation) and first-seen timestamps (used as `start_time_unix_nano` so +/// cumulative counters carry a stable collection-start reference). +/// +/// None of these maps is evicted, so a pool that goes away keeps its entry for +/// the life of the process. Eviction should cover all of them together: they +/// are keyed identically, and expiring one but not the others would leave the +/// same series half-forgotten. +#[derive(Default)] +struct CounterState { + prev_values: Mutex>, + prev_histograms: Mutex>, + start_times: Mutex>, +} + +impl CounterState { + fn start_time(&self, key: &CounterKey, now: &str) -> String { + self.start_times + .lock() + .entry(key.clone()) + .or_insert_with(|| now.to_string()) + .clone() + } + + /// Delta since the previously recorded cumulative value. Updates the + /// stored value as a side effect. Returns `None` on a negative delta + /// (counter reset), which callers should treat as a skipped data point. + fn delta(&self, key: &CounterKey, cumulative: f64) -> Option { + let mut prev = self.prev_values.lock(); + let delta = cumulative - prev.get(key).copied().unwrap_or(0.0); + prev.insert(key.clone(), cumulative); + (delta >= 0.0).then_some(delta) + } + + /// Delta since the previously recorded cumulative distribution. Updates + /// the stored value as a side effect. + /// + /// Deliberately stricter than [`CounterState::delta`] on first sight: a + /// counter reports its lifetime total as the first delta, which is a + /// legitimate interval value, whereas a histogram's lifetime bucket + /// distribution reported as a single interval would skew per-interval + /// percentiles. So the first export only latches state and returns `None`. + /// + /// Also returns `None` after a reset — pool recreated, or bucket layout + /// changed — so one interval is skipped rather than emitting negative + /// counts. + fn histogram_delta( + &self, + key: &CounterKey, + cumulative: HistogramState, + ) -> Option { + let previous = self + .prev_histograms + .lock() + .insert(key.clone(), cumulative.clone())?; + + if cumulative.count < previous.count + || cumulative.sum < previous.sum + || cumulative.buckets.len() != previous.buckets.len() + { + return None; + } + + Some(HistogramState { + buckets: cumulative + .buckets + .iter() + .zip(previous.buckets.iter()) + .map(|(cumulative, previous)| cumulative.saturating_sub(*previous)) + .collect(), + sum: cumulative.sum - previous.sum, + count: cumulative.count - previous.count, + }) + } +} + +static COUNTER_STATE: Lazy = Lazy::new(CounterState::default); + +/// Bucket counts, sum and observation count for one histogram data point. +/// +/// Holds either a cumulative snapshot or a per-interval delta, depending on +/// where it came from. Bucket counts are per-bucket, as OTLP expects, not the +/// cumulative `le` counts the OpenMetrics endpoint renders. +#[derive(Clone, Default)] +struct HistogramState { + buckets: Vec, + sum: f64, + count: u64, +} pub fn now_nanos() -> String { SystemTime::now() @@ -81,6 +171,8 @@ pub struct OtelMetric { pub gauge: Option, #[serde(skip_serializing_if = "Option::is_none")] pub sum: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub histogram: Option, } #[derive(Serialize)] @@ -89,15 +181,66 @@ pub struct Gauge { pub data_points: Vec, } +// little serde trick to let us serialize directly as the integer representation +#[derive(serde_repr::Serialize_repr, Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +pub enum AggregationTemporality { + Delta = 1, + Cumulative = 2, +} + #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct Sum { - /// 1 = DELTA, 2 = CUMULATIVE - pub aggregation_temporality: u32, + pub aggregation_temporality: AggregationTemporality, pub is_monotonic: bool, pub data_points: Vec, } +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct Histogram { + // Same OTLP enum as `Sum` carries: the temporality choice is resolved once + // per export, so sums and histograms in a batch always agree. + pub aggregation_temporality: AggregationTemporality, + pub data_points: Vec, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct HistogramDataPoint { + #[serde(skip_serializing_if = "Option::is_none")] + pub start_time_unix_nano: Option, + pub time_unix_nano: String, + #[serde(serialize_with = "u64_to_string")] + pub count: u64, + pub sum: f64, + /// Per-bucket counts; one longer than `explicit_bounds`. + #[serde(serialize_with = "u64s_to_strings")] + pub bucket_counts: Vec, + /// Upper bounds, excluding the implicit `+Inf` bucket. Shared with the + /// source measurement rather than copied per data point; serializes as a + /// plain JSON array. + pub explicit_bounds: Arc<[f64]>, + pub attributes: Vec, +} + +/// OTLP/JSON (protojson) encodes 64-bit integers as decimal strings. +fn u64_to_string(value: &u64, serializer: S) -> Result { + serializer.collect_str(value) +} + +/// Same, for a sequence of 64-bit integers. +fn u64s_to_strings(values: &[u64], serializer: S) -> Result { + use serde::ser::SerializeSeq; + + let mut seq = serializer.serialize_seq(Some(values.len()))?; + for value in values { + seq.serialize_element(&value.to_string())?; + } + seq.end() +} + #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct NumberDataPoint { @@ -199,19 +342,165 @@ fn measurement_to_f64(m: &MeasurementType) -> f64 { MeasurementType::Float(f) => *f, MeasurementType::Integer(i) => *i as f64, MeasurementType::Millis(ms) => *ms as f64, + // Histograms are exported as their own data point type, never as a + // single number. + MeasurementType::Histogram(_) => 0.0, + } +} + +/// Compute `(as_double, start_time_unix_nano)` for a single measurement. +/// +/// Returns `None` to skip this data point (only possible for a Delta counter +/// that just observed a reset — a negative delta). +fn value_for_data_point( + state: &CounterState, + metric_name: &str, + measurement: &Measurement, + metric_type: OpenMetricType, + temporality: AggregationTemporality, + now: &str, +) -> Option<(f64, Option)> { + let cumulative = measurement_to_f64(&measurement.measurement); + + match metric_type { + OpenMetricType::Gauge => Some((cumulative, None)), + OpenMetricType::Counter => { + let key = CounterKey { + metric: metric_name.into(), + labels: measurement.labels.clone(), + }; + let start = state.start_time(&key, now); + let value = match temporality { + AggregationTemporality::Cumulative => cumulative, + AggregationTemporality::Delta => state.delta(&key, cumulative)?, + }; + Some((value, Some(start))) + } + // A distribution has no scalar value, so it cannot become a + // `NumberDataPoint`. Callers fork to `histogram_data_point` before + // reaching here. + OpenMetricType::Histogram => None, + } +} + +/// Wrap a collection of data points in the correct OTLP container based on +/// the source metric type: `Gauge` for gauges, `Sum` (monotonic) for counters. +fn wrap_data_points( + metric_type: OpenMetricType, + temporality: AggregationTemporality, + data_points: Vec, +) -> (Option, Option) { + match metric_type { + OpenMetricType::Gauge => (Some(Gauge { data_points }), None), + OpenMetricType::Counter => ( + None, + Some(Sum { + aggregation_temporality: temporality, + is_monotonic: true, + data_points, + }), + ), + // Neither container fits a distribution; the `Histogram` container is + // built directly by the caller, which never routes one here. + OpenMetricType::Histogram => (None, None), } } -/// Build an `ExportMetricsServiceRequest` from a collection of `Metric` objects. +/// Merge the measurement's own labels with the process-wide resource +/// attributes into the OTLP `attributes` field for a data point. +fn build_attributes(labels: &[(String, String)], common_attrs: &[KeyValue]) -> Vec { + let mut attributes: Vec = labels + .iter() + .map(|(k, v)| KeyValue { + key: k.clone(), + value: AttributeValue { + string_value: v.clone(), + }, + }) + .collect(); + attributes.extend(common_attrs.iter().cloned()); + attributes +} + +/// Build one OTLP histogram data point from a cumulative measurement. +/// +/// Pool counts accumulate since the pool was created, so the measurement is +/// always cumulative. Cumulative exports pass it straight through; only the +/// Delta path needs per-export bookkeeping, and it may report nothing — see +/// [`CounterState::histogram_delta`]. +fn histogram_data_point( + state: &CounterState, + metric_name: &str, + measurement: &Measurement, + histogram: &HistogramMeasurement, + temporality: AggregationTemporality, + now: &str, + attributes: Vec, +) -> Option { + // The measurement already carries per-bucket counts, which is what OTLP + // wants; only the OpenMetrics `le` format needs them accumulated. + let cumulative = HistogramState { + buckets: histogram.buckets.clone(), + sum: histogram.sum, + count: histogram.count, + }; + + let key = CounterKey { + metric: metric_name.to_owned(), + labels: measurement.labels.clone(), + }; + + // Pin the collection-start reference before the Delta path can bail, so a + // series that starts reporting later still carries its original start. + let start_time_unix_nano = Some(state.start_time(&key, now)); + + let values = match temporality { + AggregationTemporality::Cumulative => cumulative, + AggregationTemporality::Delta => state.histogram_delta(&key, cumulative)?, + }; + + Some(HistogramDataPoint { + start_time_unix_nano, + time_unix_nano: now.to_owned(), + count: values.count, + sum: values.sum, + bucket_counts: values.buckets, + explicit_bounds: histogram.bounds.clone(), + attributes, + }) +} + +/// Build an `ExportMetricsServiceRequest` from a collection of `Metric` objects, +/// reading namespace and temporality preference from the global config and +/// using the process-wide counter bookkeeping. `now` is threaded in from the +/// caller so every data point in a batch shares one timestamp. pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequest { let config = crate::config::config(); - let namespace = config - .config - .otel - .namespace - .as_deref() - .unwrap_or("pgdog") - .trim_end_matches(['.', '_']); + let otel = &config.config.otel; + + let temporality = match otel.effective_temporality_preference() { + OtelTemporalityPreference::Cumulative => AggregationTemporality::Cumulative, + OtelTemporalityPreference::Delta | OtelTemporalityPreference::LowMemory => { + AggregationTemporality::Delta + } + }; + + let namespace = otel.namespace.as_deref(); + + build_request_with_state(&COUNTER_STATE, temporality, namespace, now, metrics) +} + +/// Injectable core of [`build_request`]. Takes counter state, temporality, +/// and namespace explicitly so tests can exercise the stateful counter logic +/// without touching global config or the process-wide static. +fn build_request_with_state( + state: &CounterState, + temporality: AggregationTemporality, + namespace: Option<&str>, + now: &str, + metrics: &[&Metric], +) -> ExportMetricsServiceRequest { + let namespace = namespace.unwrap_or("pgdog").trim_end_matches(['.', '_']); let namespace = if namespace.is_empty() { "pgdog" } else { @@ -224,70 +513,69 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ .iter() .map(|metric| { let name = format!("{}.{}", namespace, metric.name()); - let is_counter = metric.metric_type() == "counter"; + let metric_type = metric.metric_type(); + + // A distribution becomes a `HistogramDataPoint`, which shares no + // fields with the `NumberDataPoint` the other two shapes produce, so + // the paths fork before any data point is built. + let (gauge, sum, histogram) = match metric_type { + OpenMetricType::Histogram => { + let data_points = metric + .measurements() + .iter() + .filter_map(|m| { + let MeasurementType::Histogram(ref histogram) = m.measurement else { + return None; + }; + + histogram_data_point( + state, + &name, + m, + histogram, + temporality, + now, + build_attributes(&m.labels, common_attrs), + ) + }) + .collect(); - let data_points: Vec = metric - .measurements() - .iter() - .filter_map(|m| { - let cumulative = measurement_to_f64(&m.measurement); - - let as_double = if is_counter { - let key = CounterKey { - metric: name.clone(), - labels: m.labels.clone(), - }; - let mut prev = PREV_COUNTERS.lock(); - let delta = cumulative - prev.get(&key).copied().unwrap_or(0.0); - prev.insert(key, cumulative); - - // Skip negative deltas (counter reset). - if delta < 0.0 { - return None; - } - delta - } else { - cumulative - }; - - let mut attributes: Vec = m - .labels + ( + None, + None, + Some(Histogram { + aggregation_temporality: temporality, + data_points, + }), + ) + } + + OpenMetricType::Gauge | OpenMetricType::Counter => { + let data_points: Vec = metric + .measurements() .iter() - .map(|(k, v)| KeyValue { - key: k.clone(), - value: AttributeValue { - string_value: v.clone(), - }, + .filter_map(|m| { + let (as_double, start_time_unix_nano) = value_for_data_point( + state, + &name, + m, + metric_type, + temporality, + now, + )?; + + Some(NumberDataPoint { + start_time_unix_nano, + time_unix_nano: now.to_owned(), + as_double, + attributes: build_attributes(&m.labels, common_attrs), + }) }) .collect(); - attributes.extend(common_attrs.iter().map(|a| KeyValue { - key: a.key.clone(), - value: AttributeValue { - string_value: a.value.string_value.clone(), - }, - })); - - Some(NumberDataPoint { - start_time_unix_nano: None, - time_unix_nano: now.to_owned(), - as_double, - attributes, - }) - }) - .collect(); - - let (gauge, sum) = if is_counter { - ( - None, - Some(Sum { - aggregation_temporality: 1, // DELTA - is_monotonic: true, - data_points, - }), - ) - } else { - (Some(Gauge { data_points }), None) + let (gauge, sum) = wrap_data_points(metric_type, temporality, data_points); + (gauge, sum, None) + } }; OtelMetric { @@ -296,6 +584,7 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ unit: metric.unit().unwrap_or_else(|| "1".into()), gauge, sum, + histogram, } }) .collect(); @@ -353,6 +642,11 @@ mod test { fn counter_metric_produces_sum_json() { let _test_lock = TEST_LOCK.lock(); + use crate::config::{self, ConfigAndUsers}; + let mut cfg = ConfigAndUsers::default(); + cfg.config.otel.temporality_preference = Some(OtelTemporalityPreference::Delta); + config::set(cfg).expect("set config"); + let metric = Metric::new(PoolMetric { name: "total_query_count".into(), measurements: vec![Measurement { @@ -361,7 +655,7 @@ mod test { }], help: "Total queries".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), }); let request = build_request(&[&metric], &now_nanos()); @@ -454,7 +748,7 @@ mod test { }], help: "Transaction time".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), }); let request = build_request(&[&metric], &now_nanos()); @@ -561,6 +855,222 @@ mod test { } } + fn histogram_metric(sum: f64, count: u64, per_bucket: &[u64]) -> Metric { + Metric::new(PoolMetric { + name: "query_time_seconds".into(), + measurements: vec![Measurement { + labels: vec![("database".into(), "app".into())], + measurement: HistogramMeasurement::new(vec![0.001, 0.01], per_bucket, sum, count) + .into(), + }], + help: "Query times".into(), + unit: Some("seconds".into()), + metric_type: Some(OpenMetricType::Histogram), + }) + } + + fn export_histogram( + state: &CounterState, + temporality: AggregationTemporality, + metric: &Metric, + ) -> ExportMetricsServiceRequest { + export_histogram_at(state, temporality, &now_nanos(), metric) + } + + /// Export with a caller-supplied collection timestamp, for tests that + /// assert on which `now` a series records. + fn export_histogram_at( + state: &CounterState, + temporality: AggregationTemporality, + now: &str, + metric: &Metric, + ) -> ExportMetricsServiceRequest { + build_request_with_state(state, temporality, None, now, &[metric]) + } + + fn only_histogram(req: &ExportMetricsServiceRequest) -> &Histogram { + req.resource_metrics[0].scope_metrics[0].metrics[0] + .histogram + .as_ref() + .expect("histogram") + } + + #[test] + fn cumulative_histogram_passes_counts_through() { + let state = CounterState::default(); + + // Cumulative is the default temporality, so this is the common path: the + // measurement is already cumulative-since-pool-creation and needs no + // bookkeeping, including on the very first export. + let metric = histogram_metric(0.5, 3, &[1, 1, 1]); + let request = export_histogram(&state, AggregationTemporality::Cumulative, &metric); + let histogram = only_histogram(&request); + + assert_eq!( + histogram.aggregation_temporality, + AggregationTemporality::Cumulative + ); + assert_eq!(histogram.data_points.len(), 1); + + let point = &histogram.data_points[0]; + assert_eq!(point.count, 3); + assert!((point.sum - 0.5).abs() < f64::EPSILON); + // OTLP wants per-bucket counts even though the measurement is cumulative. + assert_eq!(point.bucket_counts, vec![1, 1, 1]); + assert_eq!(&point.explicit_bounds[..], [0.001, 0.01]); + // bucketCounts is always one longer than explicitBounds. + assert_eq!(point.bucket_counts.len(), point.explicit_bounds.len() + 1); + } + + #[test] + fn cumulative_histogram_accumulates_across_exports() { + let state = CounterState::default(); + + let first = histogram_metric(0.5, 3, &[1, 1, 1]); + export_histogram(&state, AggregationTemporality::Cumulative, &first); + + let second = histogram_metric(1.5, 5, &[2, 1, 2]); + let request = export_histogram(&state, AggregationTemporality::Cumulative, &second); + + // Reported as-is, not diffed. + let point = &only_histogram(&request).data_points[0]; + assert_eq!(point.count, 5); + assert_eq!(point.bucket_counts, vec![2, 1, 2]); + } + + #[test] + fn delta_histogram_first_export_is_skipped() { + let state = CounterState::default(); + + let metric = histogram_metric(0.5, 3, &[1, 1, 1]); + let request = export_histogram(&state, AggregationTemporality::Delta, &metric); + + let otel_metric = &request.resource_metrics[0].scope_metrics[0].metrics[0]; + + // No prior sample to diff against. Reporting the lifetime distribution as + // one interval would skew percentiles, so nothing is reported yet. + assert!( + otel_metric + .histogram + .as_ref() + .expect("histogram") + .data_points + .is_empty() + ); + assert!(otel_metric.gauge.is_none()); + assert!(otel_metric.sum.is_none()); + } + + #[test] + fn delta_histogram_exports_delta_against_previous() { + let state = CounterState::default(); + + // Prime the previous state. + let first = histogram_metric(0.5, 3, &[1, 1, 1]); + export_histogram(&state, AggregationTemporality::Delta, &first); + + // Two more observations: one in the first bucket, one in +Inf. + let second = histogram_metric(1.5, 5, &[2, 1, 2]); + let request = export_histogram(&state, AggregationTemporality::Delta, &second); + let histogram = only_histogram(&request); + + assert_eq!( + histogram.aggregation_temporality, + AggregationTemporality::Delta + ); + assert_eq!(histogram.data_points.len(), 1); + + let point = &histogram.data_points[0]; + assert_eq!(point.count, 2); + assert!((point.sum - 1.0).abs() < f64::EPSILON); + assert_eq!(point.bucket_counts, vec![1, 0, 1]); + assert_eq!(&point.explicit_bounds[..], [0.001, 0.01]); + assert_eq!(point.bucket_counts.len(), point.explicit_bounds.len() + 1); + } + + #[test] + fn delta_histogram_skips_counter_reset() { + let state = CounterState::default(); + + let first = histogram_metric(1.5, 5, &[2, 1, 2]); + export_histogram(&state, AggregationTemporality::Delta, &first); + + // Pool recreated: counts went backwards. + let reset = histogram_metric(0.1, 1, &[1, 0, 0]); + let request = export_histogram(&state, AggregationTemporality::Delta, &reset); + + assert!(only_histogram(&request).data_points.is_empty()); + } + + #[test] + fn histogram_start_time_survives_a_skipped_delta() { + let state = CounterState::default(); + + // Fixed timestamps: the property is *which* `now` gets pinned, not + // whether the wall clock happened to advance between two exports. + let first_scrape = "1700000000000000000"; + let second_scrape = "1700000015000000000"; + + // The first Delta export reports nothing, but must still pin the + // collection start so the series that follows carries its original one. + let first = histogram_metric(0.5, 3, &[1, 1, 1]); + let request = + export_histogram_at(&state, AggregationTemporality::Delta, first_scrape, &first); + assert!(only_histogram(&request).data_points.is_empty()); + + let second = histogram_metric(1.0, 4, &[2, 1, 1]); + let request = export_histogram_at( + &state, + AggregationTemporality::Delta, + second_scrape, + &second, + ); + + // The start time is the *first* scrape's, not the second's: a consumer + // must see one unbroken series, not one that began 15s in. + let point = &only_histogram(&request).data_points[0]; + assert_eq!(point.start_time_unix_nano.as_deref(), Some(first_scrape)); + assert_eq!(point.time_unix_nano, second_scrape); + } + + #[test] + fn histogram_carries_labels_and_resource_attributes() { + let state = CounterState::default(); + + let metric = histogram_metric(0.5, 3, &[1, 1, 1]); + let request = export_histogram(&state, AggregationTemporality::Cumulative, &metric); + + let point = &only_histogram(&request).data_points[0]; + + assert_eq!(point.attributes[0].key, "database"); + assert_eq!(point.attributes[0].value.string_value, "app"); + assert!(point.attributes.iter().any(|a| a.key == "service.name")); + } + + #[test] + fn histogram_serializes_to_valid_otlp_json() { + let state = CounterState::default(); + + let first = histogram_metric(0.5, 3, &[1, 1, 1]); + export_histogram(&state, AggregationTemporality::Delta, &first); + let second = histogram_metric(1.0, 4, &[2, 1, 1]); + let request = export_histogram(&state, AggregationTemporality::Delta, &second); + + let json = serde_json::to_string(&request).expect("serialize"); + + assert!(json.contains("\"histogram\"")); + assert!(json.contains("\"bucketCounts\"")); + assert!(json.contains("\"explicitBounds\"")); + assert!(!json.contains("\"gauge\"")); + assert!(!json.contains("\"sum\":null")); + + // OTLP/JSON encodes 64-bit integers as decimal strings. + assert!(json.contains("\"count\":\"1\"")); + assert!(json.contains("\"bucketCounts\":[\"1\",\"0\",\"0\"]")); + + let _: serde_json::Value = serde_json::from_str(&json).expect("valid json"); + } + #[test] fn percent_decode_handles_encoded_chars() { let _test_lock = TEST_LOCK.lock(); @@ -569,4 +1079,199 @@ mod test { assert_eq!(percent_decode("a%2Cb"), "a,b"); assert_eq!(percent_decode("plain"), "plain"); } + + fn counter(name: &str, labels: Vec<(String, String)>, value: i64) -> Metric { + Metric::new(PoolMetric { + name: name.into(), + measurements: vec![Measurement { + labels, + measurement: MeasurementType::Integer(value), + }], + help: "".into(), + unit: None, + metric_type: Some(OpenMetricType::Counter), + }) + } + + fn only_data_point(req: &ExportMetricsServiceRequest) -> &NumberDataPoint { + &req.resource_metrics[0].scope_metrics[0].metrics[0] + .sum + .as_ref() + .expect("sum") + .data_points[0] + } + + #[test] + fn delta_subtracts_previous_cumulative_value() { + let state = CounterState::default(); + + let m1 = counter("total_queries", vec![], 10); + let r1 = build_request_with_state( + &state, + AggregationTemporality::Delta, + None, + &now_nanos(), + &[&m1], + ); + assert_eq!(only_data_point(&r1).as_double, 10.0); + + let m2 = counter("total_queries", vec![], 25); + let r2 = build_request_with_state( + &state, + AggregationTemporality::Delta, + None, + &now_nanos(), + &[&m2], + ); + assert_eq!(only_data_point(&r2).as_double, 15.0); + } + + #[test] + fn counter_reset_skips_data_point() { + let state = CounterState::default(); + + let m1 = counter("total_queries", vec![], 10); + let _ = build_request_with_state( + &state, + AggregationTemporality::Delta, + None, + &now_nanos(), + &[&m1], + ); + + let m2 = counter("total_queries", vec![], 3); + let r2 = build_request_with_state( + &state, + AggregationTemporality::Delta, + None, + &now_nanos(), + &[&m2], + ); + + let sum = r2.resource_metrics[0].scope_metrics[0].metrics[0] + .sum + .as_ref() + .expect("sum"); + assert!( + sum.data_points.is_empty(), + "reset counter should skip the data point, got {:?}", + sum.data_points + .iter() + .map(|d| d.as_double) + .collect::>() + ); + } + + #[test] + fn counter_deltas_are_tracked_per_label_set() { + let state = CounterState::default(); + + let build = |alice_val: i64, bob_val: i64| { + Metric::new(PoolMetric { + name: "total_queries".into(), + measurements: vec![ + Measurement { + labels: vec![("user".into(), "alice".into())], + measurement: MeasurementType::Integer(alice_val), + }, + Measurement { + labels: vec![("user".into(), "bob".into())], + measurement: MeasurementType::Integer(bob_val), + }, + ], + help: "".into(), + unit: None, + metric_type: Some(OpenMetricType::Counter), + }) + }; + + let m1 = build(10, 100); + let _ = build_request_with_state( + &state, + AggregationTemporality::Delta, + None, + &now_nanos(), + &[&m1], + ); + + // alice advances by 5, bob stays put. + let m2 = build(15, 100); + let r2 = build_request_with_state( + &state, + AggregationTemporality::Delta, + None, + &now_nanos(), + &[&m2], + ); + + let points = &r2.resource_metrics[0].scope_metrics[0].metrics[0] + .sum + .as_ref() + .expect("sum") + .data_points; + + let find_user = |user: &str| { + points + .iter() + .find(|dp| { + dp.attributes + .iter() + .any(|a| a.key == "user" && a.value.string_value == user) + }) + .unwrap_or_else(|| panic!("data point for user={user}")) + }; + + assert_eq!(find_user("alice").as_double, 5.0); + assert_eq!(find_user("bob").as_double, 0.0); + } + + #[test] + fn counter_start_time_unix_nano_is_pinned_to_first_observation() { + let state = CounterState::default(); + + let m1 = counter("total_queries", vec![], 1); + let r1 = build_request_with_state( + &state, + AggregationTemporality::Cumulative, + None, + &now_nanos(), + &[&m1], + ); + let dp1 = only_data_point(&r1); + let first_start = dp1 + .start_time_unix_nano + .clone() + .expect("start_time_unix_nano set on counter"); + assert_eq!( + first_start, dp1.time_unix_nano, + "on first observation, start_time should equal time" + ); + + // Force `now_nanos()` to advance so we can distinguish "start reused" + // from "start == current now by coincidence". + std::thread::sleep(std::time::Duration::from_millis(2)); + + let m2 = counter("total_queries", vec![], 2); + let r2 = build_request_with_state( + &state, + AggregationTemporality::Cumulative, + None, + &now_nanos(), + &[&m2], + ); + let dp2 = only_data_point(&r2); + let second_start = dp2 + .start_time_unix_nano + .clone() + .expect("start_time_unix_nano set on counter"); + + assert_eq!( + second_start, first_start, + "start_time_unix_nano must be pinned to the first observation" + ); + assert_ne!( + dp2.time_unix_nano, second_start, + "time_unix_nano should advance while start_time_unix_nano stays put" + ); + } } diff --git a/pgdog/src/stats/otel_exporter.rs b/pgdog/src/stats/otel_exporter.rs index b78d8707d..c8b47882c 100644 --- a/pgdog/src/stats/otel_exporter.rs +++ b/pgdog/src/stats/otel_exporter.rs @@ -4,7 +4,6 @@ //! to the configured endpoint (e.g. Datadog's `/api/v2/otlp/v1/metrics`). use std::time::Duration; - use tracing::{info, warn}; use super::otel; @@ -103,7 +102,6 @@ pub async fn run() { #[cfg(test)] mod test { - use crate::config::{self, ConfigAndUsers}; use crate::stats::Metric; use crate::stats::open_metric::{Measurement, MeasurementType}; use crate::stats::otel; @@ -113,8 +111,6 @@ mod test { fn serialized_payload_is_valid_json() { let _test_lock = crate::stats::otel::TEST_LOCK.lock(); - config::set(ConfigAndUsers::default()).unwrap(); - let metric = Metric::new(PoolMetric { name: "sv_idle".into(), measurements: vec![Measurement { diff --git a/pgdog/src/stats/pools.rs b/pgdog/src/stats/pools.rs index 8f8dbee44..0fce79ab9 100644 --- a/pgdog/src/stats/pools.rs +++ b/pgdog/src/stats/pools.rs @@ -1,14 +1,18 @@ +use std::sync::Arc; + +use pgdog_stats::histogram; + use crate::backend::{self, databases::databases}; use crate::util::millis; -use super::{Measurement, Metric, OpenMetric}; +use super::{HistogramMeasurement, Measurement, Metric, OpenMetric, OpenMetricType}; pub struct PoolMetric { pub name: String, pub measurements: Vec, pub help: String, pub unit: Option, - pub metric_type: Option, + pub metric_type: Option, } impl OpenMetric for PoolMetric { @@ -28,12 +32,8 @@ impl OpenMetric for PoolMetric { self.unit.clone() } - fn metric_type(&self) -> String { - if let Some(ref metric_type) = self.metric_type { - metric_type.clone() - } else { - "gauge".into() - } + fn metric_type(&self) -> OpenMetricType { + self.metric_type.unwrap_or(OpenMetricType::Gauge) } } @@ -93,6 +93,12 @@ impl Pools { let mut avg_rows_updated = vec![]; let mut total_rows_deleted = vec![]; let mut avg_rows_deleted = vec![]; + let mut query_time_histogram = vec![]; + + let histogram_bounds = histogram::bounds(); + // Bounds are process-constant, so convert to seconds once per scrape + // and share the allocation across every pool's measurement. + let histogram_bounds_seconds: Arc<[f64]> = histogram_bounds.seconds().into(); let general = &crate::config::config().config.general; @@ -357,6 +363,18 @@ impl Pools { labels: labels.clone(), measurement: averages.rows_deleted.into(), }); + + let histogram = stats.query_time_histogram; + query_time_histogram.push(Measurement { + labels: labels.clone(), + measurement: HistogramMeasurement::new( + histogram_bounds_seconds.clone(), + histogram.buckets(histogram_bounds), + histogram.sum().as_secs_f64(), + histogram.count(), + ) + .into(), + }); } } } @@ -438,7 +456,7 @@ impl Pools { measurements: errors, help: "Errors connections in the pool have experienced.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -446,7 +464,7 @@ impl Pools { measurements: out_of_sync, help: "Connections that have been returned to the pool in a broken state.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -454,7 +472,7 @@ impl Pools { measurements: total_xact_count, help: "Total number of executed transactions.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -462,7 +480,7 @@ impl Pools { measurements: total_xact_2pc_count, help: "Total number of executed two-phase commit transactions.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -487,7 +505,7 @@ impl Pools { measurements: total_query_count, help: "Total number of executed queries.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -503,7 +521,7 @@ impl Pools { measurements: total_received, help: "Total number of bytes received.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -511,7 +529,7 @@ impl Pools { measurements: avg_received, help: "Average number of bytes received.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -519,7 +537,7 @@ impl Pools { measurements: total_sent, help: "Total number of bytes sent.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -535,7 +553,7 @@ impl Pools { measurements: total_xact_time, help: "Total time spent executing transactions.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -551,7 +569,7 @@ impl Pools { measurements: total_idle_xact_time, help: "Total time spent idling inside transactions.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -567,7 +585,7 @@ impl Pools { measurements: total_query_time, help: "Total time spent executing queries.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -583,7 +601,7 @@ impl Pools { measurements: total_close, help: "Total number of prepared statements closed because of cache evictions.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -599,7 +617,7 @@ impl Pools { measurements: total_server_errors, help: "Total number of errors returned by server connections.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -616,7 +634,7 @@ impl Pools { help: "Total number of times server connections were cleaned from client parameters." .into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -635,7 +653,7 @@ impl Pools { "Total number of abandoned transactions that had to be rolled back automatically." .into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -653,7 +671,7 @@ impl Pools { measurements: total_connect_time, help: "Total time spent connecting to servers.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -669,7 +687,7 @@ impl Pools { measurements: total_connect_count, help: "Total number of connections established to servers.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -685,7 +703,7 @@ impl Pools { measurements: total_reads, help: "Total number of read transactions.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -701,7 +719,7 @@ impl Pools { measurements: total_writes, help: "Total number of write transactions.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -725,7 +743,7 @@ impl Pools { measurements: total_auth_attempts, help: "Total number of server authentication attempts.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -741,7 +759,7 @@ impl Pools { measurements: total_rows_inserted, help: "Total rows reported affected by INSERT command tags.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -757,7 +775,7 @@ impl Pools { measurements: total_rows_updated, help: "Total rows reported affected by UPDATE command tags.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -773,7 +791,7 @@ impl Pools { measurements: total_rows_deleted, help: "Total rows reported affected by DELETE command tags.".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), })); metrics.push(Metric::new(PoolMetric { @@ -784,6 +802,14 @@ impl Pools { metric_type: None, })); + metrics.push(Metric::new(PoolMetric { + name: "query_time_seconds".into(), + measurements: query_time_histogram, + help: "Distribution of query execution times.".into(), + unit: Some("seconds".into()), + metric_type: Some(OpenMetricType::Histogram), + })); + Pools { metrics } } @@ -811,7 +837,7 @@ mod tests { metric_type: None, }; - assert_eq!(metric.metric_type(), "gauge"); + assert_eq!(metric.metric_type(), OpenMetricType::Gauge); assert!(metric.unit().is_none()); assert_eq!(metric.help(), Some("Waiting clients per pool".into())); } @@ -831,7 +857,7 @@ mod tests { }], help: "Active servers per pool".into(), unit: Some("connections".into()), - metric_type: Some("gauge".into()), + metric_type: Some(OpenMetricType::Gauge), }; let rendered = Metric::new(metric).to_string(); @@ -871,7 +897,7 @@ mod test { }], help: "How long clients wait.".into(), unit: Some("seconds".into()), - metric_type: Some("counter".into()), // Not correct, just testing display. + metric_type: Some(OpenMetricType::Counter), // Not correct, just testing display. })], }; let rendered = pools.to_string(); diff --git a/pgdog/src/stats/query_cache.rs b/pgdog/src/stats/query_cache.rs index 472b60939..02c4a8056 100644 --- a/pgdog/src/stats/query_cache.rs +++ b/pgdog/src/stats/query_cache.rs @@ -105,11 +105,11 @@ impl OpenMetric for QueryCacheMetric { self.name.clone() } - fn metric_type(&self) -> String { + fn metric_type(&self) -> OpenMetricType { if self.gauge { - "gauge".into() + OpenMetricType::Gauge } else { - "counter".into() + OpenMetricType::Counter } } @@ -209,12 +209,12 @@ mod tests { .iter() .find(|m| m.name() == "query_cache_fingerprints") .unwrap(); - assert_eq!(fingerprints_metric.metric_type(), "counter"); + assert_eq!(fingerprints_metric.metric_type(), OpenMetricType::Counter); let rendered = fingerprints_metric.to_string(); assert!(rendered.contains("query_cache_fingerprints 8")); let memory_metric = metrics.last().unwrap(); - assert_eq!(memory_metric.metric_type(), "gauge"); + assert_eq!(memory_metric.metric_type(), OpenMetricType::Gauge); let rendered = memory_metric.to_string(); assert!(rendered.contains("prepared_statements_memory_used 7")); } diff --git a/pgdog/src/stats/two_pc.rs b/pgdog/src/stats/two_pc.rs index 0253c1dd3..d773eb79d 100644 --- a/pgdog/src/stats/two_pc.rs +++ b/pgdog/src/stats/two_pc.rs @@ -2,7 +2,7 @@ use crate::frontend::client::query_engine::two_pc::Manager; -use super::{Measurement, Metric, OpenMetric}; +use super::{Measurement, Metric, OpenMetric, OpenMetricType}; pub struct TwoPc { recovered_total: u64, @@ -22,8 +22,8 @@ impl OpenMetric for TwoPc { "two_pc_recovered_total".into() } - fn metric_type(&self) -> String { - "counter".into() + fn metric_type(&self) -> OpenMetricType { + OpenMetricType::Counter } fn help(&self) -> Option {