diff --git a/.schema/pgdog.schema.json b/.schema/pgdog.schema.json index a334988f3..34442e29a 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": 10000, + "temporality_preference": "Cumulative" } }, "plugins": { @@ -1598,10 +1599,42 @@ "format": "uint64", "default": 10000, "minimum": 0 + }, + "temporality_preference": { + "description": "Describes how the exported metric points should be described.\n\nSee https://opentelemetry.io/docs/specs/otel/metrics/data-model/#metric-points\n\n_Default:_ `Cumulative`, or `Delta` when `datadog_api_key` is set.\n\nEnv: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE`", + "anyOf": [ + { + "$ref": "#/$defs/OtelTemporalityPreference" + }, + { + "type": "null" + } + ], + "default": "Cumulative" } }, "additionalProperties": false }, + "OtelTemporalityPreference": { + "description": "Aggregation temporality used when exporting OTLP metric points.", + "oneOf": [ + { + "description": "Points report the value accumulated since the exporter started.", + "type": "string", + "const": "Cumulative" + }, + { + "description": "Points report the change since the last export.", + "type": "string", + "const": "Delta" + }, + { + "description": "Delta for sums, cumulative for histograms; minimizes exporter memory.", + "type": "string", + "const": "LowMemory" + } + ] + }, "PassthroughAuth": { "description": "toggle automatic creation of connection pools given the user name, database and password.\n\nSee [passthrough authentication](https://docs.pgdog.dev/features/authentication/#passthrough-authentication).\n\n", "oneOf": [ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 23f413c36..8c8b2bf30 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -24,3 +24,4 @@ Contributions are welcome. If you see a bug, feel free to submit a PR with a fix 1. Please format your code with `cargo fmt`. 2. If you're feeling generous, `cargo clippy` as well. 3. Please write and include tests. This is production software used in one of the most important areas of the stack. +4. If changes have been made to configuration schemas, run `cargo run -p pgdog-jsonschema` diff --git a/Cargo.lock b/Cargo.lock index 7717242b0..173e16ff6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3398,6 +3398,7 @@ dependencies = [ "semver", "serde", "serde_json", + "serde_repr", "sha1", "smallvec", "socket2 0.5.10", @@ -4607,6 +4608,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_repr" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d3b1629de253c70a0508c3899572da79ca359fdab27c7920ff00406df418906" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "serde_spanned" version = "0.6.9" @@ -5115,6 +5127,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" diff --git a/pgdog-config/src/core.rs b/pgdog-config/src/core.rs index 491cd9742..0c1fe7d77 100644 --- a/pgdog-config/src/core.rs +++ b/pgdog-config/src/core.rs @@ -6,6 +6,7 @@ use std::fs::read_to_string; use std::path::{Path, PathBuf}; use tracing::{error, info, warn}; +use crate::otel_temporality::OtelTemporalityPreference; use crate::sharding::ShardedSchema; use crate::util::random_string; use crate::{ @@ -120,6 +121,7 @@ impl ConfigAndUsers { self.config.check(); self.users.check(&self.config); self.validate_server_auth()?; + self.warn_if_data_dog_cumulative(); Ok(()) } @@ -149,6 +151,33 @@ impl ConfigAndUsers { Ok(()) } + fn warn_if_data_dog_cumulative(&self) { + match ( + &self.config.otel.datadog_api_key, + &self.config.otel.temporality_preference, + ) { + (Some(_datadog_present), Some(OtelTemporalityPreference::Cumulative)) + if std::env::var("IGNORE_DATADOG_CUMULATIVE_WARNING") + .ok() + .as_deref() + != Some("1") => + { + warn!( + "Sending Cumulative OTLP sums/histograms to Datadog is stateful and lossy: \ + all points on a timeseries must reach the same Agent/exporter (constraining \ + how you scale collectors), the first point of a new series may be dropped \ + (causing gaps on restart), and histogram min/max may be missing or \ + approximated. See \ + https://docs.datadoghq.com/opentelemetry/guide/otlp_delta_temporality/?tab=python#implications-of-using-cumulative-aggregation-temporality. \ + Set IGNORE_DATADOG_CUMULATIVE_WARNING=1 to silence." + ); + } + _ => { + // valid + } + } + } + /// Prepared statements are enabled. pub fn prepared_statements(&self) -> PreparedStatements { // Disable prepared statements automatically in session mode @@ -284,6 +313,7 @@ pub struct Config { /// /// #[serde(default)] + #[schemars(default = "Otel::schema_default")] pub otel: Otel, /// HashiCorp Vault settings, required for users configured with `server_auth = "vault"`. diff --git a/pgdog-config/src/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..e7434be6e 100644 --- a/pgdog-config/src/otel.rs +++ b/pgdog-config/src/otel.rs @@ -4,6 +4,8 @@ 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,10 +62,31 @@ pub struct Otel { /// /// Env: `OTEL_METRIC_EXPORT_INTERVAL` #[serde(default = "Otel::push_interval")] + #[schemars(default = "Otel::schema_default_push_interval")] pub push_interval: u64, + + /// Describes how the exported metric points should be described. + /// + /// See https://opentelemetry.io/docs/specs/otel/metrics/data-model/#metric-points + /// + /// _Default:_ `Cumulative`, or `Delta` when `datadog_api_key` is set. + /// + /// Env: `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` + #[serde(default = "Otel::temporality_preference")] + #[schemars(default = "Otel::schema_default_temporality_preference")] + pub temporality_preference: Option, } impl Otel { + pub fn effective_temporality_preference(&self) -> OtelTemporalityPreference { + self.temporality_preference + .unwrap_or(if self.datadog_api_key.is_some() { + OtelTemporalityPreference::Delta + } else { + OtelTemporalityPreference::Cumulative + }) + } + fn env_option_string(env_var: &str) -> Option { env::var(env_var).ok().filter(|s| !s.is_empty()) } @@ -99,6 +122,31 @@ impl Otel { .and_then(|v| v.parse().ok()) .unwrap_or(10_000) } + + fn temporality_preference() -> Option { + env::var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE") + .ok() + .and_then(|v| v.parse().ok()) + } + + fn schema_default_push_interval() -> u64 { + 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() + } + } } #[cfg(test)] @@ -133,6 +181,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] @@ -143,6 +217,7 @@ mod test { namespace = "pgdog_" datadog_api_key = "my-key" push_interval = 5000 + temporality_preference = "Delta" [otel.headers] Authorization = "Bearer token" @@ -156,12 +231,37 @@ mod test { assert_eq!(config.otel.namespace.as_deref(), Some("pgdog_")); assert_eq!(config.otel.datadog_api_key.as_deref(), Some("my-key")); assert_eq!(config.otel.push_interval, 5000); + assert_eq!( + config.otel.temporality_preference, + Some(OtelTemporalityPreference::Delta) + ); assert_eq!( config.otel.headers.get("Authorization").unwrap(), "Bearer token" ); } + #[test] + fn effective_temporality_defaults_to_delta_with_datadog_key() { + let mut otel = Otel::default(); + assert_eq!( + otel.effective_temporality_preference(), + OtelTemporalityPreference::Cumulative + ); + + otel.datadog_api_key = Some("abc".into()); + assert_eq!( + otel.effective_temporality_preference(), + OtelTemporalityPreference::Delta + ); + + otel.temporality_preference = Some(OtelTemporalityPreference::Cumulative); + assert_eq!( + otel.effective_temporality_preference(), + OtelTemporalityPreference::Cumulative + ); + } + #[test] fn namespace_from_env() { let _guard = set_env_var("PGDOG_OTEL_NAMESPACE", "pgdog_"); diff --git a/pgdog-config/src/otel_temporality.rs b/pgdog-config/src/otel_temporality.rs new file mode 100644 index 000000000..c2533e5dd --- /dev/null +++ b/pgdog-config/src/otel_temporality.rs @@ -0,0 +1,107 @@ +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Aggregation temporality used when exporting OTLP metric points. +// Note: Derive FromStr is case insensitive, matching OTEL behavior, though serde Deserialize +// see https://docs.rs/derive_more/latest/derive_more/derive.FromStr.html#empty-enums +#[derive( + derive_more::FromStr, Debug, Clone, Copy, PartialEq, Eq, Default, JsonSchema, Serialize, +)] +pub enum OtelTemporalityPreference { + /// Points report the value accumulated since the exporter started. + #[default] + Cumulative, + + /// Points report the change since the last export. + Delta, + /// Delta for sums, cumulative for histograms; minimizes exporter memory. + LowMemory, +} + +// Use case insensitive deserialization to match env var behavior +impl<'de> Deserialize<'de> for OtelTemporalityPreference { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + use std::str::FromStr; + + let s = String::deserialize(deserializer)?; + + // this from_str is case insensitive + Self::from_str(&s.to_ascii_lowercase()).map_err(|_| { + serde::de::Error::unknown_variant(&s, &["Cumulative", "Delta", "LowMemory"]) + }) + } +} + +#[cfg(test)] +mod test { + use super::*; + use std::str::FromStr; + + #[test] + fn default_is_cumulative() { + assert_eq!( + OtelTemporalityPreference::default(), + OtelTemporalityPreference::Cumulative, + ); + } + + #[test] + fn from_str_is_case_insensitive() { + let cases = [ + ("cumulative", OtelTemporalityPreference::Cumulative), + ("CUMULATIVE", OtelTemporalityPreference::Cumulative), + ("Cumulative", OtelTemporalityPreference::Cumulative), + ("delta", OtelTemporalityPreference::Delta), + ("DELTA", OtelTemporalityPreference::Delta), + ("Delta", OtelTemporalityPreference::Delta), + ("lowmemory", OtelTemporalityPreference::LowMemory), + ("LOWMEMORY", OtelTemporalityPreference::LowMemory), + ("LowMemory", OtelTemporalityPreference::LowMemory), + ]; + + for (input, expected) in cases { + assert_eq!( + OtelTemporalityPreference::from_str(input).unwrap(), + expected, + "input {input:?}", + ); + } + } + + #[test] + fn from_str_rejects_unknown_variant() { + assert!(OtelTemporalityPreference::from_str("nope").is_err()); + assert!(OtelTemporalityPreference::from_str("").is_err()); + } + + #[derive(Debug, Deserialize)] + struct Wrap { + t: OtelTemporalityPreference, + } + + #[test] + fn deserialize_is_case_insensitive() { + for (raw, expected) in [ + ("delta", OtelTemporalityPreference::Delta), + ("DELTA", OtelTemporalityPreference::Delta), + ("LowMemory", OtelTemporalityPreference::LowMemory), + ("lowmemory", OtelTemporalityPreference::LowMemory), + ("Cumulative", OtelTemporalityPreference::Cumulative), + ] { + let toml = format!("t = \"{raw}\""); + let w: Wrap = toml::from_str(&toml).expect("deserialize"); + assert_eq!(w.t, expected, "input {raw:?}"); + } + } + + #[test] + fn deserialize_rejects_unknown_variant() { + let err = toml::from_str::("t = \"histogram\"").unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains("histogram"), "message was: {msg}"); + assert!(msg.contains("Cumulative"), "message was: {msg}"); + } +} diff --git a/pgdog/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/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 407c39a67..cce44cf75 100644 --- a/pgdog/src/stats/otel.rs +++ b/pgdog/src/stats/otel.rs @@ -9,11 +9,12 @@ use std::time::{SystemTime, UNIX_EPOCH}; use once_cell::sync::Lazy; use parking_lot::Mutex; +use pgdog_config::otel_temporality::OtelTemporalityPreference; use serde::Serialize; use crate::util::hostname; -use super::open_metric::{MeasurementType, Metric}; +use super::open_metric::{Measurement, MeasurementType, Metric, OpenMetricType}; static RESOURCE_ATTRIBUTES: Lazy> = Lazy::new(resource_attributes); @@ -27,9 +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) + } +} + +static COUNTER_STATE: Lazy = Lazy::new(CounterState::default); pub fn now_nanos() -> String { SystemTime::now() @@ -89,11 +117,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, Clone, Copy)] +#[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, } @@ -202,16 +237,104 @@ 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 otel = &config.config.otel; + + let temporality = match otel.effective_temporality_preference() { + OtelTemporalityPreference::Cumulative => SumAggregationTemporality::Cumulative, + OtelTemporalityPreference::Delta | OtelTemporalityPreference::LowMemory => { + SumAggregationTemporality::Delta + } + }; + + let namespace = otel.namespace.as_deref(); + + build_request_with_state(&COUNTER_STATE, temporality, namespace, now, metrics) +} + +/// Injectable core of [`build_request`]. Takes counter state, temporality, +/// and namespace explicitly so tests can exercise the stateful counter logic +/// without touching global config or the process-wide static. +fn build_request_with_state( + state: &CounterState, + temporality: 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 { @@ -224,71 +347,25 @@ pub fn build_request(metrics: &[&Metric], now: &str) -> ExportMetricsServiceRequ .iter() .map(|metric| { let name = format!("{}.{}", namespace, metric.name()); - let is_counter = metric.metric_type() == "counter"; + let metric_type = metric.metric_type(); let data_points: Vec = metric .measurements() .iter() .filter_map(|m| { - let cumulative = measurement_to_f64(&m.measurement); - - let as_double = if is_counter { - let key = CounterKey { - metric: name.clone(), - labels: m.labels.clone(), - }; - let mut prev = PREV_COUNTERS.lock(); - let delta = cumulative - prev.get(&key).copied().unwrap_or(0.0); - prev.insert(key, cumulative); - - // Skip negative deltas (counter reset). - if delta < 0.0 { - return None; - } - delta - } else { - cumulative - }; - - let mut attributes: Vec = m - .labels - .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: None, + 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: 1, // DELTA - is_monotonic: true, - data_points, - }), - ) - } else { - (Some(Gauge { data_points }), None) - }; + let (gauge, sum) = wrap_data_points(metric_type, temporality, data_points); OtelMetric { name, @@ -353,6 +430,11 @@ mod test { fn counter_metric_produces_sum_json() { let _test_lock = TEST_LOCK.lock(); + use crate::config::{self, ConfigAndUsers}; + let mut cfg = ConfigAndUsers::default(); + cfg.config.otel.temporality_preference = Some(OtelTemporalityPreference::Delta); + config::set(cfg).expect("set config"); + let metric = Metric::new(PoolMetric { name: "total_query_count".into(), measurements: vec![Measurement { @@ -361,7 +443,7 @@ mod test { }], help: "Total queries".into(), unit: None, - metric_type: Some("counter".into()), + metric_type: Some(OpenMetricType::Counter), }); let request = build_request(&[&metric], &now_nanos()); @@ -454,7 +536,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()); @@ -569,4 +651,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" + ); + } } diff --git a/pgdog/src/stats/otel_exporter.rs b/pgdog/src/stats/otel_exporter.rs index b78d8707d..c8b47882c 100644 --- a/pgdog/src/stats/otel_exporter.rs +++ b/pgdog/src/stats/otel_exporter.rs @@ -4,7 +4,6 @@ //! to the configured endpoint (e.g. Datadog's `/api/v2/otlp/v1/metrics`). use std::time::Duration; - use tracing::{info, warn}; use super::otel; @@ -103,7 +102,6 @@ pub async fn run() { #[cfg(test)] mod test { - use crate::config::{self, ConfigAndUsers}; use crate::stats::Metric; use crate::stats::open_metric::{Measurement, MeasurementType}; use crate::stats::otel; @@ -113,8 +111,6 @@ mod test { fn serialized_payload_is_valid_json() { let _test_lock = crate::stats::otel::TEST_LOCK.lock(); - config::set(ConfigAndUsers::default()).unwrap(); - let metric = Metric::new(PoolMetric { name: "sv_idle".into(), measurements: vec![Measurement { diff --git a/pgdog/src/stats/pools.rs b/pgdog/src/stats/pools.rs index 8f8dbee44..b5affff7f 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 { @@ -741,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 { @@ -757,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 { @@ -773,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 { @@ -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 {