diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7fc36a1..df43cab 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,10 +32,14 @@ jobs: run: cargo clippy --all-targets --all-features -- -D warnings - name: Run clippy (no default features) run: cargo clippy --all-targets --no-default-features -- -D warnings + - name: Run clippy (discovery only) + run: cargo clippy --all-targets --no-default-features --features discovery -- -D warnings - name: Run tests (all features) run: cargo test --all-features - name: Run tests (no default features) run: cargo test --no-default-features + - name: Run tests (discovery only) + run: cargo test --no-default-features --features discovery - name: Run clippy (OTAP compatibility suite) env: CARGO_TARGET_DIR: target @@ -56,6 +60,8 @@ jobs: run: cargo check --all-targets --all-features - name: Check MSRV (no default features) run: cargo check --all-targets --no-default-features + - name: Check MSRV (discovery only) + run: cargo check --all-targets --no-default-features --features discovery go-fixtures: name: Go testdata sync check diff --git a/Cargo.lock b/Cargo.lock index 176d167..e7d8c44 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -211,6 +211,8 @@ dependencies = [ "prost-build", "prost-types", "protox", + "serde", + "serde_json", "thiserror", ] @@ -367,6 +369,12 @@ dependencies = [ "either", ] +[[package]] +name = "itoa" +version = "1.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3bd1574e617a9a6c1f8f8181bd7e841e1f20e4277607ef28945765188d6284d4" + [[package]] name = "js-sys" version = "0.3.107" @@ -701,6 +709,49 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "serde_json" +version = "1.0.154" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + [[package]] name = "shlex" version = "2.0.1" @@ -946,3 +997,9 @@ dependencies = [ "quote", "syn 2.0.119", ] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index 8180de8..a37a80e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,7 @@ exclude = ["tests/otap-compat"] [features] default = ["arrow", "dto"] arrow = ["dep:arrow-array", "dep:arrow-buffer", "dep:arrow-schema"] +discovery = ["dep:serde", "dep:serde_json"] dto = ["dep:prost", "dep:prost-types"] [dependencies] @@ -23,6 +24,8 @@ arrow-buffer = { version = "59", default-features = false, optional = true } arrow-schema = { version = "59", default-features = false, optional = true } prost = { version = "0.14", optional = true } prost-types = { version = "0.14", optional = true } +serde = { version = "1", features = ["derive"], optional = true } +serde_json = { version = "1", optional = true } thiserror = "2" [dev-dependencies] diff --git a/src/discovery.rs b/src/discovery.rs new file mode 100644 index 0000000..cd23405 --- /dev/null +++ b/src/discovery.rs @@ -0,0 +1,250 @@ +// Copyright 2026 The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Prometheus-compatible target discovery configuration and target group types. + +use std::collections::BTreeMap; +use std::time::Duration; + +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + +/// Reserved Prometheus label holding the `:` scrape target address. +pub const ADDRESS_LABEL: &str = "__address__"; + +/// Ordered map of label names to label values matching Prometheus `model.LabelSet`. +pub type LabelSet = BTreeMap; + +/// Error returned when validating or running target discovery. +#[derive(Debug, thiserror::Error)] +pub enum DiscoveryError { + /// Configuration validation failed. + #[error("{0}")] + InvalidConfig(String), + /// I/O or filesystem error. + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + /// JSON payload decoding error. + #[error("JSON decode error: {0}")] + Json(#[from] serde_json::Error), +} + +/// Static target group configuration matching Prometheus `static_configs` entries. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StaticConfig { + /// List of target `":"` addresses. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub targets: Vec, + /// Labels assigned to all targets in the group. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub labels: LabelSet, +} + +impl StaticConfig { + /// Converts this static configuration entry into a [`TargetGroup`] with the given `source`. + pub fn into_target_group(self, source: impl Into) -> TargetGroup { + let targets = self + .targets + .into_iter() + .map(|addr| BTreeMap::from([(ADDRESS_LABEL.to_owned(), addr)])) + .collect(); + TargetGroup { + targets, + labels: self.labels, + source: source.into(), + } + } +} + +/// A group of discovered targets sharing a common label set and discovery source, +/// matching Prometheus `discovery/targetgroup.Group`. +/// +/// When serialized or deserialized directly with Serde, `TargetGroup` uses the same +/// `{ "targets": ["host:port", ...], "labels": { ... } }` wire format as Prometheus's +/// `targetgroup.Group`. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct TargetGroup { + /// Discovered target label sets (each containing at least [`ADDRESS_LABEL`]). + pub targets: Vec, + /// Common labels shared across all targets in this group. + pub labels: LabelSet, + /// Identifier of the discovery source that produced this target group. + pub source: String, +} + +impl Serialize for TargetGroup { + fn serialize(&self, serializer: S) -> Result { + #[derive(Serialize)] + struct TargetGroupWire<'a> { + targets: Vec<&'a str>, + #[serde(skip_serializing_if = "BTreeMap::is_empty")] + labels: &'a LabelSet, + } + + let targets: Vec<&str> = self + .targets + .iter() + .filter_map(|t| t.get(ADDRESS_LABEL).map(String::as_str)) + .collect(); + let raw = TargetGroupWire { + targets, + labels: &self.labels, + }; + raw.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for TargetGroup { + fn deserialize>(deserializer: D) -> Result { + let raw = StaticConfig::deserialize(deserializer)?; + Ok(raw.into_target_group(String::new())) + } +} + +/// Resolves a slice of [`StaticConfig`] entries into [`TargetGroup`]s with sequential +/// `"0"`, `"1"`, ... source identifiers matching Prometheus `discovery.NewStaticProvider`. +pub fn resolve_static_configs(configs: &[StaticConfig]) -> Vec { + configs + .iter() + .enumerate() + .map(|(idx, cfg)| cfg.clone().into_target_group(idx.to_string())) + .collect() +} + +/// Top-level discovery configuration container matching the discovery fields on a +/// Prometheus `scrape_config`. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DiscoveryConfigs { + /// Static target group configurations (`static_configs`). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub static_configs: Vec, +} + +impl DiscoveryConfigs { + /// Validates all configured service discovery sections according to Prometheus rules. + pub fn validate(&mut self) -> Result<(), DiscoveryError> { + Ok(()) + } + + /// Resolves the configured `static_configs` into [`TargetGroup`]s. + pub fn resolve_static(&self) -> Vec { + resolve_static_configs(&self.static_configs) + } +} + +/// Parses a Prometheus duration string (such as `"30s"`, `"1m"`, `"500ms"`, `"1h30m"`, `"7d"`, +/// `"1w"`, `"1y"`, or `"0"`) into a [`Duration`], matching `prometheus/common/model.ParseDuration`. +pub fn parse_duration(s: &str) -> Result { + if s == "0" { + return Ok(Duration::ZERO); + } + if s.is_empty() { + return Err(DiscoveryError::InvalidConfig( + "empty duration string".to_owned(), + )); + } + + let orig = s; + let bytes = s.as_bytes(); + let mut idx = 0; + let mut total_ms: u64 = 0; + let mut last_unit_pos: u8 = 0; + + while idx < bytes.len() { + if !bytes[idx].is_ascii_digit() { + return Err(DiscoveryError::InvalidConfig(format!( + "not a valid duration string: {orig:?}" + ))); + } + let num_start = idx; + while idx < bytes.len() && bytes[idx].is_ascii_digit() { + idx += 1; + } + let val: u64 = s[num_start..idx].parse().map_err(|_| { + DiscoveryError::InvalidConfig(format!("not a valid duration string: {orig:?}")) + })?; + + let unit_start = idx; + while idx < bytes.len() && !bytes[idx].is_ascii_digit() { + idx += 1; + } + if unit_start == idx { + return Err(DiscoveryError::InvalidConfig(format!( + "not a valid duration string: {orig:?}" + ))); + } + let unit = &s[unit_start..idx]; + let (pos, mult_ms): (u8, u64) = match unit { + "y" => (1, 365 * 24 * 60 * 60 * 1000), + "w" => (2, 7 * 24 * 60 * 60 * 1000), + "d" => (3, 24 * 60 * 60 * 1000), + "h" => (4, 60 * 60 * 1000), + "m" => (5, 60 * 1000), + "s" => (6, 1000), + "ms" => (7, 1), + _ => { + return Err(DiscoveryError::InvalidConfig(format!( + "unknown unit {unit:?} in duration {orig:?}" + ))); + } + }; + if pos <= last_unit_pos { + return Err(DiscoveryError::InvalidConfig(format!( + "not a valid duration string: {orig:?}" + ))); + } + last_unit_pos = pos; + + let part_ms = val + .checked_mul(mult_ms) + .ok_or_else(|| DiscoveryError::InvalidConfig("duration out of range".to_owned()))?; + total_ms = total_ms + .checked_add(part_ms) + .ok_or_else(|| DiscoveryError::InvalidConfig("duration out of range".to_owned()))?; + } + + Ok(Duration::from_millis(total_ms)) +} + +/// Formats a [`Duration`] as a canonical Prometheus duration string (such as `"30s"` or `"1m30s"`), +/// matching `prometheus/common/model.Duration.String`. +pub fn format_duration(d: Duration) -> String { + if d.is_zero() { + return "0s".to_owned(); + } + let mut ms: u128 = d.as_millis().max(1); + + let mut out = String::new(); + let mut append_unit = |unit: &str, mult: u128, exact: bool| { + if exact && !ms.is_multiple_of(mult) { + return; + } + let v = ms / mult; + if v > 0 { + out.push_str(&v.to_string()); + out.push_str(unit); + ms -= v * mult; + } + }; + + append_unit("y", 1000 * 60 * 60 * 24 * 365, true); + append_unit("w", 1000 * 60 * 60 * 24 * 7, true); + append_unit("d", 1000 * 60 * 60 * 24, false); + append_unit("h", 1000 * 60 * 60, false); + append_unit("m", 1000 * 60, false); + append_unit("s", 1000, false); + append_unit("ms", 1, false); + + out +} diff --git a/src/lib.rs b/src/lib.rs index b878c03..f28a3ca 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -15,6 +15,8 @@ #[cfg(feature = "arrow")] pub mod arrow; +#[cfg(feature = "discovery")] +pub mod discovery; #[cfg(feature = "dto")] pub mod dto; pub mod format; diff --git a/tests/discovery.rs b/tests/discovery.rs new file mode 100644 index 0000000..c90928a --- /dev/null +++ b/tests/discovery.rs @@ -0,0 +1,121 @@ +// Copyright 2026 The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Integration tests for Prometheus target discovery (`static_configs`, `TargetGroup`, and durations). + +#![cfg(feature = "discovery")] + +use std::time::Duration; + +use expfmt::discovery::{ + ADDRESS_LABEL, DiscoveryConfigs, LabelSet, StaticConfig, TargetGroup, format_duration, + parse_duration, +}; + +fn label_set(pairs: &[(&str, &str)]) -> LabelSet { + pairs + .iter() + .map(|(k, v)| ((*k).to_owned(), (*v).to_owned())) + .collect() +} + +#[test] +fn static_configs_and_target_group_serde_match_prometheus() { + let json = r#"{ + "static_configs": [ + { + "targets": ["127.0.0.1:9090", "[::1]:9100"], + "labels": { + "env": "prod", + "job": "prometheus" + } + }, + { + "targets": ["10.0.0.1:8080"] + } + ] + }"#; + + let mut cfg: DiscoveryConfigs = serde_json::from_str(json).expect("deserialize static_configs"); + cfg.validate().expect("validate static_configs"); + + let groups = cfg.resolve_static(); + assert_eq!(groups.len(), 2); + assert_eq!(groups[0].source, "0"); + assert_eq!( + groups[0].targets, + vec![ + label_set(&[(ADDRESS_LABEL, "127.0.0.1:9090")]), + label_set(&[(ADDRESS_LABEL, "[::1]:9100")]), + ] + ); + assert_eq!( + groups[0].labels, + label_set(&[("env", "prod"), ("job", "prometheus")]) + ); + assert_eq!(groups[1].source, "1"); + assert_eq!( + groups[1].targets, + vec![label_set(&[(ADDRESS_LABEL, "10.0.0.1:8080")])] + ); + assert!(groups[1].labels.is_empty()); + + // TargetGroup direct serde round-trip matches Prometheus targetgroup.Group. + let tg_json = serde_json::to_string(&groups[0]).expect("serialize TargetGroup"); + let roundtrip: TargetGroup = serde_json::from_str(&tg_json).expect("deserialize TargetGroup"); + assert_eq!(roundtrip.targets, groups[0].targets); + assert_eq!(roundtrip.labels, groups[0].labels); + + // Empty TargetGroup always serializes `"targets": []` matching Prometheus targetgroup.Group.MarshalJSON. + let empty_tg = TargetGroup::default(); + let empty_json = serde_json::to_string(&empty_tg).expect("serialize empty TargetGroup"); + assert_eq!(empty_json, r#"{"targets":[]}"#); + let empty_roundtrip: TargetGroup = + serde_json::from_str(&empty_json).expect("deserialize empty TargetGroup"); + assert!(empty_roundtrip.targets.is_empty()); + assert!(empty_roundtrip.labels.is_empty()); + + // Unknown fields are rejected on StaticConfig and DiscoveryConfigs. + assert!(serde_json::from_str::(r#"{"target": ["127.0.0.1:80"]}"#).is_err()); + assert!(serde_json::from_str::(r#"{"static_config": []}"#).is_err()); +} + +#[test] +fn prometheus_duration_parsing_and_formatting() { + assert_eq!(parse_duration("0").unwrap(), Duration::ZERO); + assert_eq!(parse_duration("0s").unwrap(), Duration::ZERO); + assert_eq!(parse_duration("30s").unwrap(), Duration::from_secs(30)); + assert_eq!(parse_duration("1m30s").unwrap(), Duration::from_secs(90)); + assert_eq!( + parse_duration("2h15m500ms").unwrap(), + Duration::from_millis((2 * 3600 + 15 * 60) * 1000 + 500) + ); + assert_eq!( + parse_duration("1w").unwrap(), + Duration::from_secs(7 * 24 * 3600) + ); + + assert_eq!(format_duration(Duration::ZERO), "0s"); + assert_eq!(format_duration(Duration::from_micros(500)), "1ms"); + assert_eq!(format_duration(Duration::from_secs(30)), "30s"); + assert_eq!(format_duration(Duration::from_secs(90)), "1m30s"); + assert_eq!(format_duration(Duration::from_secs(7 * 24 * 3600)), "1w"); + assert_eq!(format_duration(Duration::from_secs(8 * 24 * 3600)), "8d"); + assert_eq!(format_duration(Duration::from_secs(365 * 24 * 3600)), "1y"); + assert_ne!(format_duration(Duration::MAX), "0s"); + + assert!(parse_duration("").is_err()); + assert!(parse_duration("1s1m").is_err()); + assert!(parse_duration("10x").is_err()); + assert!(parse_duration("999999999999999y").is_err()); +}