feat(31): wire invariant recorder to DogStatsD metrics

Emit cloudprem.pomsky.invariant.checked and .violated counters with
invariant label via the metrics crate / DogStatsD exporter at process
startup, completing Layer 4 of the verification stack.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: g-talbot

quickwit/Cargo.lock

@@ -156,6 +156,8 @@ libz-sys = "1.1"
 lru = "0.16"
 matches = "0.1"
 md5 = "0.8"
+metrics = "0.24"
+metrics-exporter-dogstatsd = "0.9"
 mime_guess = "2.0"
 mini-moka = "0.10.3"
 mockall = "0.14"

quickwit/Cargo.toml

@@ -56,10 +56,14 @@ tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 tracing-subscriber = { workspace = true }
 
+metrics = { workspace = true }
+metrics-exporter-dogstatsd = { workspace = true }
+
 quickwit-actors = { workspace = true }
 quickwit-cluster = { workspace = true }
 quickwit-common = { workspace = true }
 quickwit-config = { workspace = true }
+quickwit-dst = { workspace = true }
 quickwit-index-management = { workspace = true }
 quickwit-indexing = { workspace = true }
 quickwit-ingest = { workspace = true }

quickwit/quickwit-cli/Cargo.toml

@@ -161,6 +161,65 @@ pub fn setup_logging_and_tracing(
     ))
 }
 
+#[cfg(not(any(test, feature = "testsuite")))]
+pub fn setup_dogstatsd_exporter(build_info: &BuildInfo) -> anyhow::Result<()> {
+    // Reading both `CLOUDPREM_*` and `CP_*` env vars for backward compatibility. The former is
+    // deprecated and can be removed after 2026-04-01.
+    let host: String = quickwit_common::get_from_env_opt("CLOUDPREM_DOGSTATSD_SERVER_HOST", false)
+        .unwrap_or_else(|| {
+            quickwit_common::get_from_env(
+                "CP_DOGSTATSD_SERVER_HOST",
+                "127.0.0.1".to_string(),
+                false,
+            )
+        });
+    let port: u16 = quickwit_common::get_from_env_opt("CLOUDPREM_DOGSTATSD_SERVER_PORT", false)
+        .unwrap_or_else(|| quickwit_common::get_from_env("CP_DOGSTATSD_SERVER_PORT", 8125, false));
+    let addr = format!("{host}:{port}");
+
+    let mut global_labels = vec![::metrics::Label::new("version", build_info.version.clone())];
+    let keys = [
+        ("IMAGE_NAME", "image_name"),
+        ("IMAGE_TAG", "image_tag"),
+        ("KUBERNETES_COMPONENT", "kube_component"),
+        ("KUBERNETES_NAMESPACE", "kube_namespace"),
+        ("KUBERNETES_POD_NAME", "kube_pod_name"),
+        ("QW_CLUSTER_ID", "cloudprem_cluster_id"),
+        ("QW_NODE_ID", "cloudprem_node_id"),
+    ];
+    for (env_var_key, label_key) in keys {
+        if let Some(label_val) = quickwit_common::get_from_env_opt::<String>(env_var_key, false) {
+            global_labels.push(::metrics::Label::new(label_key, label_val));
+        }
+    }
+    metrics_exporter_dogstatsd::DogStatsDBuilder::default()
+        .set_global_prefix("cloudprem")
+        .with_global_labels(global_labels)
+        .with_remote_address(addr)
+        .context("failed to parse DogStatsD server address")?
+        .install()
+        .context("failed to register DogStatsD exporter")?;
+    Ok(())
+}
+
+/// Register the invariant recorder that emits DogStatsD counters.
+///
+/// Must be called after [`setup_dogstatsd_exporter`] so the `metrics` crate
+/// has a registered recorder.
+#[cfg(not(any(test, feature = "testsuite")))]
+pub fn setup_invariant_recorder() {
+    quickwit_dst::invariants::set_invariant_recorder(invariant_recorder);
+}
+
+#[cfg(not(any(test, feature = "testsuite")))]
+fn invariant_recorder(id: quickwit_dst::invariants::InvariantId, passed: bool) {
+    let name = id.as_str();
+    metrics::counter!("pomsky.invariant.checked", "invariant" => name).increment(1);
+    if !passed {
+        metrics::counter!("pomsky.invariant.violated", "invariant" => name).increment(1);
+    }
+}
+
 /// We do not rely on the RFC3339 implementation, because it has a nanosecond precision.
 /// See discussion here: https://github.com/time-rs/time/discussions/418
 fn time_formatter() -> UtcTime<Vec<BorrowedFormatItem<'static>>> {

quickwit/quickwit-cli/src/logger.rs

@@ -101,6 +101,12 @@ async fn main_impl() -> anyhow::Result<()> {
     let (env_filter_reload_fn, tracer_provider_opt) =
         setup_logging_and_tracing(command.default_log_level(), ansi_colors, build_info)?;
 
+    #[cfg(not(any(test, feature = "testsuite")))]
+    quickwit_cli::logger::setup_dogstatsd_exporter(build_info)?;
+
+    #[cfg(not(any(test, feature = "testsuite")))]
+    quickwit_cli::logger::setup_invariant_recorder();
+
     let return_code: i32 = if let Err(command_error) = command.execute(env_filter_reload_fn).await {
         error!(error=%command_error, "command failed");
         eprintln!(