diff --git a/AGENTS.md b/AGENTS.md index c88a9ce..34445d0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,7 +24,7 @@ Do not use Rust language or std features newer than `rust-version` in `Cargo.tom ## Spec -[specs/expfmt/SPEC.md](specs/expfmt/SPEC.md) is the contract for `expfmt`. If you change behavior the spec describes, update the spec in the same PR. +[specs/SPEC.md](specs/SPEC.md) is the specification contract for `scrape.rs` (`expfmt`, `relabel`, and scrape pipeline integration). If you change behavior the spec describes, update the spec in the same PR. ## Test data diff --git a/Cargo.lock b/Cargo.lock index 176d167..5ebb7dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -211,6 +211,9 @@ dependencies = [ "prost-build", "prost-types", "protox", + "regex", + "serde", + "serde_json", "thiserror", ] @@ -367,6 +370,12 @@ dependencies = [ "either", ] +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + [[package]] name = "js-sys" version = "0.3.107" @@ -701,6 +710,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 +998,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..0db2232 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,9 +13,10 @@ categories = ["parser-implementations", "encoding"] exclude = ["tests/otap-compat"] [features] -default = ["arrow", "dto"] +default = ["arrow", "dto", "relabel"] arrow = ["dep:arrow-array", "dep:arrow-buffer", "dep:arrow-schema"] dto = ["dep:prost", "dep:prost-types"] +relabel = ["dep:regex", "dep:serde"] [dependencies] arrow-array = { version = "59", default-features = false, optional = true } @@ -23,18 +24,26 @@ 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 } +regex = { version = "1", optional = true } +serde = { version = "1", features = ["derive"], optional = true } thiserror = "2" [dev-dependencies] benchfmt = { git = "https://github.com/bwplotka/benchfmt.rs" } prost-build = "=0.14.4" protox = "0.10" +serde_json = "1" [[bench]] name = "decode" harness = false required-features = ["arrow"] +[[bench]] +name = "relabel" +harness = false +required-features = ["relabel"] + [lints.rust] missing_docs = "warn" unsafe_code = "forbid" diff --git a/README.md b/README.md index f06cc9a..be7d196 100644 --- a/README.md +++ b/README.md @@ -13,21 +13,23 @@ Service discoveries -> Target relabeling -> Scrape manager / HTTP fetch -> Streaming expfmt parser - -> Metric relabeling / drop hook + -> Stage 1: family `__name__` filter pushdown + -> Metric family assembly & string interning + -> Stage 2: dictionary-vectorized relabeling & `honor_labels` -> Columnar Apache Arrow (OTAP) batches or Protobuf DTOs ``` The project serves two main use cases: -1. Columnar scraping for OpenTelemetry Arrow. [`open-telemetry/otel-arrow#3277`](https://github.com/open-telemetry/otel-arrow/issues/3277) defines an OTAP-native `prometheus_scrape` receiver in Rust. `scrape.rs` streams scrape bytes through pre-allocation relabeling and metric family assembly directly into Apache Arrow (`OtapArrowRecords`) tables on thread-per-core `!Send` runtimes with bounded memory. +1. Columnar scraping for OpenTelemetry Arrow. [`open-telemetry/otel-arrow#3277`](https://github.com/open-telemetry/otel-arrow/issues/3277) defines an OTAP-native `prometheus_scrape` receiver in Rust. `scrape.rs` streams scrape bytes through early `__name__` family filtering, metric family assembly, and dictionary-vectorized relabeling directly into Apache Arrow (`OtapArrowRecords`) tables on thread-per-core `!Send` runtimes with bounded memory. 2. Standalone Rust Prometheus tooling. Validation harnesses such as OpenTelemetry Weaver live-checking `/metrics` endpoints against semantic conventions, CLIs, and custom collectors can use individual components (such as `expfmt` DTO decoding or `relabel`) without pulling in the full scrape runtime. ## Components and scope | Component | Status | Specification | Scope | | :--- | :--- | :--- | :--- | -| `expfmt` | In progress | [specs/expfmt/SPEC.md](./specs/expfmt/SPEC.md) | `Content-Type` and `Accept` negotiation, zero-copy streaming parsers for Prometheus Text 0.0.4, OpenMetrics 1.0, OpenMetrics 2.0, and Prometheus Protobuf, primary Apache Arrow / OTAP decoder (`ArrowDecoder`), and [`MetricFamily`](https://github.com/prometheus/prometheus/blob/main/prompb/io/prometheus/client/metrics.proto) protobuf DTO decoder. | -| `relabel` | Planned | TBD | Prometheus-compatible target and metric relabeling (`replace`, `keep`, `drop`, `hashmod`, `labelmap`, `labeldrop`, `labelkeep`, `lowercase`, `uppercase`, `keepequal`, `dropequal`), including pre-allocation filtering on borrowed parser label buffers. | +| `expfmt` | In progress | [specs/SPEC.md](./specs/SPEC.md) | `Content-Type` and `Accept` negotiation, zero-copy streaming parsers for Prometheus Text 0.0.4, OpenMetrics 1.0, OpenMetrics 2.0, and Prometheus Protobuf, primary Apache Arrow / OTAP decoder (`ArrowDecoder`), and [`MetricFamily`](https://github.com/prometheus/prometheus/blob/main/prompb/io/prometheus/client/metrics.proto) protobuf DTO decoder. | +| `relabel` | In progress | [specs/SPEC.md](./specs/SPEC.md#target-and-metric-relabeling-architecture) | Prometheus-compatible target and metric relabeling (`replace`, `keep`, `drop`, `hashmod`, `labelmap`, `labeldrop`, `labelkeep`, `lowercase`, `uppercase`, `keepequal`, `dropequal`), Stage 1 `__name__` filter pushdown, and Stage 2 dictionary-vectorized relabeling and `honor_labels` resolution. | | `discovery` | Planned | TBD | Standard Prometheus service discovery mechanisms (static targets, file-based SD, HTTP SD, DNS SD, Kubernetes SD) emitting target groups and `__meta_*` labels. | | `scrape` | Planned | TBD | Bounded HTTP scrape scheduling, target pool lifecycle, compression and content negotiation, stale marker handling, target state reporting, and self-telemetry. | @@ -51,8 +53,9 @@ Run the Go differential fixture extractor and tests (requires Go): cd tests/go-harness && go test ./... ``` -Run the decoding benchmarks in Go `benchfmt` format for [`benchstat`](https://pkg.go.dev/golang.org/x/perf/cmd/benchstat): +Run the decoding and relabeling benchmarks in Go `benchfmt` format for [`benchstat`](https://pkg.go.dev/golang.org/x/perf/cmd/benchstat): ```sh cargo bench --bench decode -- -bench '^BenchmarkDecode' -benchtime 1s -count 6 -cpu 2 -benchmem | tee decode.txt +cargo bench --bench relabel -- -bench '^BenchmarkRelabel' -benchtime 1s -count 6 -cpu 2 -benchmem | tee relabel.txt ``` diff --git a/benches/relabel.rs b/benches/relabel.rs new file mode 100644 index 0000000..d8c453e --- /dev/null +++ b/benches/relabel.rs @@ -0,0 +1,611 @@ +// 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. + +//! Throughput and allocation benchmarks for `expfmt::relabel` target discovery relabeling, +//! per-sample scalar `LabelSet` (`BTreeMap`) metric relabeling, and two-stage +//! dictionary-vectorized metric relabeling over `ArrowDecoder`-style interned `(key_id, val_id)` +//! series buffers. +//! +//! Recommended CLI invocation: +//! ```sh +//! cargo bench --bench relabel -- -bench '^BenchmarkRelabel' -benchtime 1s -count 6 -cpu 2 -benchmem | tee ${bench}.txt +//! ``` + +use std::collections::BTreeMap; +use std::hint::black_box; + +use benchfmt::{CountingAlloc, Suite}; +use expfmt::relabel::{ + Action, Config, LabelSet, METRIC_NAME_LABEL, Regexp, SingleLabelKeepCache, + SingleLabelTransformCache, TargetPopulateConfig, ValidationScheme, eval_key_keep_mask, + keep_metric_name, merge_target_and_metric_labels, populate_discovered_labels, populate_labels, + process_in_place, resolve_conflicting_exposed_keys, split_metric_name_pushdown, +}; + +#[global_allocator] +static ALLOC: CountingAlloc = CountingAlloc::new(); + +const K8S_RULES_JSON: &str = r#"[ + { + "source_labels": ["__meta_kubernetes_pod_container_port_name"], + "regex": ".*-metrics", + "action": "keep" + }, + { + "source_labels": ["__meta_kubernetes_pod_label_name"], + "action": "drop", + "regex": "" + }, + { + "source_labels": ["__meta_kubernetes_pod_phase"], + "regex": "Succeeded|Failed", + "action": "drop" + }, + { + "source_labels": ["__meta_kubernetes_pod_annotation_prometheus_io_scrape"], + "regex": "false", + "action": "drop" + }, + { + "source_labels": ["__meta_kubernetes_pod_annotation_prometheus_io_scheme"], + "target_label": "__scheme__", + "regex": "(https?)", + "replacement": "$1", + "action": "replace" + }, + { + "source_labels": ["__meta_kubernetes_pod_annotation_prometheus_io_path"], + "target_label": "__metrics_path__", + "regex": "(.+)", + "replacement": "$1", + "action": "replace" + }, + { + "source_labels": ["__address__", "__meta_kubernetes_pod_annotation_prometheus_io_port"], + "target_label": "__address__", + "regex": "(.+?)(\\:\\d+)?;(\\d+)", + "replacement": "$1:$3", + "action": "replace" + }, + { + "regex": "__meta_kubernetes_pod_annotation_prometheus_io_param_(.+)", + "replacement": "__param_$1", + "action": "labelmap" + }, + { + "regex": "__meta_kubernetes_pod_label_prometheus_io_label_(.+)", + "action": "labelmap" + }, + { + "regex": "__meta_kubernetes_pod_annotation_prometheus_io_label_(.+)", + "action": "labelmap" + }, + { + "source_labels": ["__meta_kubernetes_namespace", "__meta_kubernetes_pod_label_name"], + "separator": "/", + "target_label": "job", + "replacement": "$1", + "action": "replace" + }, + { + "source_labels": ["__meta_kubernetes_namespace"], + "target_label": "namespace", + "action": "replace" + }, + { + "source_labels": ["__meta_kubernetes_pod_name"], + "target_label": "pod", + "action": "replace" + }, + { + "source_labels": ["__meta_kubernetes_pod_container_name"], + "target_label": "container", + "action": "replace" + }, + { + "source_labels": [ + "__meta_kubernetes_pod_name", + "__meta_kubernetes_pod_container_name", + "__meta_kubernetes_pod_container_port_name" + ], + "separator": ":", + "target_label": "instance", + "action": "replace" + }, + { + "target_label": "cluster", + "replacement": "dev-us-central-0" + }, + { + "source_labels": ["__meta_kubernetes_namespace"], + "regex": "hosted-grafana", + "action": "drop" + }, + { + "source_labels": ["__address__"], + "target_label": "__tmp_hash", + "modulus": 3, + "action": "hashmod" + }, + { + "source_labels": ["__tmp_hash"], + "regex": "^1$", + "action": "keep" + }, + { + "regex": "__tmp_hash", + "action": "labeldrop" + } +]"#; + +/// Compact interned representation of a scraped series for benchmarking Stage 1 + Stage 2 +/// dictionary-vectorized relabeling against per-sample `LabelSet` relabeling. +#[derive(Clone)] +struct InternedSeries { + /// Pair of `(key_id, val_id)` attributes on this series. + attrs: Vec<(usize, usize)>, + /// Index of the `"env"` label value in `val_dict` (`None` if omitted). + env_val_id: Option, + /// Index of the `"pod"` label value in `val_dict` (`None` if omitted). + pod_val_id: Option, +} + +struct ScrapeFixture { + target_labels: LabelSet, + metric_relabel_configs: Vec, + family_names: Vec, + /// `200` families x `10` series per family = `2,000` scalar `LabelSet`s. + scalar_families: Vec>, + /// Deduplicated key dictionary (`32` unique keys). + key_dict: Vec, + /// Deduplicated value dictionary (`128` unique values). + val_dict: Vec, + /// `200` families x `10` interned series per family. + interned_families: Vec>, +} + +fn build_scrape_fixture(high_cardinality: bool) -> ScrapeFixture { + let target_labels = LabelSet::from([ + ("__address__".to_owned(), "10.0.0.1:9090".to_owned()), + ("job".to_owned(), "api-server".to_owned()), + ("instance".to_owned(), "10.0.0.1:9090".to_owned()), + ("cluster".to_owned(), "prod-eu-west-1".to_owned()), + ("method".to_owned(), "target-method".to_owned()), + ]); + + let mut metric_relabel_configs = vec![ + Config { + source_labels: vec![METRIC_NAME_LABEL.to_owned()], + regex: Regexp::must_new("(http|rpc|process)_.*"), + action: Action::Keep, + ..Config::default() + }, + Config { + regex: Regexp::must_new("ephemeral_.*"), + action: Action::LabelDrop, + ..Config::default() + }, + Config { + source_labels: vec!["env".to_owned()], + regex: Regexp::must_new("prod|staging"), + action: Action::Keep, + ..Config::default() + }, + Config { + source_labels: vec!["pod".to_owned()], + regex: Regexp::must_new("([a-z0-9-]+)-[0-9a-f]+"), + target_label: "app".to_owned(), + replacement: "$1".to_owned(), + action: Action::Replace, + ..Config::default() + }, + ]; + for cfg in &mut metric_relabel_configs { + cfg.validate(ValidationScheme::Utf8).unwrap(); + } + + // 32 unique label keys in the scrape batch. + let mut key_dict = vec![ + "env".to_owned(), + "pod".to_owned(), + "method".to_owned(), + "status".to_owned(), + "ephemeral_debug".to_owned(), + "ephemeral_trace".to_owned(), + "app".to_owned(), + ]; + while key_dict.len() < 32 { + key_dict.push(format!("dim_key_{}", key_dict.len())); + } + + let envs = ["prod", "staging", "dev"]; + let methods = ["GET", "POST", "PUT", "DELETE"]; + let statuses = ["200", "201", "400", "404", "500"]; + let mut val_dict = Vec::with_capacity(if high_cardinality { 22_050 } else { 128 }); + for e in envs { + val_dict.push(e.to_owned()); // 0..3 + } + for m in methods { + val_dict.push(m.to_owned()); // 3..7 + } + for s in statuses { + val_dict.push(s.to_owned()); // 7..12 + } + + let pod_base = val_dict.len(); + let pod_count = if high_cardinality { 2_000 } else { 64 }; + for i in 0..pod_count { + val_dict.push(format!("checkout-service-{i:04}-{:04x}", 0x7f00 + i)); + } + + let extra_base = val_dict.len(); + if high_cardinality { + // 20,000 additional unique values across 10 per-series high-cardinality keys (22,012 total). + for i in 0..20_000 { + val_dict.push(format!("req-trace-uid-{i:05}")); + } + } else { + while val_dict.len() < 128 { + val_dict.push(format!("val_{}", val_dict.len())); + } + } + + let prefixes = ["http", "go", "rpc", "node"]; + let mut family_names = Vec::with_capacity(200); + let mut scalar_families = Vec::with_capacity(200); + let mut interned_families = Vec::with_capacity(200); + + for fam_idx in 0..200 { + let prefix = prefixes[fam_idx % prefixes.len()]; + let family_name = format!("{prefix}_requests_family_{fam_idx}_total"); + family_names.push(family_name.clone()); + + let mut scalar_series = Vec::with_capacity(10); + let mut interned_series = Vec::with_capacity(10); + + for s_idx in 0..10 { + let global_s_idx = fam_idx * 10 + s_idx; + let env_val_id = (fam_idx + s_idx) % envs.len(); + let pod_val_id = pod_base + (global_s_idx % pod_count); + let method_val_id = 3 + (s_idx % methods.len()); + let status_val_id = 7 + ((fam_idx + s_idx) % statuses.len()); + let eph_val_id = extra_base + (global_s_idx % 40); + + let mut ls = LabelSet::new(); + ls.insert(METRIC_NAME_LABEL.to_owned(), family_name.clone()); + ls.insert("env".to_owned(), val_dict[env_val_id].clone()); + ls.insert("pod".to_owned(), val_dict[pod_val_id].clone()); + ls.insert("method".to_owned(), val_dict[method_val_id].clone()); + ls.insert("status".to_owned(), val_dict[status_val_id].clone()); + ls.insert("ephemeral_debug".to_owned(), val_dict[eph_val_id].clone()); + + let mut attrs = vec![ + (0, env_val_id), + (1, pod_val_id), + (2, method_val_id), + (3, status_val_id), + (4, eph_val_id), + ]; + if high_cardinality { + for extra_k in 0..10 { + let kid = 7 + extra_k; + let vid = extra_base + global_s_idx * 10 + extra_k; + ls.insert(key_dict[kid].clone(), val_dict[vid].clone()); + attrs.push((kid, vid)); + } + } + scalar_series.push(ls); + + interned_series.push(InternedSeries { + attrs, + env_val_id: Some(env_val_id), + pod_val_id: Some(pod_val_id), + }); + } + + scalar_families.push(scalar_series); + interned_families.push(interned_series); + } + + ScrapeFixture { + target_labels, + metric_relabel_configs, + family_names, + scalar_families, + key_dict, + val_dict, + interned_families, + } +} + +fn main() { + let mut k8s_cfgs: Vec = serde_json::from_str(K8S_RULES_JSON).unwrap(); + for cfg in &mut k8s_cfgs { + cfg.validate(ValidationScheme::Utf8).unwrap(); + } + let k8s_target = LabelSet::from([ + ("__address__".to_owned(), "10.132.183.40:80".to_owned()), + ( + "__meta_kubernetes_namespace".to_owned(), + "loki-boltdb-shipper".to_owned(), + ), + ( + "__meta_kubernetes_pod_container_init".to_owned(), + "false".to_owned(), + ), + ( + "__meta_kubernetes_pod_container_name".to_owned(), + "promtail".to_owned(), + ), + ( + "__meta_kubernetes_pod_container_port_name".to_owned(), + "http-metrics".to_owned(), + ), + ( + "__meta_kubernetes_pod_container_port_number".to_owned(), + "80".to_owned(), + ), + ( + "__meta_kubernetes_pod_container_port_protocol".to_owned(), + "TCP".to_owned(), + ), + ( + "__meta_kubernetes_pod_label_name".to_owned(), + "promtail-loki-boltdb-shipper".to_owned(), + ), + ( + "__meta_kubernetes_pod_name".to_owned(), + "promtail-loki-boltdb-shipper-jgtr7".to_owned(), + ), + ( + "__meta_kubernetes_pod_phase".to_owned(), + "Running".to_owned(), + ), + ("__meta_kubernetes_pod_ready".to_owned(), "true".to_owned()), + ]); + let k8s_pop_cfg = TargetPopulateConfig { + job_name: "kubernetes-pods".to_owned(), + scrape_interval: "15s".to_owned(), + scrape_timeout: "10s".to_owned(), + metrics_path: "/metrics".to_owned(), + scheme: "http".to_owned(), + params: BTreeMap::new(), + }; + let k8s_discovered = populate_discovered_labels(&k8s_target, &BTreeMap::new(), &k8s_pop_cfg); + + let low_card_fixture = build_scrape_fixture(false); + let high_card_fixture = build_scrape_fixture(true); + + let mut suite = Suite::from_env("expfmt") + .with_alloc(&ALLOC) + .bench("Relabel/case=kubernetes_target/mode=populate_labels", |b| { + let empty_group = BTreeMap::new(); + b.report_allocs(); + while b.keep_running() { + let res = populate_labels( + black_box(&k8s_target), + black_box(&empty_group), + black_box(&k8s_pop_cfg), + black_box(&k8s_cfgs), + ) + .unwrap(); + black_box(res); + } + }) + .bench( + "Relabel/case=kubernetes_target/mode=process_in_place", + |b| { + let mut labels = k8s_discovered.clone(); + b.report_allocs(); + while b.keep_running() { + labels.clone_from(black_box(&k8s_discovered)); + let kept = process_in_place(&mut labels, black_box(&k8s_cfgs)); + black_box(kept); + } + }, + ); + + for (case_name, fixture) in [ + ("scrape_2000_series", low_card_fixture), + ("scrape_2000_series_22k_vals", high_card_fixture), + ] { + let fixture = std::rc::Rc::new(fixture); + let scalar_fixture = std::rc::Rc::clone(&fixture); + let dict_fixture = std::rc::Rc::clone(&fixture); + + suite = suite + .bench( + format!("Relabel/case={case_name}/mode=scalar_per_sample"), + move |b| { + let fixture = &*scalar_fixture; + let mut work = LabelSet::new(); + b.report_allocs(); + while b.keep_running() { + let mut kept_series = 0usize; + for family in black_box(&fixture.scalar_families) { + for sample in family { + work.clone_from(sample); + merge_target_and_metric_labels( + &mut work, + black_box(&fixture.target_labels), + false, + ); + if process_in_place( + &mut work, + black_box(&fixture.metric_relabel_configs), + ) { + kept_series += 1; + } + } + } + black_box(kept_series); + } + }, + ) + .bench( + format!("Relabel/case={case_name}/mode=two_stage_dictionary"), + move |b| { + let fixture = &*dict_fixture; + let app_key_id = fixture + .key_dict + .iter() + .position(|k| k == "app") + .expect("app key in dictionary"); + let mut renamed_keys = Vec::new(); + let mut remapped_key_ids = Vec::new(); + let mut key_keep_mask = Vec::new(); + let mut keep_cache = SingleLabelKeepCache::new(); + let mut xform_cache = SingleLabelTransformCache::new(); + let mut mapped_val_ids: Vec> = Vec::new(); + let mut extra_key_arena = String::new(); + let mut extra_key_spans: Vec<(usize, usize)> = Vec::new(); + let mut extra_val_arena = String::new(); + let mut extra_val_spans: Vec<(usize, usize)> = Vec::new(); + let mut out_attrs: Vec<(usize, usize)> = Vec::new(); + let mut out_series_offsets: Vec = Vec::new(); + + let mut run_once = |fixture: &ScrapeFixture| -> (usize, usize) { + let (stage1, stage2) = + split_metric_name_pushdown(black_box(&fixture.metric_relabel_configs)); + let [label_drop_cfg, keep_env_cfg, replace_pod_cfg] = stage2 else { + panic!("expected 3 Stage 2 relabel rules in fixture"); + }; + + extra_key_arena.clear(); + extra_key_spans.clear(); + let has_collision = resolve_conflicting_exposed_keys( + black_box(&fixture.key_dict), + black_box(&fixture.target_labels), + false, + &mut renamed_keys, + ); + remapped_key_ids.clear(); + if has_collision { + remapped_key_ids.reserve(fixture.key_dict.len()); + for (kid, renamed) in renamed_keys.iter().enumerate() { + if let Some(exported) = renamed { + let new_kid = fixture.key_dict.len() + extra_key_spans.len(); + let start = extra_key_arena.len(); + extra_key_arena.push_str(exported); + extra_key_spans.push((start, extra_key_arena.len())); + remapped_key_ids.push(new_kid); + } else { + remapped_key_ids.push(kid); + } + } + } + + eval_key_keep_mask( + black_box(&fixture.key_dict), + label_drop_cfg, + &mut key_keep_mask, + ) + .unwrap(); + assert!(keep_cache.prepare(fixture.val_dict.len(), keep_env_cfg)); + assert!(xform_cache.prepare(fixture.val_dict.len(), replace_pod_cfg)); + + extra_val_arena.clear(); + extra_val_spans.clear(); + mapped_val_ids.clear(); + mapped_val_ids.resize(fixture.val_dict.len(), None); + out_attrs.clear(); + out_series_offsets.clear(); + out_series_offsets.push(0); + + for (fam_idx, family_name) in + black_box(&fixture.family_names).iter().enumerate() + { + if !keep_metric_name(family_name, stage1) { + continue; + } + for series in &fixture.interned_families[fam_idx] { + let env_kept = match series.env_val_id { + Some(vid) => keep_cache.keep_value( + vid, + black_box(&fixture.val_dict), + keep_env_cfg, + ), + None => keep_cache.keep_when_absent(), + }; + if !env_kept { + continue; + } + for &(kid, vid) in &series.attrs { + if key_keep_mask[kid] && kid != app_key_id { + let out_kid = if has_collision { + remapped_key_ids[kid] + } else { + kid + }; + out_attrs.push((out_kid, vid)); + } + } + match series.pod_val_id { + Some(vid) => { + let out_vid = if let Some(cached_vid) = mapped_val_ids[vid] + { + Some(cached_vid) + } else { + let src = &fixture.val_dict[vid]; + let xform = xform_cache.transform_value( + vid, + black_box(&fixture.val_dict), + replace_pod_cfg, + ); + xform.target_value(src).map(|new_val| { + let new_vid = + fixture.val_dict.len() + extra_val_spans.len(); + let start = extra_val_arena.len(); + extra_val_arena.push_str(new_val); + extra_val_spans + .push((start, extra_val_arena.len())); + mapped_val_ids[vid] = Some(new_vid); + new_vid + }) + }; + if let Some(new_vid) = out_vid { + out_attrs.push((app_key_id, new_vid)); + } + } + None => { + if let Some(new_val) = + xform_cache.when_absent().target_value("") + { + let new_vid = + fixture.val_dict.len() + extra_val_spans.len(); + let start = extra_val_arena.len(); + extra_val_arena.push_str(new_val); + extra_val_spans.push((start, extra_val_arena.len())); + out_attrs.push((app_key_id, new_vid)); + } + } + } + out_series_offsets.push(out_attrs.len()); + } + } + (out_series_offsets.len() - 1, out_attrs.len()) + }; + + // Warm scratch capacities once to measure steady-state scrape reuse. + black_box(run_once(fixture)); + + b.report_allocs(); + while b.keep_running() { + black_box(run_once(fixture)); + } + }, + ); + } + + suite.run(); +} diff --git a/specs/expfmt/SPEC.md b/specs/SPEC.md similarity index 72% rename from specs/expfmt/SPEC.md rename to specs/SPEC.md index 23ee72a..839466d 100644 --- a/specs/expfmt/SPEC.md +++ b/specs/SPEC.md @@ -1,6 +1,6 @@ -# `expfmt` specification and implementation plan +# `scrape.rs` specification and implementation plan -The `expfmt` crate in `scrape.rs` provides official Rust decoders for Prometheus and OpenMetrics wire exposition formats. +`scrape.rs` (`expfmt` crate) provides official Rust decoders for Prometheus and OpenMetrics wire exposition formats along with integrated Prometheus target and metric relabeling (`relabel`). ## Background and goals @@ -8,7 +8,7 @@ Rust metric pipelines and tools currently lack an official, spec-compliant parse This repository targets two use cases: -1. High-throughput columnar ingestion for OpenTelemetry Arrow. [`open-telemetry/otel-arrow#3277`](https://github.com/open-telemetry/otel-arrow/issues/3277) defines an OTAP-native `prometheus_scrape` receiver in Rust. Its steady-state path streams scrape bytes through pre-allocation relabeling and metric family assembly directly into Apache Arrow metric records. That avoids intermediate Go-style `pmetric` or SDK `metricdata` object graphs and fits thread-per-core `!Send` runtimes with bounded memory. +1. High-throughput columnar ingestion for OpenTelemetry Arrow. [`open-telemetry/otel-arrow#3277`](https://github.com/open-telemetry/otel-arrow/issues/3277) defines an OTAP-native `prometheus_scrape` receiver in Rust. Its steady-state path streams scrape bytes through early `__name__` series filtering, metric family assembly, and dictionary-vectorized relabeling directly into Apache Arrow metric records. That avoids intermediate Go-style `pmetric` or SDK `metricdata` object graphs and fits thread-per-core `!Send` runtimes with bounded memory. 2. Programmatic inspection and ecosystem tooling. Validation tools like OpenTelemetry Weaver live-checking `/metrics` endpoints against semantic conventions, test harnesses, and Rust CLIs need an ergonomic API that decodes any Prometheus or OpenMetrics payload into standard [`io.prometheus.client.MetricFamily`](https://github.com/prometheus/prometheus/blob/main/prompb/io/prometheus/client/metrics.proto) protobuf structs, matching Go's [`prometheus/common/expfmt`](https://github.com/prometheus/common/tree/main/expfmt). ## Supported formats and content negotiation @@ -40,16 +40,17 @@ Decoding is split into three chained layers so streaming callers pay zero per-sa │ Layer 0: Streaming event parser (text / proto) │ │ - Zero-copy borrowed slices (&[u8] / &str) │ │ - Yields Entry::{Help, Type, Unit, Series, Histogram} │ - │ - Pre-allocation metric relabel and drop hook │ └────────────────────────────┬────────────────────────────┘ │ ▼ Primary API ┌─────────────────────────────────────────────────────────┐ │ Layer 1: Arrow / OTAP decoder │ - │ - Metric family assembly │ - │ - Direct Arrow RecordBatches │ - │ - Reusable builders/interners │ - │ - Fits otel-arrow#3277 │ + │ - Stage 1 pushdown: series `__name__` keep/drop filter │ + │ - Metric family assembly & arena string interners │ + │ - Stage 2: dictionary-vectorized relabeling & │ + │ `honor_labels` key resolution │ + │ - Direct Arrow RecordBatches (fits otel-arrow#3277 │ + │ and DataFusion SIMD execution) │ └────────────────────────────┬────────────────────────────┘ │ ▼ Ergonomic API @@ -115,18 +116,39 @@ This is the primary API, designed for the [`open-telemetry/otel-arrow#3277`](htt ```text scrape bytes -> streaming parser - -> metric relabel/drop - -> metric family assembler - -> direct OTAP metric records + -> Stage 1: series `__name__` keep/drop filter pushdown + -> metric family assembler & arena string interners (`key_interner`, `val_interner`) + -> Stage 2: dictionary-vectorized relabeling & `honor_labels` + -> direct OTAP metric records (ready for DataFusion / Arrow SIMD execution) ``` - `ArrowDecoder` appends parsed metrics, attributes, number data points, summary data points, classic and exponential histogram data points, and exemplars directly into Arrow array builders (`arrow-array` and `arrow-schema`), matching the relational `id` and `parent_id` tables in [`rust/otap-dataflow/crates/pdata`](https://github.com/open-telemetry/otel-arrow/tree/main/rust/otap-dataflow/crates/pdata). Callers can materialize one batch set per scrape with `ArrowDecoder::decode`, pull bounded chunks on demand with `ArrowDecoder::decode_stream` to cap scratch memory and stay within OTAP's `u16` per-batch metric ID limit, or yield cooperatively between bounded steps with `DecodeBudget`, `DecodeStep`, `DecodeStream::step`, `DecodeCursor` (`ArrowDecoder::start_decode` / `start_decode_stream` / `step_decode`), `ArrowDecoder::decode_cooperative`, and `DecodeStream::next_cooperative` so multi-megabyte scrapes do not starve timers or peer tasks on single-threaded `!Send` `LocalSet` runtimes. - Metric families in text exposition are decoded as contiguous runs, with `# HELP`, `# TYPE`, and `# UNIT` metadata lines preceding that family's samples. Like OpenMetrics 1.0/2.0 and Prometheus's scrape loop, `ArrowDecoder` tracks only the active family cursor instead of maintaining a whole-scrape hash table to reassemble or reject previously seen names, so a non-contiguous repetition of an earlier metric name starts a fresh untyped run unless preceded by its own metadata. -- A `LabelFilter` / `Relabeler` hook runs on the borrowed `LabelsBuffer` from Layer 0 so dropped series never allocate in Arrow builders or metric family state. +- Rather than materializing a per-sample `BTreeMap` or running regular expressions on every scraped text line, `metric_relabel_configs` and `honor_labels` will wire into `ArrowDecoder` (Phase 4, `#35`) as a two-stage pipeline (see [Target and metric relabeling architecture](#target-and-metric-relabeling-architecture)): + 1. **Stage 1 (`__name__` series filter pushdown)**: Leading `keep`, `drop`, `keepequal`, and `dropequal` rules that inspect only `__name__` will run once per contiguous wire series-name transition (e.g. `foo_bucket`, `foo_sum`, `foo_count`, matching Prometheus `scrape/scrape.go`) so dropped series runs skip float parsing, label unescaping, string interning, and family assembly. + 2. **Stage 2 (dictionary-vectorized relabeling and `honor_labels`)**: Once kept series have been assembled into metric families (collapsing classic histogram `_bucket`, `_sum`, and `_count` lines into a single `HistogramDataPoints` row) and their label keys and values have been deduplicated in `key_interner` and `val_interner`, attribute-level `honor_labels` collision resolution, `labeldrop` / `labelkeep`, and single-label `keep` / `drop` / value transformations (`replace`, `lowercase`, `uppercase`, `hashmod`) will execute across unique dictionary entries (memoized per `source_labels[0]` value ID) instead of $O(N_{\text{samples}})$. - The metric family assembler groups classic histogram `_bucket`, `_sum`, and `_count` lines and summary `quantile`, `_sum`, and `_count` lines (plus OpenMetrics `_created` start timestamps) into complete metric families. - Configurable translation policies control counter `_total` and unit suffix handling and untyped series handling. - `ArrowDecoder` owns its scratch buffers and Arrow builders without internal locks or global state, so a per-core `!Send` receiver can reuse allocations across scrapes with bounded memory. +### Target and metric relabeling architecture + +The `relabel` module implements Prometheus `relabel_configs`, `metric_relabel_configs`, and `honor_labels` across three execution tiers matched to data cardinality: + +1. **Pre-scrape target discovery relabeling (`relabel_configs`)**: + - Runs once per discovered target endpoint (cold path) over `relabel::LabelSet` (`BTreeMap`) via `populate_discovered_labels` and `populate_labels`. + - Populates reserved discovery labels (`__address__`, `__scheme__`, `__metrics_path__`, `__scrape_interval__`, `__scrape_timeout__`, `__param_*`, `job`), executes all 11 Prometheus relabel actions, strips `__meta_*` labels, and defaults `instance` to `__address__`. + - Non-reserved target labels (`PopulatedTarget::public_labels()`) attach once per scrape to the batch's `ResourceAttrs` (`job` -> `service.name`, `instance` -> `service.instance.id`) rather than duplicating unchanged target labels onto every data-point attribute row. When a `metric_relabel_configs` rule reads or mutates a public target label key (such as `source_labels: ["job"]`, `target_label: "job"`, `labeldrop`, `labelkeep`, or `labelmap`), Stage 2 resolves that key against the post-`honor_labels` target label value (using the target label when `honor_labels = false` or when the series omits that key under `honor_labels = true`) and materializes per-point attribute overrides only if a rule mutates or drops the target label. +2. **Stage 1 pre-staging `__name__` filter pushdown (Layer 1 ingestion, wired in Phase 4 / `#35`)**: + - `relabel::split_metric_name_pushdown` extracts the leading prefix of `metric_relabel_configs` rules (`keep`, `drop`, `keepequal`, `dropequal`) that depend exclusively on `__name__`. + - `ArrowDecoder` will evaluate `relabel::keep_metric_name` against the wire series `__name__` (such as `foo_bucket` or `foo_count`, not the stripped family base name `foo`) once when transitioning to a new contiguous series-name run, discarding dropped series runs before parsing sample floats, unescaping labels, interning strings, or staging histogram buckets. +3. **Stage 2 dictionary-vectorized relabeling and Layer 3 DataFusion SIMD execution (`ArrowDecoder` wiring in Phase 4 / `#35`)**: + - After `ArrowDecoder` interns label keys into `key_interner` ($U_{\text{keys}}$ unique keys, typically ~32) and label values into `val_interner` ($U_{\text{vals}}$ unique values across all keys), dictionary-level primitives in `relabel` avoid per-sample regex and string allocations: + - **`honor_labels = false` (`relabel::resolve_conflicting_exposed_keys`)**: Checks `key_interner` against public target labels in $O(U_{\text{keys}})$. When no sample label key collides with a target label key (the common case), `honor_labels` is a zero-copy no-op. When a collision occurs, only the colliding dictionary entries are renamed to `exported_` in ascending key-length order. Note that whereas Go [`resolveConflictingExposedLabels`](https://github.com/prometheus/prometheus/blob/236a2b1fe8e0471586ff29cb8fd8929372643650/scrape/scrape.go#L682-L697) checks `exported_` occupancy per series (like scalar `merge_target_and_metric_labels`), `resolve_conflicting_exposed_keys` intentionally resolves collisions once across the batch-wide `key_interner` dictionary so key indices do not have to be rewritten per series; if another series in the same batch already exposes `exported_`, `` is renamed to `exported_exported_` across the batch dictionary. + - **`labeldrop` / `labelkeep` (`relabel::eval_key_keep_mask`)**: Evaluates the key regex once per unique entry in `key_interner` ($O(U_{\text{keys}})$) to build a boolean key-ID mask used to filter `(key_id, val_id)` attribute rows. + - **Single-label `keep` / `drop` (`relabel::SingleLabelKeepCache`, `relabel::eval_single_label_keep_mask`) and value transformations (`relabel::SingleLabelTransformCache`, `relabel::eval_single_label_value_transform`)**: Memoizes single-source-label predicates and transformations (`replace`, `lowercase`, `uppercase`, `hashmod`) lazily across only the `val_id`s seen under `source_labels[0]` on surviving series ($\le \min(U_{\text{source\_key}}, N_{\text{kept\_series}})$), avoiding touching unrelated high-cardinality values in `val_interner`. + - **Layer 3 DataFusion / Arrow compute in `otel-arrow`**: Downstream cross-series aggregations, SQL projections, and vectorized SIMD filtering execute directly on the dictionary-encoded `OtapArrowRecords` (`RecordBatch`es) inside `open-telemetry/otel-arrow`'s DataFusion engine without pulling `datafusion` into `expfmt`. + ### Layer 2: Proto DTO decoder For tests, CLIs, and validation tools such as OpenTelemetry Weaver live-check, the `dto` module is chained on top of Layer 1 (`ArrowDecoder`). It decodes any supported wire format via `ArrowDecoder` and converts the resulting Arrow / OTAP metric records into [`io.prometheus.client.MetricFamily`](https://github.com/prometheus/prometheus/blob/main/prompb/io/prometheus/client/metrics.proto) structs generated with `prost`. Keeping metric family assembly in Layer 1 avoids duplicating histogram and summary reconstruction logic and means every `dto` caller exercises the primary Arrow pipeline. The API matches [`expfmt.NewDecoder`](https://github.com/prometheus/common/blob/main/expfmt/decode.go) and [`expfmt.TextParser`](https://github.com/prometheus/common/blob/main/expfmt/text_parse.go) in Go: @@ -192,7 +214,7 @@ To verify that `ArrowDecoder` meets the [`open-telemetry/otel-arrow#3277`](https - Extend the Go `expfmt` encoder harness to cover `FmtOpenMetrics_2_0_0` and `FmtProtoDelim`. ### Phase 4: Relabeling hooks, pooling, benchmarks, fuzzing, and initial release -- Add the pre-allocation `LabelFilter` and relabeling hook on Layer 0 `LabelsBuffer` and dictionary/buffer reuse across scrapes in `ArrowDecoder`. +- Add Stage 1 family `__name__` filter pushdown and Stage 2 dictionary-vectorized `honor_labels` and attribute relabeling on `ArrowDecoder` interners, plus dictionary/buffer reuse across scrapes. - Add `cargo-fuzz` targets comparing Rust decoding against Go `expfmt` and `textparse`. - Add `criterion` benchmarks measuring throughput in MB/s and allocations per sample across Layer 0 `Parser`, Layer 1 `ArrowDecoder`, and Layer 2 `dto`. - Publish the `0.1.0` crate release and integrate with `open-telemetry/otel-arrow`. diff --git a/src/lib.rs b/src/lib.rs index b878c03..1199f1c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -21,3 +21,5 @@ pub mod format; // Layer 0 parsers remain crate-internal; `arrow` and `dto` expose the public decoding APIs. #[allow(dead_code, unused_imports)] pub(crate) mod parse; +#[cfg(feature = "relabel")] +pub mod relabel; diff --git a/src/relabel.rs b/src/relabel.rs new file mode 100644 index 0000000..4953152 --- /dev/null +++ b/src/relabel.rs @@ -0,0 +1,1982 @@ +// 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 and metric relabeling (`relabel_configs` and +//! `metric_relabel_configs`), ported from Prometheus `model/relabel` and `scrape/target.go`. + +use std::borrow::Cow; +use std::collections::{BTreeMap, HashSet}; +use std::fmt; +use std::sync::OnceLock; + +use regex::{CaptureLocations, Regex}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + +/// Reserved Prometheus label holding the metric name (`"__name__"`). +pub const METRIC_NAME_LABEL: &str = "__name__"; +/// Reserved Prometheus label holding the `:` scrape target address. +pub const ADDRESS_LABEL: &str = "__address__"; +/// Reserved Prometheus label holding the scrape HTTP scheme (`"http"` or `"https"`). +pub const SCHEME_LABEL: &str = "__scheme__"; +/// Reserved Prometheus label holding the scrape HTTP request path (e.g. `"/metrics"`). +pub const METRICS_PATH_LABEL: &str = "__metrics_path__"; +/// Reserved Prometheus label holding the target scrape interval duration string. +pub const SCRAPE_INTERVAL_LABEL: &str = "__scrape_interval__"; +/// Reserved Prometheus label holding the target scrape timeout duration string. +pub const SCRAPE_TIMEOUT_LABEL: &str = "__scrape_timeout__"; +/// Prefix for reserved Prometheus labels mapping to URL query parameters (`"__param_"`). +pub const PARAM_LABEL_PREFIX: &str = "__param_"; +/// Prefix for service-discovery metadata labels (`"__meta_"`). +pub const META_LABEL_PREFIX: &str = "__meta_"; +/// Prefix for temporary labels used during relabeling (`"__tmp_"`). +pub const TMP_LABEL_PREFIX: &str = "__tmp_"; +/// Prefix for all internal/reserved Prometheus labels (`"__"`). +pub const RESERVED_LABEL_PREFIX: &str = "__"; +/// Standard Prometheus job label (`"job"`). +pub const JOB_LABEL: &str = "job"; +/// Standard Prometheus instance label (`"instance"`). +pub const INSTANCE_LABEL: &str = "instance"; +/// Prefix prepended to conflicting metric labels when `honor_labels` is `false` (`"exported_"`). +pub const EXPORTED_LABEL_PREFIX: &str = "exported_"; + +/// Default `separator` in a Prometheus relabel configuration (`";"`). +pub const DEFAULT_SEPARATOR: &str = ";"; +/// Default `replacement` in a Prometheus relabel configuration (`"$1"`). +pub const DEFAULT_REPLACEMENT: &str = "$1"; +/// Default unanchored `regex` pattern in a Prometheus relabel configuration (`"(.*)"`). +pub const DEFAULT_REGEX_PATTERN: &str = "(.*)"; + +/// Ordered map of label names to label values matching Prometheus `model.LabelSet`. +pub type LabelSet = BTreeMap; + +/// Error returned when parsing, validating, or applying relabel configurations. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum RelabelError { + /// Relabel configuration validation error. + #[error("{0}")] + InvalidConfig(String), + /// Invalid regular expression pattern. + #[error("invalid regex: {0}")] + InvalidRegex(String), + /// Target population error after relabeling. + #[error("{0}")] + TargetError(String), +} + +/// Relabeling action matching Prometheus `relabel.Action`. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] +pub enum Action { + /// Performs a regular expression replacement on `target_label` (default). + #[default] + Replace, + /// Drops targets/samples for which the concatenated source labels do not match `regex`. + Keep, + /// Drops targets/samples for which the concatenated source labels match `regex`. + Drop, + /// Drops targets/samples for which the concatenated source labels do not equal `target_label`. + KeepEqual, + /// Drops targets/samples for which the concatenated source labels equal `target_label`. + DropEqual, + /// Sets `target_label` to the MD5 hash modulus of the concatenated source labels. + HashMod, + /// Copies labels whose names match `regex` to new label names computed via `replacement`. + LabelMap, + /// Drops any label whose name matches `regex`. + LabelDrop, + /// Drops any label whose name does not match `regex`. + LabelKeep, + /// Maps the concatenated source labels to lowercase and stores the result in `target_label`. + Lowercase, + /// Maps the concatenated source labels to uppercase and stores the result in `target_label`. + Uppercase, +} + +impl Action { + /// Returns the lowercase Prometheus string representation of this action. + pub const fn as_str(self) -> &'static str { + match self { + Self::Replace => "replace", + Self::Keep => "keep", + Self::Drop => "drop", + Self::KeepEqual => "keepequal", + Self::DropEqual => "dropequal", + Self::HashMod => "hashmod", + Self::LabelMap => "labelmap", + Self::LabelDrop => "labeldrop", + Self::LabelKeep => "labelkeep", + Self::Lowercase => "lowercase", + Self::Uppercase => "uppercase", + } + } + + /// Parses a case-insensitive relabel action name matching Prometheus `Action.UnmarshalYAML`. + pub fn parse(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "replace" => Ok(Self::Replace), + "keep" => Ok(Self::Keep), + "drop" => Ok(Self::Drop), + "keepequal" => Ok(Self::KeepEqual), + "dropequal" => Ok(Self::DropEqual), + "hashmod" => Ok(Self::HashMod), + "labelmap" => Ok(Self::LabelMap), + "labeldrop" => Ok(Self::LabelDrop), + "labelkeep" => Ok(Self::LabelKeep), + "lowercase" => Ok(Self::Lowercase), + "uppercase" => Ok(Self::Uppercase), + _ => Err(RelabelError::InvalidConfig(format!( + "unknown relabel action {s:?}" + ))), + } + } +} + +impl fmt::Display for Action { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl Serialize for Action { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_str(self.as_str()) + } +} + +impl<'de> Deserialize<'de> for Action { + fn deserialize>(deserializer: D) -> Result { + let s = Option::::deserialize(deserializer)?; + match s { + None => Ok(Self::Replace), + Some(val) => Self::parse(&val).map_err(serde::de::Error::custom), + } + } +} + +/// Label name validation scheme matching Prometheus `model.ValidationScheme`. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum ValidationScheme { + /// Unset scheme; inherits the global validation scheme passed to [`Config::validate`]. + #[default] + Unset, + /// UTF-8 label name validation (`model.UTF8Validation`, default in Prometheus v3+). + Utf8, + /// Legacy ASCII `[a-zA-Z_][a-zA-Z0-9_]*` label name validation (`model.LegacyValidation`). + Legacy, +} + +impl ValidationScheme { + /// Returns `true` if `name` is a valid Prometheus label name under this scheme. + pub fn is_valid_label_name(self, name: &str) -> bool { + if name.is_empty() { + return false; + } + match self { + Self::Utf8 | Self::Unset => true, + Self::Legacy => is_valid_legacy_label_name(name), + } + } +} + +/// Returns `true` if `name` matches legacy Prometheus label name syntax `[a-zA-Z_][a-zA-Z0-9_]*`. +pub fn is_valid_legacy_label_name(name: &str) -> bool { + let bytes = name.as_bytes(); + let Some(&first) = bytes.first() else { + return false; + }; + if !(first.is_ascii_alphabetic() || first == b'_') { + return false; + } + bytes[1..] + .iter() + .all(|&b| b.is_ascii_alphanumeric() || b == b'_') +} + +/// Returns `true` if `target` matches Prometheus `relabelTargetLegacy` +/// (`^(?:(?:[a-zA-Z_]|\$(?:\{\w+\}|\w+))+\w*)+$`), which permits `$1` / `${var}` expansions +/// within legacy ASCII label names. +pub fn is_valid_legacy_target_label(target: &str) -> bool { + static RE: OnceLock = OnceLock::new(); + let re = RE.get_or_init(|| { + Regex::new(r"^(?:(?:[a-zA-Z_]|\$(?:\{[0-9A-Za-z_]+\}|[0-9A-Za-z_]+))+[0-9A-Za-z_]*)+$") + .expect("valid relabelTargetLegacy regex") + }); + re.is_match(target) +} + +/// Anchored regular expression matching Prometheus `relabel.Regexp` (`^(?s:)$`). +#[derive(Clone, Debug)] +pub struct Regexp { + raw: String, + compiled: Regex, + named_groups: Box<[(Box, usize)]>, + is_implicit_default: bool, +} + +impl PartialEq for Regexp { + fn eq(&self, other: &Self) -> bool { + self.raw == other.raw + } +} + +impl Eq for Regexp {} + +impl Default for Regexp { + fn default() -> Self { + Self::default_regexp() + } +} + +impl Regexp { + /// Compiles a new anchored regular expression `^(?s:)$` matching Prometheus + /// `relabel.NewRegexp` (including Go RE2 ASCII semantics for `\d`, `\D`, `\w`, `\W`, `\s`, + /// `\S`, `\b`, and `\B`). + pub fn new(pattern: &str) -> Result { + let normalized = rewrite_re2_ascii_perl_escapes(pattern); + let anchored = format!("^(?s:{normalized})$"); + let compiled = + Regex::new(&anchored).map_err(|e| RelabelError::InvalidRegex(e.to_string()))?; + let named_groups = compiled + .capture_names() + .enumerate() + .filter_map(|(idx, name)| name.map(|n| (n.into(), idx))) + .collect(); + Ok(Self { + raw: pattern.to_owned(), + compiled, + named_groups, + is_implicit_default: false, + }) + } + + /// Compiles a new anchored regular expression or panics if invalid, matching + /// Prometheus `relabel.MustNewRegexp`. + pub fn must_new(pattern: &str) -> Self { + Self::new(pattern).unwrap_or_else(|e| panic!("invalid relabel regex {pattern:?}: {e}")) + } + + /// Returns the default Prometheus relabel regex (`"(.*)"`). + pub fn default_regexp() -> Self { + let mut r = Self::must_new(DEFAULT_REGEX_PATTERN); + r.is_implicit_default = true; + r + } + + /// Returns an empty anchored regex (`""` -> `^(?s:)$`). + pub fn empty() -> Self { + Self::must_new("") + } + + /// Returns the unanchored source pattern string, matching `relabel.Regexp.String()`. + pub fn as_str(&self) -> &str { + &self.raw + } + + /// Returns `true` if `text` matches the anchored regular expression. + pub fn is_match(&self, text: &str) -> bool { + self.compiled.is_match(text) + } + + /// Returns `true` if this regex has the default `"(.*)"` pattern. + pub fn is_default(&self) -> bool { + self.raw == DEFAULT_REGEX_PATTERN + } + + const fn is_implicit_default(&self) -> bool { + self.is_implicit_default + } +} + +impl Serialize for Regexp { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_str(&self.raw) + } +} + +impl<'de> Deserialize<'de> for Regexp { + fn deserialize>(deserializer: D) -> Result { + let opt = Option::::deserialize(deserializer)?; + let pattern = opt.unwrap_or_default(); + Self::new(&pattern).map_err(serde::de::Error::custom) + } +} + +/// Rewrites Go RE2 ASCII Perl escapes (`\d`, `\D`, `\w`, `\W`, `\s`, `\S`, `\b`, `\B`) so +/// `regex::Regex` keeps `.` and `\p{...}` Unicode-aware while matching ASCII-only Perl classes +/// like Go `regexp/syntax`. +fn rewrite_re2_ascii_perl_escapes(pattern: &str) -> Cow<'_, str> { + if !pattern.contains('\\') { + return Cow::Borrowed(pattern); + } + + let mut out = String::with_capacity(pattern.len() + 8); + let mut chars = pattern.chars().peekable(); + let mut bracket_depth = 0usize; + + while let Some(ch) = chars.next() { + match ch { + '\\' => match chars.next() { + None => out.push('\\'), + Some('Q') => { + let mut quoted = String::new(); + while let Some(qch) = chars.next() { + if qch == '\\' && chars.peek() == Some(&'E') { + chars.next(); + break; + } + quoted.push(qch); + } + out.push_str(®ex::escape("ed)); + } + Some('d') => out.push_str("[0-9]"), + Some('D') => out.push_str("[^0-9]"), + Some('w') => out.push_str("[0-9A-Za-z_]"), + Some('W') => out.push_str("[^0-9A-Za-z_]"), + Some('s') => out.push_str("[\t\n\x0c\r ]"), + Some('S') => out.push_str("[^\t\n\x0c\r ]"), + Some('b') if bracket_depth == 0 => out.push_str(r"(?-u:\b)"), + Some('B') if bracket_depth == 0 => out.push_str(r"(?-u:\B)"), + Some(esc) => { + out.push('\\'); + out.push(esc); + } + }, + '[' => { + out.push('['); + if bracket_depth > 0 && chars.peek() == Some(&':') { + chars.next(); + out.push(':'); + while let Some(pch) = chars.next() { + out.push(pch); + if pch == ':' && chars.peek() == Some(&']') { + chars.next(); + out.push(']'); + break; + } + } + } else { + bracket_depth += 1; + if chars.peek() == Some(&'^') { + chars.next(); + out.push('^'); + } + if chars.peek() == Some(&']') { + chars.next(); + out.push(']'); + } + } + } + ']' if bracket_depth > 0 => { + bracket_depth -= 1; + out.push(']'); + } + other => out.push(other), + } + } + + Cow::Owned(out) +} + +fn default_separator() -> String { + DEFAULT_SEPARATOR.to_owned() +} + +fn default_replacement() -> String { + DEFAULT_REPLACEMENT.to_owned() +} + +const fn is_zero_u64(v: &u64) -> bool { + *v == 0 +} + +/// Configuration for a single relabeling step matching Prometheus `relabel.Config`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Config { + /// List of source label names whose values are concatenated with `separator` in order. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub source_labels: Vec, + /// Separator placed between concatenated source label values (defaults to `";"`). + #[serde(default = "default_separator")] + pub separator: String, + /// Anchored regular expression against which the concatenated value is matched + /// (defaults to `"(.*)"`). + #[serde( + default = "Regexp::default_regexp", + skip_serializing_if = "Regexp::is_implicit_default" + )] + pub regex: Regexp, + /// Modulus to take of the MD5 hash of the concatenated source label values (`hashmod`). + #[serde(default, skip_serializing_if = "is_zero_u64")] + pub modulus: u64, + /// Target label name to which the resulting string is written (`replace`, `hashmod`, + /// `lowercase`, `uppercase`) or compared (`keepequal`, `dropequal`). + #[serde(default, skip_serializing_if = "String::is_empty")] + pub target_label: String, + /// Regex replacement template (defaults to `"$1"`). + #[serde(default = "default_replacement")] + pub replacement: String, + /// Relabeling action to perform (defaults to [`Action::Replace`]). + #[serde(default)] + pub action: Action, + /// Label name validation scheme (not serialized in YAML/JSON, matching Prometheus). + #[serde(skip)] + pub name_validation_scheme: ValidationScheme, +} + +impl Default for Config { + fn default() -> Self { + Self { + source_labels: Vec::new(), + separator: default_separator(), + regex: Regexp::default_regexp(), + modulus: 0, + target_label: String::new(), + replacement: default_replacement(), + action: Action::Replace, + name_validation_scheme: ValidationScheme::Unset, + } + } +} + +impl Config { + /// Validates this relabel configuration according to Prometheus `relabel.Config.Validate`. + pub fn validate(&mut self, default_scheme: ValidationScheme) -> Result<(), RelabelError> { + if self.modulus == 0 && self.action == Action::HashMod { + return Err(RelabelError::InvalidConfig( + "relabel configuration for hashmod requires non-zero modulus".to_owned(), + )); + } + if matches!( + self.action, + Action::Replace + | Action::HashMod + | Action::Lowercase + | Action::Uppercase + | Action::KeepEqual + | Action::DropEqual + ) && self.target_label.is_empty() + { + return Err(RelabelError::InvalidConfig(format!( + "relabel configuration for {} action requires 'target_label' value", + self.action + ))); + } + + if self.name_validation_scheme == ValidationScheme::Unset { + self.name_validation_scheme = match default_scheme { + ValidationScheme::Unset => ValidationScheme::Utf8, + other => other, + }; + } + + let scheme = self.name_validation_scheme; + let is_valid_with_regex_var = |value: &str| match scheme { + ValidationScheme::Utf8 | ValidationScheme::Unset => scheme.is_valid_label_name(value), + ValidationScheme::Legacy => is_valid_legacy_target_label(value), + }; + + if self.action == Action::Replace + && !var_in_regex_template(&self.target_label) + && !scheme.is_valid_label_name(&self.target_label) + { + return Err(RelabelError::InvalidConfig(format!( + "{:?} is invalid 'target_label' for {} action", + self.target_label, self.action + ))); + } + if self.action == Action::Replace + && var_in_regex_template(&self.target_label) + && !is_valid_with_regex_var(&self.target_label) + { + return Err(RelabelError::InvalidConfig(format!( + "{:?} is invalid 'target_label' for {} action", + self.target_label, self.action + ))); + } + if matches!( + self.action, + Action::Lowercase | Action::Uppercase | Action::KeepEqual | Action::DropEqual + ) && !scheme.is_valid_label_name(&self.target_label) + { + return Err(RelabelError::InvalidConfig(format!( + "{:?} is invalid 'target_label' for {} action", + self.target_label, self.action + ))); + } + if matches!( + self.action, + Action::Lowercase | Action::Uppercase | Action::KeepEqual | Action::DropEqual + ) && self.replacement != DEFAULT_REPLACEMENT + { + return Err(RelabelError::InvalidConfig(format!( + "'replacement' can not be set for {} action", + self.action + ))); + } + if self.action == Action::LabelMap && !is_valid_with_regex_var(&self.replacement) { + return Err(RelabelError::InvalidConfig(format!( + "{:?} is invalid 'replacement' for {} action", + self.replacement, self.action + ))); + } + if self.action == Action::HashMod && !scheme.is_valid_label_name(&self.target_label) { + return Err(RelabelError::InvalidConfig(format!( + "{:?} is invalid 'target_label' for {} action", + self.target_label, self.action + ))); + } + + if matches!(self.action, Action::DropEqual | Action::KeepEqual) + && (!self.regex.is_implicit_default + || self.modulus != 0 + || self.separator != DEFAULT_SEPARATOR + || self.replacement != DEFAULT_REPLACEMENT) + { + return Err(RelabelError::InvalidConfig(format!( + "{} action requires only 'source_labels' and `target_label`, and no other fields", + self.action + ))); + } + + if matches!(self.action, Action::LabelDrop | Action::LabelKeep) + && (!self.source_labels.is_empty() + || !self.target_label.is_empty() + || self.modulus != 0 + || self.separator != DEFAULT_SEPARATOR + || self.replacement != DEFAULT_REPLACEMENT) + { + return Err(RelabelError::InvalidConfig(format!( + "{} action requires only 'regex', and no other fields", + self.action + ))); + } + + Ok(()) + } +} + +fn var_in_regex_template(template: &str) -> bool { + template.contains('$') +} + +/// Returns the replacement slice without running regex captures when `cfg.regex` is the default +/// `(.*)` pattern (where `$0` and `$1` always equal `val`) and `cfg.replacement` is either static +/// or a direct `$0`/`$1` capture reference. +fn direct_replacement<'a>(cfg: &'a Config, val: &'a str) -> Option<&'a str> { + if !cfg.regex.is_default() { + return None; + } + if !var_in_regex_template(&cfg.replacement) { + Some(cfg.replacement.as_str()) + } else if matches!(cfg.replacement.as_str(), "$1" | "${1}" | "$0" | "${0}") { + Some(val) + } else { + None + } +} + +/// Applies a slice of relabel [`Config`] rules in order to a cloned [`LabelSet`]. +/// +/// Returns `Some(labels)` if the target/sample is kept, or `None` if a rule drops it. +pub fn process(labels: &LabelSet, cfgs: &[Config]) -> Option { + let mut out = labels.clone(); + if process_in_place(&mut out, cfgs) { + Some(out) + } else { + None + } +} + +/// Applies a slice of relabel [`Config`] rules in-place to `labels`, matching +/// Prometheus `relabel.ProcessBuilder`. +/// +/// Any label whose value is set to `""` is removed from `labels`. +/// Returns `true` if the target/sample is kept, or `false` if dropped. +pub fn process_in_place(labels: &mut LabelSet, cfgs: &[Config]) -> bool { + // Strip any incoming empty-valued labels to match Prometheus `labels.Builder` invariants. + labels.retain(|_, v| !v.is_empty()); + + let mut val_buf = String::new(); + for cfg in cfgs { + if !apply_single(cfg, labels, &mut val_buf) { + return false; + } + } + true +} + +fn apply_single(cfg: &Config, labels: &mut LabelSet, val: &mut String) -> bool { + val.clear(); + for (idx, ln) in cfg.source_labels.iter().enumerate() { + if idx > 0 { + val.push_str(&cfg.separator); + } + if let Some(v) = labels.get(ln.as_str()) { + val.push_str(v); + } + } + + match cfg.action { + Action::Drop => { + if cfg.regex.is_match(val) { + return false; + } + } + Action::Keep => { + if !cfg.regex.is_match(val) { + return false; + } + } + Action::DropEqual => { + let target_val = labels + .get(cfg.target_label.as_str()) + .map_or("", String::as_str); + if target_val == val { + return false; + } + } + Action::KeepEqual => { + let target_val = labels + .get(cfg.target_label.as_str()) + .map_or("", String::as_str); + if target_val != val { + return false; + } + } + Action::Replace => { + if !var_in_regex_template(&cfg.target_label) + && let Some(res) = direct_replacement(cfg, val) + { + if cfg + .name_validation_scheme + .is_valid_label_name(&cfg.target_label) + { + set_or_del(labels, Cow::Borrowed(&cfg.target_label), Cow::Borrowed(res)); + } + return true; + } + + let mut locs = cfg.regex.compiled.capture_locations(); + if cfg.regex.compiled.captures_read(&mut locs, val).is_none() { + return true; + } + let target = expand_template(&cfg.target_label, val, &locs, &cfg.regex.named_groups); + if !cfg.name_validation_scheme.is_valid_label_name(&target) { + return true; + } + let res = expand_template(&cfg.replacement, val, &locs, &cfg.regex.named_groups); + set_or_del(labels, target, res); + } + Action::Lowercase => { + if cfg + .name_validation_scheme + .is_valid_label_name(&cfg.target_label) + { + let lower = go_to_lowercase(val); + set_or_del(labels, Cow::Borrowed(&cfg.target_label), lower); + } + } + Action::Uppercase => { + if cfg + .name_validation_scheme + .is_valid_label_name(&cfg.target_label) + { + let upper = go_to_uppercase(val); + set_or_del(labels, Cow::Borrowed(&cfg.target_label), upper); + } + } + Action::HashMod => { + if cfg.modulus > 0 + && cfg + .name_validation_scheme + .is_valid_label_name(&cfg.target_label) + { + let modulo = hashmod_value(val, cfg.modulus); + set_or_del(labels, Cow::Borrowed(&cfg.target_label), Cow::Owned(modulo)); + } + } + Action::LabelMap => { + let mut locs = cfg.regex.compiled.capture_locations(); + let mut mapped = Vec::new(); + for (name, value) in labels.iter() { + if cfg.regex.compiled.captures_read(&mut locs, name).is_some() { + let res = + expand_template(&cfg.replacement, name, &locs, &cfg.regex.named_groups); + if cfg.name_validation_scheme.is_valid_label_name(&res) { + mapped.push((res.into_owned(), value.clone())); + } + } + } + for (res, value) in mapped { + set_or_del(labels, Cow::Owned(res), Cow::Owned(value)); + } + } + Action::LabelDrop => { + labels.retain(|name, _| !cfg.regex.is_match(name)); + } + Action::LabelKeep => { + labels.retain(|name, _| cfg.regex.is_match(name)); + } + } + + true +} + +fn set_or_del(labels: &mut LabelSet, key: Cow<'_, str>, value: Cow<'_, str>) { + if key.is_empty() { + return; + } + if value.is_empty() { + labels.remove(key.as_ref()); + } else if let Some(existing) = labels.get_mut(key.as_ref()) { + match value { + Cow::Borrowed(s) => { + existing.clear(); + existing.push_str(s); + } + Cow::Owned(s) => *existing = s, + } + } else { + labels.insert(key.into_owned(), value.into_owned()); + } +} + +/// Expands Go `regexp.ExpandString`-compatible `$1`, `${1}`, `$name`, `${name}`, and `$$` +/// references against `locs` captured from `haystack`. +fn expand_template<'a>( + template: &'a str, + haystack: &'a str, + locs: &CaptureLocations, + named_groups: &[(Box, usize)], +) -> Cow<'a, str> { + if !template.contains('$') { + return Cow::Borrowed(template); + } + if matches!(template, "$1" | "${1}" | "$0" | "${0}") { + let idx = usize::from(matches!(template, "$1" | "${1}")); + return Cow::Borrowed(locs.get(idx).map_or("", |(s, e)| &haystack[s..e])); + } + + let mut out = String::with_capacity(template.len()); + let mut rest = template; + + while let Some(dollar_idx) = rest.find('$') { + out.push_str(&rest[..dollar_idx]); + let after_dollar = &rest[dollar_idx + 1..]; + if let Some(stripped) = after_dollar.strip_prefix('$') { + out.push('$'); + rest = stripped; + continue; + } + if let Some((name, num, remaining)) = extract_template_var(after_dollar) { + append_capture(&mut out, name, num, haystack, locs, named_groups); + rest = remaining; + } else { + out.push('$'); + rest = after_dollar; + } + } + out.push_str(rest); + + Cow::Owned(out) +} + +/// Parses a variable reference after `'$'` matching Go `regexp.extract`. +fn extract_template_var(str_after_dollar: &str) -> Option<(&str, Option, &str)> { + if str_after_dollar.is_empty() { + return None; + } + let (brace, body) = if let Some(stripped) = str_after_dollar.strip_prefix('{') { + (true, stripped) + } else { + (false, str_after_dollar) + }; + + let body_bytes = body.as_bytes(); + let mut end = 0; + while end < body_bytes.len() { + let b = body_bytes[end]; + if b.is_ascii_alphanumeric() || b == b'_' { + end += 1; + } else if b >= 0x80 { + // Fall back to Unicode General Categories `L` (`unicode.IsLetter`) and `Nd` + // (`unicode.IsDigit`) so non-decimal numbers such as `'²'` (`No`) terminate `$1²` + // just like Go `regexp.extract`. + if let Some(m) = ident_tail_regex().find(&body[end..]) { + end += m.end(); + } + break; + } else { + break; + } + } + if end == 0 { + return None; + } + + let remaining = if brace { + let after_name = &body[end..]; + if !after_name.starts_with('}') { + return None; + } + &after_name[1..] + } else { + &body[end..] + }; + + let name = &body[..end]; + let name_bytes = name.as_bytes(); + // Go `regexp.extract` rejects leading zeros on multi-digit group numbers and numbers >= 1e8. + let num = if name_bytes.len() > 1 && name_bytes[0] == b'0' { + None + } else { + let mut parsed: usize = 0; + let mut ok = true; + for &b in name_bytes { + if !b.is_ascii_digit() || parsed >= 100_000_000 { + ok = false; + break; + } + parsed = parsed * 10 + usize::from(b - b'0'); + } + ok.then_some(parsed) + }; + + Some((name, num, remaining)) +} + +fn ident_tail_regex() -> &'static Regex { + static IDENT_TAIL_RE: OnceLock = OnceLock::new(); + IDENT_TAIL_RE.get_or_init(|| { + Regex::new(r"^[\p{L}\p{Nd}_]+").expect("valid template var identifier regex") + }) +} + +fn append_capture( + out: &mut String, + name: &str, + num: Option, + haystack: &str, + locs: &CaptureLocations, + named_groups: &[(Box, usize)], +) { + let group_idx = num.or_else(|| { + named_groups + .iter() + .find(|(n, _)| n.as_ref() == name) + .map(|&(_, idx)| idx) + }); + if let Some(idx) = group_idx + && let Some((start, end)) = locs.get(idx) + { + out.push_str(&haystack[start..end]); + } +} + +/// Scrape job defaults used to populate pre-relabel discovered target labels, +/// matching the scrape configuration fields consumed by Prometheus `scrape.PopulateLabels`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TargetPopulateConfig { + /// Scrape `job_name`. + pub job_name: String, + /// Scrape interval duration string (for example `"15s"`). + pub scrape_interval: String, + /// Scrape timeout duration string (for example `"10s"`). + pub scrape_timeout: String, + /// Default metrics path (defaults to `"/metrics"`). + pub metrics_path: String, + /// Default URL scheme (defaults to `"http"`). + pub scheme: String, + /// Optional HTTP query parameters mapped to `__param_` (first value). + pub params: BTreeMap>, +} + +impl Default for TargetPopulateConfig { + fn default() -> Self { + Self { + job_name: String::new(), + scrape_interval: "1m".to_owned(), + scrape_timeout: "10s".to_owned(), + metrics_path: "/metrics".to_owned(), + scheme: "http".to_owned(), + params: BTreeMap::new(), + } + } +} + +/// Result of populating and relabeling a discovered target via [`populate_labels`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PopulatedTarget { + /// Final target labels after `relabel_configs` and `__meta_*` removal (includes `job`, + /// `instance`, `__address__`, `__scheme__`, `__metrics_path__`, and any `__param_*` labels). + pub labels: LabelSet, + /// Pre-relabel discovered labels produced by [`populate_discovered_labels`]. + pub discovered_labels: LabelSet, +} + +impl PopulatedTarget { + /// Returns the public (non-`__`-prefixed) target labels emitted onto scraped samples or + /// resource attributes (`job`, `instance`, and user labels). + pub fn public_labels(&self) -> LabelSet { + self.labels + .iter() + .filter(|(k, _)| !k.starts_with(RESERVED_LABEL_PREFIX)) + .map(|(k, v)| (k.clone(), v.clone())) + .collect() + } +} + +/// Builds the pre-relabel discovered label set from a target label set, common target-group +/// label set, and [`TargetPopulateConfig`], matching Prometheus `scrape.PopulateDiscoveredLabels`. +pub fn populate_discovered_labels( + target: &LabelSet, + group_labels: &LabelSet, + cfg: &TargetPopulateConfig, +) -> LabelSet { + let mut out = LabelSet::new(); + for (k, v) in target { + if !v.is_empty() { + out.insert(k.clone(), v.clone()); + } + } + for (k, v) in group_labels { + if !v.is_empty() && !target.contains_key(k) { + out.insert(k.clone(), v.clone()); + } + } + + let defaults = [ + (JOB_LABEL, cfg.job_name.as_str()), + (SCRAPE_INTERVAL_LABEL, cfg.scrape_interval.as_str()), + (SCRAPE_TIMEOUT_LABEL, cfg.scrape_timeout.as_str()), + (METRICS_PATH_LABEL, cfg.metrics_path.as_str()), + (SCHEME_LABEL, cfg.scheme.as_str()), + ]; + for (k, v) in defaults { + if !v.is_empty() && !out.contains_key(k) { + out.insert(k.to_owned(), v.to_owned()); + } + } + + // Encode scrape query parameters into `__param_` labels using the first value. + for (k, vals) in &cfg.params { + if let Some(first) = vals.first() + && !first.is_empty() + { + let param_key = format!("{PARAM_LABEL_PREFIX}{k}"); + out.entry(param_key).or_insert_with(|| first.clone()); + } + } + + out +} + +/// Populates discovered labels, applies `relabel_configs`, validates `__address__`, +/// `__scrape_interval__`, and `__scrape_timeout__`, strips `__meta_*` labels, and sets the default +/// `instance` label, matching Prometheus `scrape.PopulateLabels`. +/// +/// Returns `Ok(Some(PopulatedTarget))` if the target is kept, `Ok(None)` if dropped by +/// `relabel_configs`, or `Err(RelabelError::TargetError)` if `__address__`, `__scrape_interval__`, +/// or `__scrape_timeout__` is missing or invalid. +pub fn populate_labels( + target: &LabelSet, + group_labels: &LabelSet, + cfg: &TargetPopulateConfig, + relabel_configs: &[Config], +) -> Result, RelabelError> { + let discovered_labels = populate_discovered_labels(target, group_labels, cfg); + let mut labels = discovered_labels.clone(); + if !process_in_place(&mut labels, relabel_configs) { + return Ok(None); + } + + let Some(addr) = labels.get(ADDRESS_LABEL).cloned() else { + return Err(RelabelError::TargetError("no address".to_owned())); + }; + if addr.is_empty() { + return Err(RelabelError::TargetError("no address".to_owned())); + } + if addr.contains('/') { + return Err(RelabelError::TargetError(format!( + "{addr:?} is not a valid hostname" + ))); + } + + let interval_str = labels.get(SCRAPE_INTERVAL_LABEL).map_or("", String::as_str); + let interval_ns = parse_prometheus_duration(interval_str) + .map_err(|e| RelabelError::TargetError(format!("error parsing scrape interval: {e}")))?; + if interval_ns == 0 { + return Err(RelabelError::TargetError( + "scrape interval cannot be 0".to_owned(), + )); + } + + let timeout_str = labels.get(SCRAPE_TIMEOUT_LABEL).map_or("", String::as_str); + let timeout_ns = parse_prometheus_duration(timeout_str) + .map_err(|e| RelabelError::TargetError(format!("error parsing scrape timeout: {e}")))?; + if timeout_ns == 0 { + return Err(RelabelError::TargetError( + "scrape timeout cannot be 0".to_owned(), + )); + } + + if timeout_ns > interval_ns { + return Err(RelabelError::TargetError(format!( + "scrape timeout cannot be greater than scrape interval ({timeout_str:?} > {interval_str:?})" + ))); + } + + for (label_name, desc) in [ + ( + "__convert_classic_histograms_to_nhcb__", + "convert classic histograms to nhcb", + ), + ( + "__always_scrape_classic_histograms__", + "always scrape classic histograms", + ), + ("__scrape_native_histograms__", "scrape native histograms"), + ] { + if let Some(val) = labels.get(label_name) + && !val.is_empty() + && !is_valid_go_bool(val) + { + return Err(RelabelError::TargetError(format!( + "error parsing {desc}: invalid syntax {val:?}" + ))); + } + } + + // Remove all `__meta_*` labels after target relabeling. + labels.retain(|k, _| !k.starts_with(META_LABEL_PREFIX)); + + // Default `instance` to `__address__` if not set by relabeling. + if !labels.contains_key(INSTANCE_LABEL) { + labels.insert(INSTANCE_LABEL.to_owned(), addr); + } + + Ok(Some(PopulatedTarget { + labels, + discovered_labels, + })) +} + +/// Parses a non-negative Prometheus duration string into nanoseconds matching +/// `github.com/prometheus/common/model.ParseDuration` (`y`, `w`, `d`, `h`, `m`, `s`, `ms` in +/// strictly decreasing unit order, plus `"0"`). +fn parse_prometheus_duration(s: &str) -> Result { + if s == "0" { + return Ok(0); + } + if s.is_empty() { + return Err("empty duration string".to_owned()); + } + + const MS_NS: u64 = 1_000_000; + const SEC_NS: u64 = 1_000 * MS_NS; + const MIN_NS: u64 = 60 * SEC_NS; + const HOUR_NS: u64 = 60 * MIN_NS; + const DAY_NS: u64 = 24 * HOUR_NS; + const WEEK_NS: u64 = 7 * DAY_NS; + const YEAR_NS: u64 = 365 * DAY_NS; + const MAX_DURATION_NS: u64 = i64::MAX as u64; + + let orig = s; + let mut rest = s; + let mut total_ns = 0u64; + let mut last_unit_pos = 0u8; + + while !rest.is_empty() { + let digit_len = rest.bytes().take_while(|b| b.is_ascii_digit()).count(); + if digit_len == 0 { + return Err(format!("not a valid duration string: {orig:?}")); + } + let val: u64 = rest[..digit_len] + .parse() + .map_err(|_| format!("not a valid duration string: {orig:?}"))?; + rest = &rest[digit_len..]; + + let unit_len = rest.bytes().take_while(|b| !b.is_ascii_digit()).count(); + if unit_len == 0 { + return Err(format!("not a valid duration string: {orig:?}")); + } + let unit = &rest[..unit_len]; + rest = &rest[unit_len..]; + + let (pos, mult) = match unit { + "y" => (1u8, YEAR_NS), + "w" => (2, WEEK_NS), + "d" => (3, DAY_NS), + "h" => (4, HOUR_NS), + "m" => (5, MIN_NS), + "s" => (6, SEC_NS), + "ms" => (7, MS_NS), + _ => return Err(format!("unknown unit {unit:?} in duration {orig:?}")), + }; + if pos <= last_unit_pos { + return Err(format!("not a valid duration string: {orig:?}")); + } + last_unit_pos = pos; + + if val > (1u64 << 63) / mult { + return Err("duration out of range".to_owned()); + } + total_ns = total_ns + .checked_add(val * mult) + .filter(|&d| d <= MAX_DURATION_NS) + .ok_or_else(|| "duration out of range".to_owned())?; + } + + Ok(total_ns) +} + +fn is_valid_go_bool(s: &str) -> bool { + matches!( + s, + "1" | "t" | "T" | "true" | "TRUE" | "True" | "0" | "f" | "F" | "false" | "FALSE" | "False" + ) +} + +/// Merges non-reserved target labels onto a scraped sample's `metric_labels` according to +/// Prometheus `honor_labels` semantics (`scrape/scrape.go`): +/// +/// - Internal target labels starting with `"__"` are never copied onto samples. +/// - If `honor_labels` is `true`, existing sample labels take precedence over target labels. +/// - If `honor_labels` is `false`, conflicting sample labels are renamed to +/// `"exported_