From 18a4d8425b4830fd18f49d66180d9784e728ceed Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 18:08:36 -0400 Subject: [PATCH 01/11] use serde_repr to strictly type SumAggregationTemporality --- Cargo.lock | 23 +++++++++++++++++++++++ pgdog/Cargo.toml | 1 + pgdog/src/stats/otel.rs | 13 ++++++++++--- 3 files changed, 34 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8402aecd3..bb0479a1a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3397,6 +3397,7 @@ dependencies = [ "semver", "serde", "serde_json", + "serde_repr", "sha1", "smallvec", "socket2 0.5.10", @@ -4606,6 +4607,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" @@ -5114,6 +5126,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/pgdog/Cargo.toml b/pgdog/Cargo.toml index 5a23d3c9e..7cc59502e 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/stats/otel.rs b/pgdog/src/stats/otel.rs index 407c39a67..ce2d7dcd2 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -89,11 +89,18 @@ pub struct Gauge { pub data_points: Vec, } +// little serde trick to let us serialize directly as the integer representation +#[derive(serde_repr::Serialize_repr)] +#[repr(u8)] +pub enum SumAggregationTemporality { + Delta = 1, + Cumulative = 2, +} + #[derive(Serialize)] #[serde(rename_all = "camelCase")] pub struct Sum { - /// 1 = DELTA, 2 = CUMULATIVE - pub aggregation_temporality: u32, + pub aggregation_temporality: SumAggregationTemporality, pub is_monotonic: bool, pub data_points: Vec, } @@ -281,7 +288,7 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ ( None, Some(Sum { - aggregation_temporality: 1, // DELTA + aggregation_temporality: SumAggregationTemporality::Delta, is_monotonic: true, data_points, }), From 916354feee9d6ca429dd8b7015345aecbe0b2c0d Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 19:16:36 -0400 Subject: [PATCH 02/11] case-insensitive parse out OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE in both [otel] section and env var form --- pgdog-config/src/lib.rs | 1 + pgdog-config/src/otel.rs | 24 +++++++++++++++--- pgdog-config/src/otel_temporality.rs | 38 ++++++++++++++++++++++++++++ 3 files changed, 60 insertions(+), 3 deletions(-) create mode 100644 pgdog-config/src/otel_temporality.rs diff --git a/pgdog-config/src/lib.rs b/pgdog-config/src/lib.rs index 142713fec..425ccaf35 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..747c9d14e 100644 --- a/pgdog-config/src/otel.rs +++ b/pgdog-config/src/otel.rs @@ -1,8 +1,8 @@ -use std::collections::HashMap; -use std::env; - +use crate::otel_temporality::OtelTemporalityPreference; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::env; /// OpenTelemetry push exporter settings. /// @@ -61,6 +61,16 @@ pub struct Otel { /// Env: `OTEL_METRIC_EXPORT_INTERVAL` #[serde(default = "Otel::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` + /// + /// Env: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` + #[serde(default = "Otel::temporality_preference")] + pub temporality_preference: OtelTemporalityPreference, } impl Otel { @@ -99,6 +109,14 @@ impl Otel { .and_then(|v| v.parse().ok()) .unwrap_or(10_000) } + + fn temporality_preference() -> OtelTemporalityPreference { + env::var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE") + .ok() + .and_then(|v| v.parse().ok()) + // defaults to cumulative + .unwrap_or_default() + } } #[cfg(test)] diff --git a/pgdog-config/src/otel_temporality.rs b/pgdog-config/src/otel_temporality.rs new file mode 100644 index 000000000..509196286 --- /dev/null +++ b/pgdog-config/src/otel_temporality.rs @@ -0,0 +1,38 @@ +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"]) + }) + } +} From 3e2212a8f039b0e40e48394e422a24d9f168978c Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 19:35:35 -0400 Subject: [PATCH 03/11] return data points according to temporality --- pgdog/src/stats/otel.rs | 48 +++++++++++++++++++++++++++++------------ 1 file changed, 34 insertions(+), 14 deletions(-) diff --git a/pgdog/src/stats/otel.rs b/pgdog/src/stats/otel.rs index ce2d7dcd2..568442da4 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -9,6 +9,7 @@ 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; @@ -90,7 +91,7 @@ pub struct Gauge { } // little serde trick to let us serialize directly as the integer representation -#[derive(serde_repr::Serialize_repr)] +#[derive(serde_repr::Serialize_repr, Clone, Copy)] #[repr(u8)] pub enum SumAggregationTemporality { Delta = 1, @@ -227,6 +228,13 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ let common_attrs = &*RESOURCE_ATTRIBUTES; + let aggregation_temporality = match config.config.otel.temporality_preference { + OtelTemporalityPreference::Cumulative => SumAggregationTemporality::Cumulative, + OtelTemporalityPreference::Delta | OtelTemporalityPreference::LowMemory => { + SumAggregationTemporality::Delta + } + }; + let otel_metrics: Vec = metrics .iter() .map(|metric| { @@ -240,19 +248,31 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ 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; + // todo: This is pretty nested, we should probably look + // at refactoring how we calculate these values to flatten + // the logic a bit, counters and sums should probably not + // use the same data point code + + // NOTE: if aggregation_temporality changes state during program + // execution, the data may be stale, but this should be impossible + match aggregation_temporality { + SumAggregationTemporality::Cumulative => cumulative, + SumAggregationTemporality::Delta => { + 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 + } } - delta } else { cumulative }; @@ -288,7 +308,7 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ ( None, Some(Sum { - aggregation_temporality: SumAggregationTemporality::Delta, + aggregation_temporality, is_monotonic: true, data_points, }), From e1ef30a939419229bbb6a3ae7c44354be670ac5f Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 19:46:25 -0400 Subject: [PATCH 04/11] return start_time_unix_nano with otel counters and sum --- pgdog/src/stats/otel.rs | 34 ++++++++++++++++++++++++---------- 1 file changed, 24 insertions(+), 10 deletions(-) diff --git a/pgdog/src/stats/otel.rs b/pgdog/src/stats/otel.rs index 568442da4..7e9a0a6d6 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -32,6 +32,11 @@ struct CounterKey { static PREV_COUNTERS: Lazy>> = Lazy::new(|| Mutex::new(HashMap::new())); +/// First-seen timestamp per counter data point, used as `start_time_unix_nano` +/// so cumulative counters carry a stable collection-start reference. +static COUNTER_START_TIMES: Lazy>> = + Lazy::new(|| Mutex::new(HashMap::new())); + pub fn now_nanos() -> String { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -247,21 +252,28 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ .filter_map(|m| { let cumulative = measurement_to_f64(&m.measurement); - let as_double = if is_counter { + let (as_double, start_time_unix_nano) = if is_counter { // todo: This is pretty nested, we should probably look // at refactoring how we calculate these values to flatten // the logic a bit, counters and sums should probably not // use the same data point code + let key = CounterKey { + metric: name.clone(), + labels: m.labels.clone(), + }; + + let start = COUNTER_START_TIMES + .lock() + .entry(key.clone()) + .or_insert_with(|| now.to_owned()) + .clone(); + // NOTE: if aggregation_temporality changes state during program - // execution, the data may be stale, but this should be impossible - match aggregation_temporality { + // execution, the data may be stale, but this is currently impossible + let value = match aggregation_temporality { SumAggregationTemporality::Cumulative => cumulative, SumAggregationTemporality::Delta => { - 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); @@ -272,9 +284,11 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ } delta } - } + }; + + (value, Some(start)) } else { - cumulative + (cumulative, None) }; let mut attributes: Vec = m @@ -296,7 +310,7 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ })); Some(NumberDataPoint { - start_time_unix_nano: None, + start_time_unix_nano, time_unix_nano: now.to_owned(), as_double, attributes, From 93716b20b5488b68d4f9836cc0dba82cb27276b4 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 20:01:10 -0400 Subject: [PATCH 05/11] use explicit OpenMetricType enum instead of magic strings --- pgdog/src/stats/listeners.rs | 20 ++++++------ pgdog/src/stats/mirror_stats.rs | 34 +++++++++---------- pgdog/src/stats/open_metric.rs | 23 +++++++++++-- pgdog/src/stats/otel.rs | 8 ++--- pgdog/src/stats/pools.rs | 58 +++++++++++++++------------------ pgdog/src/stats/query_cache.rs | 10 +++--- pgdog/src/stats/two_pc.rs | 6 ++-- 7 files changed, 87 insertions(+), 72 deletions(-) diff --git a/pgdog/src/stats/listeners.rs b/pgdog/src/stats/listeners.rs index 5bc619e54..ebcbc5d83 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}; @@ -35,19 +35,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, }), ] } @@ -57,7 +57,7 @@ struct ListenerMetric { name: String, measurements: Vec, help: String, - metric_type: String, + metric_type: OpenMetricType, } impl OpenMetric for ListenerMetric { @@ -73,8 +73,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 } } @@ -95,8 +95,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/mirror_stats.rs b/pgdog/src/stats/mirror_stats.rs index d67a34ca5..5234e0881 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..d21dd31d1 100644 --- a/pgdog/src/stats/open_metric.rs +++ b/pgdog/src/stats/open_metric.rs @@ -6,21 +6,40 @@ 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, +} + +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", + }; + f.write_str(s) + } +} + #[derive(Debug, Clone)] pub enum MeasurementType { Float(f64), diff --git a/pgdog/src/stats/otel.rs b/pgdog/src/stats/otel.rs index 7e9a0a6d6..48ff879d0 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -14,7 +14,7 @@ use serde::Serialize; use crate::util::hostname; -use super::open_metric::{MeasurementType, Metric}; +use super::open_metric::{MeasurementType, Metric, OpenMetricType}; static RESOURCE_ATTRIBUTES: Lazy> = Lazy::new(resource_attributes); @@ -244,7 +244,7 @@ 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 is_counter = matches!(metric.metric_type(), OpenMetricType::Counter); let data_points: Vec = metric .measurements() @@ -402,7 +402,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()); @@ -495,7 +495,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()); diff --git a/pgdog/src/stats/pools.rs b/pgdog/src/stats/pools.rs index b6755bd8b..67de8d533 100644 --- a/pgdog/src/stats/pools.rs +++ b/pgdog/src/stats/pools.rs @@ -1,14 +1,14 @@ use crate::backend::{self, databases::databases}; use crate::util::millis; -use super::{Measurement, Metric, OpenMetric}; +use super::{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 +28,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) } } @@ -402,7 +398,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 { @@ -410,7 +406,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 { @@ -418,7 +414,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 { @@ -426,7 +422,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 { @@ -451,7 +447,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 { @@ -467,7 +463,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 { @@ -475,7 +471,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 { @@ -483,7 +479,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 { @@ -499,7 +495,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 { @@ -515,7 +511,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 { @@ -531,7 +527,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 { @@ -547,7 +543,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 { @@ -563,7 +559,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 { @@ -580,7 +576,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 { @@ -599,7 +595,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 { @@ -617,7 +613,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 { @@ -633,7 +629,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 { @@ -649,7 +645,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 { @@ -665,7 +661,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 { @@ -689,7 +685,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 { @@ -727,7 +723,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())); } @@ -747,7 +743,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(); @@ -787,7 +783,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 { From 131a088d48ab56d6500cfcd132a5cc860e5d42ec Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 20:28:36 -0400 Subject: [PATCH 06/11] flatten and extract helpers to make relationship between OpenMetricType and AggregationTemporality easier to read --- pgdog/src/stats/otel.rs | 423 +++++++++++++++++++++++++++++++--------- 1 file changed, 328 insertions(+), 95 deletions(-) diff --git a/pgdog/src/stats/otel.rs b/pgdog/src/stats/otel.rs index 48ff879d0..e58f91f39 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -14,7 +14,7 @@ use serde::Serialize; use crate::util::hostname; -use super::open_metric::{MeasurementType, Metric, OpenMetricType}; +use super::open_metric::{Measurement, MeasurementType, Metric, OpenMetricType}; static RESOURCE_ATTRIBUTES: Lazy> = Lazy::new(resource_attributes); @@ -28,14 +28,36 @@ 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). +#[derive(Default)] +struct CounterState { + prev_values: 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) + } +} -/// First-seen timestamp per counter data point, used as `start_time_unix_nano` -/// so cumulative counters carry a stable collection-start reference. -static COUNTER_START_TIMES: Lazy>> = - Lazy::new(|| Mutex::new(HashMap::new())); +static COUNTER_STATE: Lazy = Lazy::new(CounterState::default); pub fn now_nanos() -> String { SystemTime::now() @@ -215,16 +237,101 @@ fn measurement_to_f64(m: &MeasurementType) -> f64 { } } -/// Build an `ExportMetricsServiceRequest` from a collection of `Metric` objects. +/// 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: SumAggregationTemporality, + 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 { + SumAggregationTemporality::Cumulative => cumulative, + SumAggregationTemporality::Delta => state.delta(&key, cumulative)?, + }; + Some((value, Some(start))) + } + } +} + +/// 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: SumAggregationTemporality, + 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, + }), + ), + } +} + +/// 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 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 temporality = match config.config.otel.temporality_preference { + OtelTemporalityPreference::Cumulative => SumAggregationTemporality::Cumulative, + OtelTemporalityPreference::Delta | OtelTemporalityPreference::LowMemory => { + SumAggregationTemporality::Delta + } + }; + let namespace = config.config.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: SumAggregationTemporality, + 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 { @@ -233,103 +340,29 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ let common_attrs = &*RESOURCE_ATTRIBUTES; - let aggregation_temporality = match config.config.otel.temporality_preference { - OtelTemporalityPreference::Cumulative => SumAggregationTemporality::Cumulative, - OtelTemporalityPreference::Delta | OtelTemporalityPreference::LowMemory => { - SumAggregationTemporality::Delta - } - }; - let otel_metrics: Vec = metrics .iter() .map(|metric| { let name = format!("{}.{}", namespace, metric.name()); - let is_counter = matches!(metric.metric_type(), OpenMetricType::Counter); + let metric_type = metric.metric_type(); let data_points: Vec = metric .measurements() .iter() .filter_map(|m| { - let cumulative = measurement_to_f64(&m.measurement); - - let (as_double, start_time_unix_nano) = if is_counter { - // todo: This is pretty nested, we should probably look - // at refactoring how we calculate these values to flatten - // the logic a bit, counters and sums should probably not - // use the same data point code - - let key = CounterKey { - metric: name.clone(), - labels: m.labels.clone(), - }; - - let start = COUNTER_START_TIMES - .lock() - .entry(key.clone()) - .or_insert_with(|| now.to_owned()) - .clone(); - - // NOTE: if aggregation_temporality changes state during program - // execution, the data may be stale, but this is currently impossible - let value = match aggregation_temporality { - SumAggregationTemporality::Cumulative => cumulative, - SumAggregationTemporality::Delta => { - 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 - } - }; - - (value, Some(start)) - } else { - (cumulative, None) - }; - - let mut attributes: Vec = m - .labels - .iter() - .map(|(k, v)| KeyValue { - key: k.clone(), - value: AttributeValue { - string_value: v.clone(), - }, - }) - .collect(); - - attributes.extend(common_attrs.iter().map(|a| KeyValue { - key: a.key.clone(), - value: AttributeValue { - string_value: a.value.string_value.clone(), - }, - })); + 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, + attributes: build_attributes(&m.labels, common_attrs), }) }) .collect(); - let (gauge, sum) = if is_counter { - ( - None, - Some(Sum { - aggregation_temporality, - is_monotonic: true, - data_points, - }), - ) - } else { - (Some(Gauge { data_points }), None) - }; + let (gauge, sum) = wrap_data_points(metric_type, temporality, data_points); OtelMetric { name, @@ -394,6 +427,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 = OtelTemporalityPreference::Delta; + config::set(cfg).expect("set config"); + let metric = Metric::new(PoolMetric { name: "total_query_count".into(), measurements: vec![Measurement { @@ -610,4 +648,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, + SumAggregationTemporality::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, + SumAggregationTemporality::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, + SumAggregationTemporality::Delta, + None, + &now_nanos(), + &[&m1], + ); + + let m2 = counter("total_queries", vec![], 3); + let r2 = build_request_with_state( + &state, + SumAggregationTemporality::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, + SumAggregationTemporality::Delta, + None, + &now_nanos(), + &[&m1], + ); + + // alice advances by 5, bob stays put. + let m2 = build(15, 100); + let r2 = build_request_with_state( + &state, + SumAggregationTemporality::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, + SumAggregationTemporality::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, + SumAggregationTemporality::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" + ); + } } From 2d0d1a7518bc6118f090fc031a914f04035ac4bb Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 21:01:15 -0400 Subject: [PATCH 07/11] Update JSON schema --- .schema/pgdog.schema.json | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/.schema/pgdog.schema.json b/.schema/pgdog.schema.json index 745cb39d2..ca4c87a41 100644 --- a/.schema/pgdog.schema.json +++ b/.schema/pgdog.schema.json @@ -172,7 +172,8 @@ "endpoint": null, "headers": {}, "namespace": null, - "push_interval": 0 + "push_interval": 0, + "temporality_preference": "Cumulative" } }, "plugins": { @@ -1507,10 +1508,35 @@ "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`\n\nEnv: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE`", + "$ref": "#/$defs/OtelTemporalityPreference", + "default": "Cumulative" } }, "additionalProperties": false }, + "OtelTemporalityPreference": { + "description": "Aggregation temporality used when exporting OTLP metric points.\n\n", + "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": [ From bf698cdcbdc348e4e023c99af788c063a89675b6 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 21:23:26 -0400 Subject: [PATCH 08/11] test OtelTemporalityPreference to make codecov happy --- pgdog-config/src/otel_temporality.rs | 73 +++++++++++++++++++++++++++- 1 file changed, 71 insertions(+), 2 deletions(-) diff --git a/pgdog-config/src/otel_temporality.rs b/pgdog-config/src/otel_temporality.rs index 509196286..c2533e5dd 100644 --- a/pgdog-config/src/otel_temporality.rs +++ b/pgdog-config/src/otel_temporality.rs @@ -2,8 +2,6 @@ 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( @@ -36,3 +34,74 @@ impl<'de> Deserialize<'de> for OtelTemporalityPreference { }) } } + +#[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}"); + } +} From 9d6649a7757ef982011422b9b50c01a57523f3fb Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 21:25:54 -0400 Subject: [PATCH 09/11] regen json schema --- .schema/pgdog.schema.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.schema/pgdog.schema.json b/.schema/pgdog.schema.json index ca4c87a41..72f0167cf 100644 --- a/.schema/pgdog.schema.json +++ b/.schema/pgdog.schema.json @@ -1518,7 +1518,7 @@ "additionalProperties": false }, "OtelTemporalityPreference": { - "description": "Aggregation temporality used when exporting OTLP metric points.\n\n", + "description": "Aggregation temporality used when exporting OTLP metric points.", "oneOf": [ { "description": "Points report the value accumulated since the exporter started.", From cab17505b9ee40fc31066a739330e23fbf2b77ac Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Sat, 1 Aug 2026 04:44:48 -0400 Subject: [PATCH 10/11] seperate out OtelRawToml and explicitly resolve with environment variables fixes issue with missing [otel] section dropping environment variables --- .schema/pgdog.schema.json | 4 +- pgdog-config/src/otel.rs | 234 ++++++++++++++++++++++++++++++++++---- 2 files changed, 212 insertions(+), 26 deletions(-) diff --git a/.schema/pgdog.schema.json b/.schema/pgdog.schema.json index 72f0167cf..196856502 100644 --- a/.schema/pgdog.schema.json +++ b/.schema/pgdog.schema.json @@ -172,7 +172,7 @@ "endpoint": null, "headers": {}, "namespace": null, - "push_interval": 0, + "push_interval": 10000, "temporality_preference": "Cumulative" } }, @@ -1510,7 +1510,7 @@ "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`\n\nEnv: `OTEL_EXPORTER_OTLP_METRICS_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`", "$ref": "#/$defs/OtelTemporalityPreference", "default": "Cumulative" } diff --git a/pgdog-config/src/otel.rs b/pgdog-config/src/otel.rs index 747c9d14e..3ef4267ee 100644 --- a/pgdog-config/src/otel.rs +++ b/pgdog-config/src/otel.rs @@ -3,6 +3,7 @@ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::env; +use tracing::warn; /// OpenTelemetry push exporter settings. /// @@ -10,7 +11,7 @@ use std::env; /// to the configured URL. /// /// -#[derive(JsonSchema, Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[derive(JsonSchema, Serialize, Debug, Clone, PartialEq)] #[serde(deny_unknown_fields)] pub struct Otel { /// Full URL of the OTLP metrics ingest endpoint @@ -18,7 +19,7 @@ pub struct Otel { /// When not set, the push exporter is disabled. /// /// Env: `OTEL_EXPORTER_OTLP_ENDPOINT` - #[serde(default = "Otel::endpoint")] + #[schemars(default)] pub endpoint: Option, /// Prefix added to all metric names emitted by the OTEL exporter. @@ -31,7 +32,7 @@ pub struct Otel { /// Env: `PGDOG_OTEL_NAMESPACE` /// /// - #[serde(default = "Otel::namespace")] + #[schemars(default)] pub namespace: Option, /// HTTP headers sent with each OTLP push request. @@ -44,14 +45,14 @@ pub struct Otel { /// ``` /// /// Env: `OTEL_EXPORTER_OTLP_HEADERS` (comma-separated `key=value` pairs) - #[serde(default = "Otel::headers")] + #[schemars(default)] pub headers: HashMap, /// Datadog API key. Convenience shorthand that adds a `DD-API-KEY` header /// to OTLP push requests. /// /// Env: `DD_API_KEY` - #[serde(default = "Otel::datadog_api_key")] + #[schemars(default)] pub datadog_api_key: Option, /// How often, in milliseconds, to push metrics to the OTLP endpoint. @@ -59,63 +60,172 @@ pub struct Otel { /// _Default:_ `10000` /// /// Env: `OTEL_METRIC_EXPORT_INTERVAL` - #[serde(default = "Otel::push_interval")] + #[schemars(default = "Otel::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` + /// _Default:_ `Cumulative`, or `Delta` when `datadog_api_key` is set. /// /// Env: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` - #[serde(default = "Otel::temporality_preference")] + #[schemars(default)] pub temporality_preference: OtelTemporalityPreference, } +impl Default for Otel { + fn default() -> Self { + OtelRawToml::default().resolve_with_env() + } +} + impl Otel { fn env_option_string(env_var: &str) -> Option { env::var(env_var).ok().filter(|s| !s.is_empty()) } - fn endpoint() -> Option { + fn default_push_interval() -> u64 { + 10_000 + } + + fn endpoint_from_env() -> Option { Self::env_option_string("OTEL_EXPORTER_OTLP_ENDPOINT") } - fn namespace() -> Option { + fn namespace_from_env() -> Option { Self::env_option_string("PGDOG_OTEL_NAMESPACE") } - fn headers() -> HashMap { + fn headers_from_env() -> Option> { + let raw = Self::env_option_string("OTEL_EXPORTER_OTLP_HEADERS")?; let mut map = HashMap::new(); - if let Some(raw) = Self::env_option_string("OTEL_EXPORTER_OTLP_HEADERS") { - for pair in raw.split(',') { - let pair = pair.trim(); - if let Some((k, v)) = pair.split_once('=') { - map.insert(k.trim().to_owned(), v.trim().to_owned()); - } + for pair in raw.split(',') { + let pair = pair.trim(); + if let Some((k, v)) = pair.split_once('=') { + map.insert(k.trim().to_owned(), v.trim().to_owned()); } } - map + Some(map) } - fn datadog_api_key() -> Option { + fn datadog_api_key_from_env() -> Option { Self::env_option_string("DD_API_KEY") } - fn push_interval() -> u64 { + fn push_interval_from_env() -> Option { env::var("OTEL_METRIC_EXPORT_INTERVAL") .ok() .and_then(|v| v.parse().ok()) - .unwrap_or(10_000) } - fn temporality_preference() -> OtelTemporalityPreference { + fn temporality_preference_from_env() -> Option { env::var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE") .ok() .and_then(|v| v.parse().ok()) - // defaults to cumulative - .unwrap_or_default() + } +} + +fn warn_if_disagree_prefer_env( + field: &str, + toml_val: Option, + env_val: Option, +) -> Option { + if let (Some(t), Some(e)) = (toml_val.as_ref(), env_val.as_ref()) + && t != e + { + warn!( + "otel.{field} in TOML ({t:?}) disagrees with environment variable ({e:?}); using env value" + ); + } + + env_val.or(toml_val) +} + +#[derive(Deserialize, Default)] +#[serde(deny_unknown_fields)] +struct OtelRawToml { + endpoint: Option, + namespace: Option, + headers: Option>, + datadog_api_key: Option, + push_interval: Option, + temporality_preference: Option, +} + +impl OtelRawToml { + /// Resolve cross-field defaults and build an `Otel`. For each field, env wins over + /// TOML; if both are set and differ, warns. `temporality_preference` additionally + /// falls back to `Delta` when `datadog_api_key` is set. + fn resolve_with_env(self) -> Otel { + let endpoint = + warn_if_disagree_prefer_env("endpoint", self.endpoint, Otel::endpoint_from_env()); + let namespace = + warn_if_disagree_prefer_env("namespace", self.namespace, Otel::namespace_from_env()); + let headers = + warn_if_disagree_prefer_env("headers", self.headers, Otel::headers_from_env()) + .unwrap_or_default(); + + let datadog_api_key = warn_if_disagree_prefer_env( + "datadog_api_key", + self.datadog_api_key, + Otel::datadog_api_key_from_env(), + ); + + let push_interval = warn_if_disagree_prefer_env( + "push_interval", + self.push_interval, + Otel::push_interval_from_env(), + ) + .unwrap_or(Otel::default_push_interval()); + + let configured_temporality = warn_if_disagree_prefer_env( + "temporality_preference", + self.temporality_preference, + Otel::temporality_preference_from_env(), + ); + + let temporality_preference = match configured_temporality { + Some(explicitly_configured) => explicitly_configured, + None if datadog_api_key.is_some() => OtelTemporalityPreference::Delta, + None => OtelTemporalityPreference::default(), + }; + + if datadog_api_key.is_some() + && temporality_preference == OtelTemporalityPreference::Cumulative + && 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." + ); + } + + Otel { + endpoint, + namespace, + headers, + datadog_api_key, + push_interval, + temporality_preference, + } + } +} + +impl<'de> Deserialize<'de> for Otel { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + OtelRawToml::deserialize(deserializer).map(OtelRawToml::resolve_with_env) } } @@ -151,6 +261,82 @@ mod test { assert!(otel.endpoint.is_none()); assert!(otel.datadog_api_key.is_none()); assert_eq!(otel.push_interval, 10_000); + assert_eq!( + otel.temporality_preference, + OtelTemporalityPreference::Cumulative + ); + } + + #[test] + fn datadog_api_key_defaults_to_delta() { + let toml = r#"datadog_api_key = "abc""#; + let otel: Otel = toml::from_str(toml).expect("parse"); + assert_eq!( + otel.temporality_preference, + OtelTemporalityPreference::Delta + ); + } + + #[test] + fn no_datadog_api_key_defaults_to_cumulative() { + let otel: Otel = toml::from_str("").expect("parse"); + assert_eq!( + otel.temporality_preference, + OtelTemporalityPreference::Cumulative + ); + } + + #[test] + fn explicit_temporality_wins_over_dd_default() { + let toml = r#" + datadog_api_key = "abc" + temporality_preference = "Cumulative" + "#; + let otel: Otel = toml::from_str(toml).expect("parse"); + assert_eq!( + otel.temporality_preference, + OtelTemporalityPreference::Cumulative + ); + } + + #[test] + fn dd_api_key_from_env_triggers_delta_default() { + let _guard = set_env_var("DD_API_KEY", "abc"); + let otel: Otel = toml::from_str("").expect("parse"); + assert_eq!(otel.datadog_api_key.as_deref(), Some("abc")); + assert_eq!( + otel.temporality_preference, + OtelTemporalityPreference::Delta + ); + } + + #[test] + fn endpoint_env_wins_over_toml() { + 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://env.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 env_temporality_wins_over_dd_default() { + let _dd = set_env_var("DD_API_KEY", "abc"); + let _t = set_env_var( + "OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE", + "Cumulative", + ); + let otel: Otel = toml::from_str("").expect("parse"); + assert_eq!( + otel.temporality_preference, + OtelTemporalityPreference::Cumulative + ); } #[test] From 7a59ae5e688aee814de8cb5e55c0665b19ce20e8 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Sat, 1 Aug 2026 04:45:31 -0400 Subject: [PATCH 11/11] add pgdog-jsonschema line to CONTRIBUTING.md --- CONTRIBUTING.md | 1 + 1 file changed, 1 insertion(+) 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`