feat(31): shared invariants module in quickwit-dst

quickwit/Cargo.toml

@@ -13,6 +13,7 @@ members = [
   "quickwit-datetime",
   "quickwit-directories",
   "quickwit-doc-mapper",
+  "quickwit-dst",
   "quickwit-index-management",
   "quickwit-indexing",
   "quickwit-ingest",
@@ -155,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"
@@ -233,6 +236,7 @@ serde_yaml = "0.9"
 serial_test = { version = "3.2", features = ["file_locks"] }
 sha2 = "0.10"
 siphasher = "1.0"
+stateright = "0.30"
 smallvec = "1"
 sqlx = { version = "0.8", features = [
   "migrate",
@@ -352,6 +356,7 @@ quickwit-control-plane = { path = "quickwit-control-plane" }
 quickwit-datetime = { path = "quickwit-datetime" }
 quickwit-directories = { path = "quickwit-directories" }
 quickwit-doc-mapper = { path = "quickwit-doc-mapper" }
+quickwit-dst = { path = "quickwit-dst" }
 quickwit-index-management = { path = "quickwit-index-management" }
 quickwit-indexing = { path = "quickwit-indexing" }
 quickwit-ingest = { path = "quickwit-ingest" }

quickwit/quickwit-cli/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/src/logger.rs

@@ -161,6 +161,65 @@ pub fn setup_logging_and_tracing(
     ))
 }
 
+#[cfg(not(test))]
+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(test))]
+pub fn setup_invariant_recorder() {
+    quickwit_dst::invariants::set_invariant_recorder(invariant_recorder);
+}
+
+#[cfg(not(test))]
+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>>> {