feat(metrics): config-driven bounded request labels (adds service to oneshot_pipeline.duration)

apps/skit/src/config.rs

@@ -231,6 +231,145 @@ impl Default for CorsConfig {
     }
 }
 
+fn default_label_fallback() -> String {
+    "other".to_string()
+}
+
+/// A bounded metric label sourced from a trusted request header.
+///
+/// The header value is trimmed and lowercased, then matched against `allowed`;
+/// anything not in the allowlist (or a missing header) collapses to `fallback`,
+/// so client-supplied headers can never inflate metric cardinality.
+#[derive(Deserialize, Serialize, Debug, Clone, JsonSchema)]
+pub struct RequestLabelConfig {
+    /// Metric label key (e.g. `service`).
+    pub name: String,
+    /// Trusted request header to read the value from (e.g. `X-StreamKit-Service`).
+    ///
+    /// Read before auth middleware runs, so point this at a header set by a
+    /// trusted upstream (e.g. the gateway, which strips client-supplied copies)
+    /// — not at an auth-injected header such as `X-StreamKit-Role`, whose
+    /// pre-auth value is client-controlled.
+    pub header: String,
+    /// Permitted values, matched case-insensitively after trimming.
+    #[serde(default)]
+    pub allowed: Vec<String>,
+    /// Value emitted when the header is absent or its value is not in `allowed`.
+    #[serde(default = "default_label_fallback")]
+    pub fallback: String,
+}
+
+/// Configuration for request-scoped metric labeling.
+///
+/// Empty by default: no request metric gains a configured label unless an
+/// operator opts in. Declaring `request_labels` sets the full list (figment
+/// does not merge sequences). See the commented example in `samples/skit.toml`.
+#[derive(Deserialize, Serialize, Debug, Clone, Default, JsonSchema)]
+pub struct MetricsConfig {
+    /// Bounded labels attached to request metrics, each sourced from a trusted
+    /// request header. Applied to all HTTP request metrics and to oneshot
+    /// pipeline metrics.
+    #[serde(default)]
+    pub request_labels: Vec<RequestLabelConfig>,
+}
+
+/// Prometheus sanitizes any character outside `[a-zA-Z0-9_]` in a label key to
+/// `_`, so `http.method` and `http_method` collapse to the same series key. We
+/// compare sanitized keys to catch collisions that only appear after scrape.
+fn sanitize_label_key(name: &str) -> String {
+    name.chars().map(|c| if c.is_ascii_alphanumeric() || c == '_' { c } else { '_' }).collect()
+}
+
+/// A metric label name must be a non-empty identifier (dots allowed, per the
+/// OpenTelemetry convention used by the built-in keys) so it survives export.
+fn is_valid_label_name(name: &str) -> bool {
+    let mut chars = name.chars();
+    match chars.next() {
+        Some(c) if c.is_ascii_alphabetic() || c == '_' => {},
+        _ => return false,
+    }
+    chars.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '.')
+}
+
+impl MetricsConfig {
+    /// Normalize then validate — the single chokepoint that readies a metrics
+    /// config for use. Callers decide how to treat the error: `load()` rejects
+    /// the file, `create_app_state()` warns and disables the labels.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if validation fails after normalization.
+    pub fn prepare(&mut self) -> Result<(), String> {
+        self.normalize();
+        self.validate()
+    }
+
+    /// Lowercase and trim every allowlist entry and each `fallback` so the
+    /// per-request hot path only has to normalize the incoming header value and
+    /// every emitted value shares one normalized space.
+    pub fn normalize(&mut self) {
+        for label in &mut self.request_labels {
+            for allowed in &mut label.allowed {
+                *allowed = crate::metrics_labels::normalize(allowed);
+            }
+            label.fallback = crate::metrics_labels::normalize(&label.fallback);
+        }
+    }
+
+    /// Reject label configs that would silently corrupt metrics: invalid names,
+    /// names colliding (after Prometheus sanitization) with a built-in key or
+    /// each other, invalid/empty headers, and empty allowlist or fallback values.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error describing the first offending label.
+    pub fn validate(&self) -> Result<(), String> {
+        let reserved: std::collections::HashSet<String> =
+            crate::metrics_labels::RESERVED_LABEL_KEYS
+                .iter()
+                .map(|k| sanitize_label_key(k))
+                .collect();
+        let mut seen = std::collections::HashSet::new();
+        for label in &self.request_labels {
+            if !is_valid_label_name(&label.name) {
+                return Err(format!(
+                    "metrics request_label name '{}' is not a valid metric label name",
+                    label.name
+                ));
+            }
+            let key = sanitize_label_key(&label.name);
+            if reserved.contains(&key) {
+                return Err(format!(
+                    "metrics request_label name '{}' collides with a built-in metric key",
+                    label.name
+                ));
+            }
+            if !seen.insert(key) {
+                return Err(format!("duplicate metrics request_label name '{}'", label.name));
+            }
+            if axum::http::HeaderName::try_from(label.header.as_str()).is_err() {
+                return Err(format!(
+                    "metrics request_label '{}' has an invalid header '{}'",
+                    label.name, label.header
+                ));
+            }
+            if label.allowed.iter().any(|v| v.trim().is_empty()) {
+                return Err(format!(
+                    "metrics request_label '{}' has an empty allowed value",
+                    label.name
+                ));
+            }
+            if label.fallback.trim().is_empty() {
+                return Err(format!(
+                    "metrics request_label '{}' has an empty fallback value",
+                    label.name
+                ));
+            }
+        }
+        Ok(())
+    }
+}
+
 /// Telemetry and observability configuration (OpenTelemetry, tokio-console).
 #[derive(Deserialize, Serialize, Debug, Clone, JsonSchema)]
 pub struct TelemetryConfig {
@@ -322,6 +461,9 @@ pub struct ServerConfig {
     /// CORS configuration for cross-origin requests
     #[serde(default)]
     pub cors: CorsConfig,
+    /// Bounded request-metric labeling configuration.
+    #[serde(default)]
+    pub metrics: MetricsConfig,
     #[cfg(feature = "moq")]
     pub moq_address: Option<String>,
     /// TLS certificate for the MoQ WebTransport listener.
@@ -350,6 +492,7 @@ impl Default for ServerConfig {
             max_body_size: default_max_body_size(),
             base_path: None,
             cors: CorsConfig::default(),
+            metrics: MetricsConfig::default(),
             #[cfg(feature = "moq")]
             moq_address: None,
             #[cfg(feature = "moq")]
@@ -1035,6 +1178,9 @@ pub fn load(config_path: &str) -> Result<ConfigLoadResult, Box<figment::Error>>
     if let Err(e) = config.mcp.validate() {
         return Err(Box::new(figment::Error::from(e)));
     }
+    if let Err(e) = config.server.metrics.prepare() {
+        return Err(Box::new(figment::Error::from(e)));
+    }
 
     Ok(ConfigLoadResult { config, file_missing })
 }
@@ -1445,4 +1591,156 @@ allowed_plugins = []
             Ok(())
         });
     }
