fix(mimir): apply custom labels to all series and drop reserved-name collisions

Custom per-target labels were moved into the probe_success series and so
never reached any other metric. Build every series through a shared merge
step so user labels apply to all of them, with phase-duration series adding
their phase label on top.

Drop custom labels that reuse a reserved name (__name__, instance, job,
module, target, phase); a duplicate label name is rejected by Mimir.

Also simplify handle_target_probe (drop the redundant labels argument) and
use saturating_sub for the poll interval to satisfy clippy.

Author: tluijken

oxybox/src/http_probe/probe.rs

@@ -322,7 +322,6 @@ pub async fn run_probe_loop(
             let target = target.clone();
             let tenant_name = tenant_name.clone();
             let org_id = org_config.organisation_id.clone();
-            let labels = target.labels.clone();
             let mimir_endpoint = mimir_endpoint.clone();
 
             handles.push(tokio::spawn(async move {
@@ -340,7 +339,6 @@ pub async fn run_probe_loop(
                         &resolver,
                         &mimir_endpoint,
                         max_org_width,
-                        labels,
                     )
                     .await
                 })
@@ -359,10 +357,7 @@ pub async fn run_probe_loop(
         }
 
         let elapsed = start_time.elapsed().as_secs();
-        let wait = org_config
-            .polling_interval_seconds
-            .checked_sub(elapsed)
-            .unwrap_or(0);
+        let wait = org_config.polling_interval_seconds.saturating_sub(elapsed);
 
         sleep(Duration::from_secs(wait)).await;
     }
@@ -388,7 +383,6 @@ fn to_fixed_width(input: &str, width: usize) -> String {
 ///     * `resolver` - The DNS resolver for resolving hostnames.
 ///     * `mimir_target` - The Mimir endpoint to send metrics to.
 ///     * `max_width` - The maximum width for tenant name formatting in logs.
-///     * `labels` - Optional additional labels to include in the metrics.
 async fn handle_target_probe(
     tenant: String,
     org_id: &str,
@@ -397,7 +391,6 @@ async fn handle_target_probe(
     resolver: &TokioAsyncResolver,
     mimir_target: &str,
     max_width: usize,
-    labels: Option<Vec<(String, String)>>,
 ) {
     let url = &target.url;
     let result = probe_url(tls_connector, resolver, url).await;
@@ -408,7 +401,7 @@ async fn handle_target_probe(
         .as_secs_f64();
     let padded_tenant = to_fixed_width(&tenant, max_width);
 
-    let labels = labels.as_ref().map(|l| {
+    let labels = target.labels.as_ref().map(|l| {
         l.iter()
             .map(|(k, v)| (k.as_str(), v.as_str()))
             .collect::<Vec<(&str, &str)>>()

oxybox/src/mimir/client.rs

@@ -127,38 +127,31 @@ mod tests {
         let mimir_url = "http://localhost:9009"; // Adjust to your Mimir instance
         let tenant_id = Some("demo"); // Optional, remove if Mimir is single-tenant
 
-        let mut metrics_to_send = Vec::new();
-
-        // Metric 1: A counter for requests
-        metrics_to_send.push(create_time_series(
-            "my_app_http_requests_total",
-            &[("method", "GET"), ("status", "200")],
-            1.0,  // For a counter, typically increment by 1 per event
-            None, // Use current timestamp
-        ));
-
-        // Metric 2: A gauge for CPU usage
-        metrics_to_send.push(create_time_series(
-            "my_app_cpu_usage_percent",
-            &[("host", "server-a")],
-            25.5,                                // Current value for a gauge
-            Some(Utc::now().timestamp_millis()), // Specific timestamp
-        ));
-
-        // Metric 3: Another counter with different labels
-        metrics_to_send.push(create_time_series(
-            "my_app_database_queries_total",
-            &[("db", "users"), ("type", "read")],
-            1.0,
-            None,
-        ));
-
-        // Attempt to send
+        let metrics_to_send = vec![
+            create_time_series(
+                "my_app_http_requests_total",
+                &[("method", "GET"), ("status", "200")],
+                1.0,
+                None,
+            ),
+            create_time_series(
+                "my_app_cpu_usage_percent",
+                &[("host", "server-a")],
+                25.5,
+                Some(Utc::now().timestamp_millis()),
+            ),
+            create_time_series(
+                "my_app_database_queries_total",
+                &[("db", "users"), ("type", "read")],
+                1.0,
+                None,
+            ),
+        ];
+
+        // Attempt to send; a missing local Mimir is tolerated so this stays a smoke test.
         match send_to_mimir(mimir_url, tenant_id, metrics_to_send).await {
             Ok(_) => log::debug!("Test metrics sent successfully."),
             Err(e) => log::debug!("Failed to send test metrics: {}", e),
         }
-
-        assert!(true);
     }
 }

oxybox/src/mimir/mod.rs

@@ -4,10 +4,23 @@ use crate::http_probe::result::ProbeResult;
 
 pub mod client;
 
+const NAME_LABEL: &str = "__name__";
 const INSTANCE_LABEL: &str = "instance";
 const JOB_LABEL: &str = "job";
 const MODULE_LABEL: &str = "module";
 const TARGET_LABEL: &str = "target";
+const PHASE_LABEL: &str = "phase";
+
+/// Label names owned by Oxybox. A user-supplied label reusing one of these would
+/// produce a duplicate label name, which Mimir rejects, so such labels are dropped.
+const RESERVED_LABELS: [&str; 6] = [
+    NAME_LABEL,
+    INSTANCE_LABEL,
+    JOB_LABEL,
+    MODULE_LABEL,
+    TARGET_LABEL,
+    PHASE_LABEL,
+];
 const PROBE_SUCCESS_METRIC: &str = "probe_success";
 const PROBE_DURATION_METRIC: &str = "probe_duration_seconds";
 const PROBE_HTTP_STATUS_METRIC: &str = "probe_http_status_code";
@@ -62,18 +75,30 @@ pub fn create_probe_metrics(
     probe_success: bool,
     labels: Option<Vec<(&str, &str)>>,
 ) -> Vec<prompb::TimeSeries> {
-    let mut metrics = Vec::new();
-    let probe_successful = match probe_success {
-        true => 1.0,
-        false => 0.0,
+    let custom: Vec<(&str, &str)> = labels
+        .unwrap_or_default()
+        .into_iter()
+        .filter(|(key, _)| {
+            let reserved = RESERVED_LABELS.contains(key);
+            if reserved {
+                log::warn!(
+                    "Ignoring custom label '{key}' for {}: it collides with a reserved label",
+                    probe_result.url
+                );
+            }
+            !reserved
+        })
+        .collect();
+    let series = |metric: &str, value: f64, phase: Option<&str>| {
+        let mut extra = custom.clone();
+        if let Some(phase) = phase {
+            extra.push((PHASE_LABEL, phase));
+        }
+        create_time_series(metric, &probe_result.url, value, Some(extra))
     };
 
-    metrics.push(create_time_series(
-        PROBE_SUCCESS_METRIC,
-        &probe_result.url,
-        probe_successful,
-        labels,
-    ));
+    let probe_successful = if probe_success { 1.0 } else { 0.0 };
+    let mut metrics = vec![series(PROBE_SUCCESS_METRIC, probe_successful, None)];
 
     let phases = [
         (probe_result.dns_time, "resolve"),
@@ -82,77 +107,153 @@ pub fn create_probe_metrics(
         (probe_result.processing_time, "processing"),
         (probe_result.transfer_time, "transfer"),
     ];
-
     for (duration_opt, phase) in phases.iter() {
         if let Some(duration) = duration_opt {
-            metrics.push(create_time_series(
-                PROBE_HTTP_DURATION_METRIC,
-                &probe_result.url,
-                *duration,
-                Some(vec![("phase", *phase)]),
-            ));
+            metrics.push(series(PROBE_HTTP_DURATION_METRIC, *duration, Some(*phase)));
         }
     }
 
-    metrics.push(create_time_series(
+    metrics.push(series(
         PROBE_DURATION_METRIC,
-        &probe_result.url,
         probe_result.total_probe_time,
         None,
     ));
+
     if let Some(http_status) = probe_result.http_status {
-        metrics.push(create_time_series(
-            PROBE_HTTP_STATUS_METRIC,
-            &probe_result.url,
-            http_status as f64,
-            None,
-        ));
+        metrics.push(series(PROBE_HTTP_STATUS_METRIC, http_status as f64, None));
     }
 
     if let Some(dns_time) = probe_result.dns_time {
-        metrics.push(create_time_series(
-            PROBE_DNS_LOOKUP_TIME_METRIC,
-            &probe_result.url,
-            dns_time,
-            None,
-        ));
+        metrics.push(series(PROBE_DNS_LOOKUP_TIME_METRIC, dns_time, None));
     }
 
-    let ssl_enabled = match probe_result.cert_validity_seconds {
-        Some(_) => 1.0,
-        None => 0.0,
+    let ssl_enabled = if probe_result.cert_validity_seconds.is_some() {
+        1.0
+    } else {
+        0.0
     };
-
-    metrics.push(create_time_series(
-        PROBE_HTTP_SSL_ENABLED_METRIC,
-        &probe_result.url,
-        ssl_enabled,
-        None,
-    ));
+    metrics.push(series(PROBE_HTTP_SSL_ENABLED_METRIC, ssl_enabled, None));
 
     if let Some(cert_validity_seconds) = probe_result.cert_validity_seconds {
-        metrics.push(create_time_series(
+        metrics.push(series(
             PROBE_HTTP_SSL_EARLIEST_EXPIERY_METRIC,
-            &probe_result.url,
             cert_validity_seconds,
             None,
         ));
     }
+
     if let Some(http_version) = probe_result.http_version {
-        metrics.push(create_time_series(
-            PROBE_HTTP_VERSION_METRIC,
-            &probe_result.url,
-            http_version,
-            None,
-        ));
+        metrics.push(series(PROBE_HTTP_VERSION_METRIC, http_version, None));
     }
 
-    metrics.push(create_time_series(
+    metrics.push(series(
         PROBE_HTTP_REDIRECTS_METRIC,
-        &probe_result.url,
         probe_result.redirects as f64,
         None,
     ));
 
     metrics
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::http_probe::result::ProbeResult;
+
+    fn sample_probe() -> ProbeResult {
+        ProbeResult {
+            url: "https://example.com".to_string(),
+            dns_time: Some(0.01),
+            connect_time: Some(0.02),
+            tls_time: Some(0.03),
+            processing_time: Some(0.04),
+            transfer_time: Some(0.05),
+            cert_validity_seconds: Some(1_700_000_000.0),
+            http_status: Some(200),
+            http_version: Some(2.0),
+            total_probe_time: 0.15,
+            redirects: 1,
+        }
+    }
+
+    fn metric_name(ts: &prompb::TimeSeries) -> String {
+        ts.labels
+            .iter()
+            .find(|l| l.name == "__name__")
+            .map(|l| l.value.clone())
+            .unwrap_or_default()
+    }
+
+    #[test]
+    fn custom_labels_apply_to_every_series() {
+        let labels = Some(vec![("environment", "production"), ("region", "eu")]);
+        let metrics = create_probe_metrics(&sample_probe(), true, labels);
+
+        assert!(metrics.len() > 1);
+        for ts in &metrics {
+            for (key, value) in [("environment", "production"), ("region", "eu")] {
+                assert!(
+                    ts.labels.iter().any(|l| l.name == key && l.value == value),
+                    "series `{}` is missing custom label {key}={value}",
+                    metric_name(ts),
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn phase_series_keep_their_phase_label_alongside_custom_labels() {
+        let labels = Some(vec![("environment", "production")]);
+        let metrics = create_probe_metrics(&sample_probe(), true, labels);
+
+        let phase_series: Vec<_> = metrics
+            .iter()
+            .filter(|ts| metric_name(ts) == PROBE_HTTP_DURATION_METRIC)
+            .collect();
+        assert_eq!(phase_series.len(), 5);
+        for ts in phase_series {
+            assert!(ts.labels.iter().any(|l| l.name == "phase"));
+            assert!(
+                ts.labels
+                    .iter()
+                    .any(|l| l.name == "environment" && l.value == "production")
+            );
+        }
+    }
+
+    #[test]
+    fn reserved_custom_labels_are_dropped() {
+        let labels = Some(vec![
+            (INSTANCE_LABEL, "evil"),
+            (PHASE_LABEL, "nope"),
+            ("environment", "production"),
+        ]);
+        let metrics = create_probe_metrics(&sample_probe(), true, labels);
+
+        assert!(metrics.len() > 1);
+        for ts in &metrics {
+            assert_eq!(
+                ts.labels.iter().filter(|l| l.name == INSTANCE_LABEL).count(),
+                1,
+                "series `{}` has a duplicate instance label",
+                metric_name(ts),
+            );
+            assert!(ts.labels.iter().all(|l| l.value != "evil"));
+            assert!(ts.labels.iter().all(|l| l.value != "nope"));
+            assert!(
+                ts.labels
+                    .iter()
+                    .any(|l| l.name == "environment" && l.value == "production")
+            );
+        }
+    }
+
+    #[test]
+    fn no_custom_labels_still_produces_valid_series() {
+        let metrics = create_probe_metrics(&sample_probe(), true, None);
+        assert!(metrics.iter().any(|ts| metric_name(ts) == PROBE_SUCCESS_METRIC));
+        for ts in &metrics {
+            assert!(ts.labels.iter().any(|l| l.name == INSTANCE_LABEL));
+        }
+    }
+}