From 2d443716a22904047921255525042d49caf5e159 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 18:08:36 -0400 Subject: [PATCH 01/12] 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 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/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/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 3478261de2c755f11b426120d80ab621dd5e4d4a Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 19:16:36 -0400 Subject: [PATCH 02/12] 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 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..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 fed5e1428e7e6e021732f7d432c897d6814e817e Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 19:35:35 -0400 Subject: [PATCH 03/12] 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 9a46c79cb95dc4feef4399feab1e51581ccd6db5 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 19:46:25 -0400 Subject: [PATCH 04/12] 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 c838155c5cfd1e5c3068295ff7e7d7f39320e08e Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 20:01:10 -0400 Subject: [PATCH 05/12] use explicit OpenMetricType enum instead of magic strings --- pgdog/src/stats/listeners.rs | 20 ++++++------ pgdog/src/stats/lookup.rs | 34 +++++++++---------- 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 ++-- 8 files changed, 104 insertions(+), 89 deletions(-) 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..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 8f8dbee44..c31ddf6e5 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) } } @@ -438,7 +434,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 +442,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 +450,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 +458,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 +483,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 +499,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 +507,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 +515,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 +531,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 +547,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 +563,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 +579,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 +595,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 +612,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 +631,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 +649,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 +665,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 +681,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 +697,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 +721,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 { @@ -811,7 +807,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 +827,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 +867,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 91321342ba6dbb8710df81dd1a489e650cf5b675 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 20:28:36 -0400 Subject: [PATCH 06/12] 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 d63fad7e30042944dfce43d5019a65da7bfb49fc Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 21:01:15 -0400 Subject: [PATCH 07/12] 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 a334988f3..975d78bf6 100644 --- a/.schema/pgdog.schema.json +++ b/.schema/pgdog.schema.json @@ -175,7 +175,8 @@ "endpoint": null, "headers": {}, "namespace": null, - "push_interval": 0 + "push_interval": 0, + "temporality_preference": "Cumulative" } }, "plugins": { @@ -1598,10 +1599,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.", + "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 5d59018623a226e44a7cbbfbbe7deecb2a849958 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Thu, 30 Jul 2026 21:23:26 -0400 Subject: [PATCH 08/12] 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 1ced1c3632ba611a98a66d6c279e36299aa163a6 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Sat, 1 Aug 2026 04:44:48 -0400 Subject: [PATCH 09/12] otel: default temporality by datadog presence and centralize warning Make Otel::temporality_preference an Option; in ConfigAndUsers::load default it to Cumulative, or Delta when datadog_api_key is set. Move the Datadog-cumulative warning into ConfigAndUsers::check (with the match arm collapsed to a guard for readability). Add schema-only defaults so the generated JSON schema keeps the documented values instead of the derived Default (0 / null). --- .schema/pgdog.schema.json | 13 ++++-- pgdog-config/src/core.rs | 48 ++++++++++++++++++++++ pgdog-config/src/otel.rs | 68 ++++++++++++++++++++++++++++---- pgdog/src/stats/otel.rs | 37 +++++++++++++++-- pgdog/src/stats/otel_exporter.rs | 1 - 5 files changed, 152 insertions(+), 15 deletions(-) diff --git a/.schema/pgdog.schema.json b/.schema/pgdog.schema.json index 975d78bf6..34442e29a 100644 --- a/.schema/pgdog.schema.json +++ b/.schema/pgdog.schema.json @@ -175,7 +175,7 @@ "endpoint": null, "headers": {}, "namespace": null, - "push_interval": 0, + "push_interval": 10000, "temporality_preference": "Cumulative" } }, @@ -1601,8 +1601,15 @@ "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", + "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" } }, diff --git a/pgdog-config/src/core.rs b/pgdog-config/src/core.rs index 491cd9742..359284004 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::{ @@ -104,6 +105,24 @@ impl ConfigAndUsers { warn!("admin password has been randomly generated"); } + match ( + &mut config.otel.temporality_preference, + &config.otel.datadog_api_key, + ) { + // Here if temporality_preference isn't present, we set it based on + // if datadog is present + (default_cumulative_temporality @ None, None) => { + *default_cumulative_temporality = Some(OtelTemporalityPreference::Cumulative) + } + // datadog is present so we set it to delta + (delta_temporality_because_of_datadog @ None, Some(_datadog_api)) => { + *delta_temporality_because_of_datadog = Some(OtelTemporalityPreference::Delta) + } + (Some(_), _) => { + // We don't have to set temporality + } + } + let config_and_users = ConfigAndUsers { config, users, @@ -120,6 +139,7 @@ impl ConfigAndUsers { self.config.check(); self.users.check(&self.config); self.validate_server_auth()?; + self.warn_if_data_dog_cumulative(); Ok(()) } @@ -149,6 +169,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 +331,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/otel.rs b/pgdog-config/src/otel.rs index 747c9d14e..80fbde430 100644 --- a/pgdog-config/src/otel.rs +++ b/pgdog-config/src/otel.rs @@ -1,9 +1,11 @@ -use crate::otel_temporality::OtelTemporalityPreference; -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::env; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +use crate::otel_temporality::OtelTemporalityPreference; + /// OpenTelemetry push exporter settings. /// /// When `endpoint` is set, PgDog periodically POSTs OTLP JSON metrics @@ -60,17 +62,19 @@ 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` + /// _Default:_ `Cumulative`, or `Delta` when `datadog_api_key` is set. /// /// Env: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` #[serde(default = "Otel::temporality_preference")] - pub temporality_preference: OtelTemporalityPreference, + #[schemars(default = "Otel::schema_default_temporality_preference")] + pub temporality_preference: Option, } impl Otel { @@ -110,12 +114,29 @@ impl Otel { .unwrap_or(10_000) } - fn temporality_preference() -> OtelTemporalityPreference { + fn temporality_preference() -> Option { env::var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE") .ok() .and_then(|v| v.parse().ok()) - // defaults to cumulative - .unwrap_or_default() + } + + fn schema_default_push_interval() -> u64 { + 10_000 + } + + 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() + } } } @@ -151,6 +172,32 @@ mod test { assert!(otel.endpoint.is_none()); assert!(otel.datadog_api_key.is_none()); assert_eq!(otel.push_interval, 10_000); + 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] @@ -161,6 +208,7 @@ mod test { namespace = "pgdog_" datadog_api_key = "my-key" push_interval = 5000 + temporality_preference = "Delta" [otel.headers] Authorization = "Bearer token" @@ -174,6 +222,10 @@ 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" diff --git a/pgdog/src/stats/otel.rs b/pgdog/src/stats/otel.rs index e58f91f39..4a9bc97ac 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -310,13 +310,19 @@ fn build_attributes(labels: &[(String, String)], common_attrs: &[KeyValue]) -> V /// 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 temporality = match config.config.otel.temporality_preference { + let otel = &config.config.otel; + + let temporality = match otel + .temporality_preference + .expect("temporality preference is filled in config::load") + { OtelTemporalityPreference::Cumulative => SumAggregationTemporality::Cumulative, OtelTemporalityPreference::Delta | OtelTemporalityPreference::LowMemory => { SumAggregationTemporality::Delta } }; - let namespace = config.config.otel.namespace.as_deref(); + + let namespace = otel.namespace.as_deref(); build_request_with_state(&COUNTER_STATE, temporality, namespace, now, metrics) } @@ -429,7 +435,7 @@ mod test { use crate::config::{self, ConfigAndUsers}; let mut cfg = ConfigAndUsers::default(); - cfg.config.otel.temporality_preference = OtelTemporalityPreference::Delta; + cfg.config.otel.temporality_preference = Some(OtelTemporalityPreference::Delta); config::set(cfg).expect("set config"); let metric = Metric::new(PoolMetric { @@ -843,4 +849,29 @@ mod test { "time_unix_nano should advance while start_time_unix_nano stays put" ); } + + #[test] + fn datadog_api_key_defaults_to_delta() { + let _test_lock = TEST_LOCK.lock(); + + use crate::config::{self, ConfigAndUsers}; + let mut cfg = ConfigAndUsers::default(); + cfg.config.otel.datadog_api_key = Some("abc".into()); + config::set(cfg).expect("set config"); + + let metric = Metric::new(PoolMetric { + name: "total_query_count".into(), + measurements: vec![Measurement { + labels: vec![], + measurement: MeasurementType::Integer(1), + }], + help: "Total queries".into(), + unit: None, + metric_type: Some(OpenMetricType::Counter), + }); + + let request = build_request(&[&metric], &now_nanos()); + let json = serde_json::to_string(&request).expect("serialize"); + assert!(json.contains("\"aggregationTemporality\":1")); + } } diff --git a/pgdog/src/stats/otel_exporter.rs b/pgdog/src/stats/otel_exporter.rs index b78d8707d..6fec3faa1 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; From 3c3d165b267ed69be689ae4d06a4b2fa22e5f532 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Sat, 1 Aug 2026 04:45:31 -0400 Subject: [PATCH 10/12] 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` From 74dfe77931075cce1902d0e9e5a770e26ba5923d Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Fri, 7 Aug 2026 13:40:33 -0400 Subject: [PATCH 11/12] update 9918e9 to use typed OpenMetricType --- pgdog/src/stats/pools.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pgdog/src/stats/pools.rs b/pgdog/src/stats/pools.rs index c31ddf6e5..b5affff7f 100644 --- a/pgdog/src/stats/pools.rs +++ b/pgdog/src/stats/pools.rs @@ -737,7 +737,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 { @@ -753,7 +753,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 { @@ -769,7 +769,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 { From 81dc14f7e29de434c6276ad207d5d777964a6473 Mon Sep 17 00:00:00 2001 From: Kennan Hunter Date: Fri, 7 Aug 2026 14:34:56 -0400 Subject: [PATCH 12/12] otel: derive effective temporality at read time Moves the datadog-implied mapping into Otel::effective_temporality_preference() and calls it from the OTLP request builder. Fixes tests that install a config directly bypassing ConfigAndUsers::load --- pgdog-config/src/core.rs | 18 ------------------ pgdog-config/src/otel.rs | 30 ++++++++++++++++++++++++++++++ pgdog/src/stats/otel.rs | 30 +----------------------------- pgdog/src/stats/otel_exporter.rs | 3 --- 4 files changed, 31 insertions(+), 50 deletions(-) diff --git a/pgdog-config/src/core.rs b/pgdog-config/src/core.rs index 359284004..0c1fe7d77 100644 --- a/pgdog-config/src/core.rs +++ b/pgdog-config/src/core.rs @@ -105,24 +105,6 @@ impl ConfigAndUsers { warn!("admin password has been randomly generated"); } - match ( - &mut config.otel.temporality_preference, - &config.otel.datadog_api_key, - ) { - // Here if temporality_preference isn't present, we set it based on - // if datadog is present - (default_cumulative_temporality @ None, None) => { - *default_cumulative_temporality = Some(OtelTemporalityPreference::Cumulative) - } - // datadog is present so we set it to delta - (delta_temporality_because_of_datadog @ None, Some(_datadog_api)) => { - *delta_temporality_because_of_datadog = Some(OtelTemporalityPreference::Delta) - } - (Some(_), _) => { - // We don't have to set temporality - } - } - let config_and_users = ConfigAndUsers { config, users, diff --git a/pgdog-config/src/otel.rs b/pgdog-config/src/otel.rs index 80fbde430..e7434be6e 100644 --- a/pgdog-config/src/otel.rs +++ b/pgdog-config/src/otel.rs @@ -78,6 +78,15 @@ pub struct Otel { } 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()) } @@ -232,6 +241,27 @@ mod test { ); } + #[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/src/stats/otel.rs b/pgdog/src/stats/otel.rs index 4a9bc97ac..cce44cf75 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -312,10 +312,7 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ let config = crate::config::config(); let otel = &config.config.otel; - let temporality = match otel - .temporality_preference - .expect("temporality preference is filled in config::load") - { + let temporality = match otel.effective_temporality_preference() { OtelTemporalityPreference::Cumulative => SumAggregationTemporality::Cumulative, OtelTemporalityPreference::Delta | OtelTemporalityPreference::LowMemory => { SumAggregationTemporality::Delta @@ -849,29 +846,4 @@ mod test { "time_unix_nano should advance while start_time_unix_nano stays put" ); } - - #[test] - fn datadog_api_key_defaults_to_delta() { - let _test_lock = TEST_LOCK.lock(); - - use crate::config::{self, ConfigAndUsers}; - let mut cfg = ConfigAndUsers::default(); - cfg.config.otel.datadog_api_key = Some("abc".into()); - config::set(cfg).expect("set config"); - - let metric = Metric::new(PoolMetric { - name: "total_query_count".into(), - measurements: vec![Measurement { - labels: vec![], - measurement: MeasurementType::Integer(1), - }], - help: "Total queries".into(), - unit: None, - metric_type: Some(OpenMetricType::Counter), - }); - - let request = build_request(&[&metric], &now_nanos()); - let json = serde_json::to_string(&request).expect("serialize"); - assert!(json.contains("\"aggregationTemporality\":1")); - } } diff --git a/pgdog/src/stats/otel_exporter.rs b/pgdog/src/stats/otel_exporter.rs index 6fec3faa1..c8b47882c 100644 --- a/pgdog/src/stats/otel_exporter.rs +++ b/pgdog/src/stats/otel_exporter.rs @@ -102,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; @@ -112,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 {