+
+    fn request_label(name: &str) -> RequestLabelConfig {
+        RequestLabelConfig {
+            name: name.to_string(),
+            header: "X-Test".to_string(),
+            allowed: vec![],
+            fallback: "other".to_string(),
+        }
+    }
+
+    #[test]
+    fn metrics_validate_rejects_reserved_label_name() {
+        let metrics = MetricsConfig { request_labels: vec![request_label("status")] };
+        assert!(metrics.validate().is_err());
+    }
+
+    #[test]
+    fn metrics_validate_rejects_duplicate_label_name() {
+        let metrics = MetricsConfig {
+            request_labels: vec![request_label("service"), request_label("service")],
+        };
+        assert!(metrics.validate().is_err());
+    }
+
+    #[test]
+    fn metrics_validate_accepts_default() {
+        assert!(MetricsConfig::default().validate().is_ok());
+    }
+
+    #[test]
+    fn metrics_default_is_empty_opt_in() {
+        assert!(MetricsConfig::default().request_labels.is_empty());
+    }
+
+    #[test]
+    fn metrics_validate_rejects_empty_or_invalid_label_name() {
+        assert!(MetricsConfig { request_labels: vec![request_label("")] }.validate().is_err());
+        assert!(MetricsConfig { request_labels: vec![request_label("   ")] }.validate().is_err());
+        assert!(MetricsConfig { request_labels: vec![request_label("1service")] }
+            .validate()
+            .is_err());
+    }
+
+    #[test]
+    fn metrics_validate_rejects_sanitized_reserved_collision() {
+        // `http_method` sanitizes to the same Prometheus key as `http.method`.
+        let metrics = MetricsConfig { request_labels: vec![request_label("http_method")] };
+        assert!(metrics.validate().is_err());
+    }
+
+    #[test]
+    fn metrics_validate_rejects_empty_allowed_value() {
+        let metrics = MetricsConfig {
+            request_labels: vec![RequestLabelConfig {
+                name: "service".to_string(),
+                header: "X-Test".to_string(),
+                allowed: vec!["tts".to_string(), " ".to_string()],
+                fallback: "other".to_string(),
+            }],
+        };
+        assert!(metrics.validate().is_err());
+    }
+
+    #[test]
+    fn metrics_validate_rejects_empty_or_invalid_header() {
+        let empty = RequestLabelConfig {
+            name: "service".to_string(),
+            header: String::new(),
+            allowed: vec!["tts".to_string()],
+            fallback: "other".to_string(),
+        };
+        assert!(MetricsConfig { request_labels: vec![empty] }.validate().is_err());
+        let bad = RequestLabelConfig {
+            name: "service".to_string(),
+            header: "bad header".to_string(),
+            allowed: vec!["tts".to_string()],
+            fallback: "other".to_string(),
+        };
+        assert!(MetricsConfig { request_labels: vec![bad] }.validate().is_err());
+    }
+
+    #[test]
+    fn metrics_validate_rejects_empty_fallback() {
+        let metrics = MetricsConfig {
+            request_labels: vec![RequestLabelConfig {
+                name: "service".to_string(),
+                header: "X-Test".to_string(),
+                allowed: vec!["tts".to_string()],
+                fallback: "  ".to_string(),
+            }],
+        };
+        assert!(metrics.validate().is_err());
+    }
+
+    #[test]
+    fn metrics_prepare_normalizes_then_validates() {
+        let mut metrics = MetricsConfig {
+            request_labels: vec![RequestLabelConfig {
+                name: "service".to_string(),
+                header: "X-StreamKit-Service".to_string(),
+                allowed: vec!["  TTS ".to_string()],
+                fallback: "Other".to_string(),
+            }],
+        };
+        assert!(metrics.prepare().is_ok());
+        assert_eq!(metrics.request_labels[0].allowed, vec!["tts".to_string()]);
+        assert_eq!(metrics.request_labels[0].fallback, "other");
+    }
+
+    #[test]
+    fn metrics_normalize_lowercases_fallback() {
+        let mut metrics = MetricsConfig {
+            request_labels: vec![RequestLabelConfig {
+                name: "service".to_string(),
+                header: "X-Test".to_string(),
+                allowed: vec!["tts".to_string()],
+                fallback: " Other ".to_string(),
+            }],
+        };
+        metrics.normalize();
+        assert_eq!(metrics.request_labels[0].fallback, "other");
+    }
+
+    #[test]
+    fn metrics_normalize_lowercases_allowlist() {
+        let mut metrics = MetricsConfig {
+            request_labels: vec![RequestLabelConfig {
+                name: "service".to_string(),
+                header: "X-StreamKit-Service".to_string(),
+                allowed: vec!["  TTS ".to_string(), "Stt".to_string()],
+                fallback: "other".to_string(),
+            }],
+        };
+        metrics.normalize();
+        assert_eq!(metrics.request_labels[0].allowed, vec!["tts".to_string(), "stt".to_string()]);
+    }
+
+    #[test]
+    fn load_rejects_reserved_metrics_label_name() {
+        figment::Jail::expect_with(|jail| {
+            jail.create_file(
+                "skit.toml",
+                r#"[[server.metrics.request_labels]]
+name = "http.route"
+header = "X-StreamKit-Service"
+allowed = ["tts"]
+"#,
+            )?;
+            assert!(load("skit.toml").is_err(), "reserved label name must fail load");
+            Ok(())
+        });
+    }
 }

apps/skit/src/lib.rs

@@ -14,6 +14,7 @@ pub mod marketplace_installer;
 pub mod marketplace_security;
 #[cfg(feature = "mcp")]
 pub mod mcp;
+pub mod metrics_labels;
 #[cfg(feature = "moq")]
 pub mod moq_gateway;
 pub mod mse_gateway;

apps/skit/src/main.rs

@@ -35,6 +35,7 @@ mod marketplace_installer;
 mod marketplace_security;
 #[cfg(feature = "mcp")]
 mod mcp;
+mod metrics_labels;
 #[cfg(feature = "moq")]
 mod moq_gateway;
 mod mse_gateway;