orchestratord: add reconciliation metrics and Kubernetes events

Cargo.toml

@@ -391,7 +391,7 @@ itertools = "0.15.0"
 jemalloc_pprof = "0.9.0"
 jsonwebtoken = { version = "11.1.0", features = ["aws_lc_rs"] }
 junit-report = "0.9.1"
-k8s-controller = "0.12.0"
+k8s-controller = "0.13.0"
 k8s-openapi = { version = "0.27.0", features = ["schemars", "v1_34"] }
 kube = { version = "3.1.0", default-features = false, features = ["aws-lc-rs", "client", "derive", "runtime", "rustls-tls"] }
 launchdarkly-server-sdk = { version = "3.3.0", default-features = false, features = ["hyper-rustls-native-roots", "crypto-aws-lc-rs"] }

ci/nightly/pipeline.template.yml

@@ -2800,6 +2800,20 @@ steps:
         agents:
           queue: hetzner-aarch64-16cpu-32gb
 
+      - id: orchestratord-failure-events
+        topics: [self-managed]
+        label: "Orchestratord reconciliation failure event tests"
+        artifact_paths: ["mz_debug_*.zip"]
+        depends_on: devel-docker-tags
+        timeout_in_minutes: 60
+        plugins:
+          - ./ci/plugins/mzcompose:
+              composition: orchestratord
+              run: failure-events
+              ci-builder: stable
+        agents:
+          queue: hetzner-aarch64-16cpu-32gb
+
       - id: orchestratord-manually-promote
         topics: [0dt-migration, self-managed]
         label: "Orchestratord ManuallyPromote tests"

doc/user/data/metrics.yml

@@ -4357,6 +4357,66 @@ metrics:
   help: Whether this operator replica holds the controller leadership lease, and is therefore the replica reconciling. Summed across the replicas this should be 1. A sustained 0 means no replica can take the lease, for instance because the service account lacks permission on leases, and the operator is reconciling nothing.
   source: src/orchestratord/src/metrics.rs
   visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_bucket
+  help: Time spent in one reconciliation pass.
+  labels:
+  - controller
+  - le
+  - phase
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_count
+  help: Time spent in one reconciliation pass.
+  labels:
+  - controller
+  - phase
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_sum
+  help: Time spent in one reconciliation pass.
+  labels:
+  - controller
+  - phase
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_bucket
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - le
+  - step
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_count
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - step
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_sum
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - step
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
+- name: orchestratord_reconciliation_steps_total
+  help: Count of reconciliation steps, by controller, by step, and by what the step concluded. An outcome of `abandoned` means the step did not conclude, either because an error propagated out of it or because the pass was cancelled.
+  labels:
+  - controller
+  - outcome
+  - step
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
+- name: orchestratord_reconciliations_total
+  help: Count of reconciliation passes, by controller, by the phase of the resource's lifecycle handled, and by what the pass concluded. An outcome of `failed` means the reconciler returned an error.
+  labels:
+  - controller
+  - outcome
+  - phase
+  source: src/orchestratord/src/metrics.rs
+  visibility: internal
 - name: outer_join_lowering_cases
   help: How many times the different outer join lowering cases happened.
   labels:

misc/helm-charts/operator/templates/clusterrole.yaml

@@ -37,6 +37,12 @@ rules:
   - get
   - list
   - watch
+- apiGroups: ["events.k8s.io"]
+  resources:
+  - events
+  verbs:
+  - create
+  - patch
 - apiGroups: ["coordination.k8s.io"]
   resources:
   - leases

misc/helm-charts/operator/tests/clusterrole_test.yaml

@@ -32,6 +32,12 @@ tests:
             apiGroups: ["rbac.authorization.k8s.io"]
             resources: ["roles", "rolebindings"]
             verbs: ["create", "update", "patch", "delete", "get", "list", "watch"]
+      - contains:
+          path: rules
+          content:
+            apiGroups: ["events.k8s.io"]
+            resources: ["events"]
+            verbs: ["create", "patch"]
 
   - it: should not create a clusterrole when RBAC is disabled
     set:

src/metrics-catalog/Cargo.toml

@@ -11,13 +11,20 @@ workspace = true
 
 [dependencies]
 anyhow.workspace = true
+k8s-controller = { workspace = true, features = ["prometheus"] }
+k8s-openapi.workspace = true
 mz-metrics = { path = "../metrics" }
 mz-ore = { path = "../ore", default-features = false, features = ["async"] }
+prometheus.workspace = true
 quote.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_yaml.workspace = true
 syn = { workspace = true, features = ["visit"] }
 walkdir.workspace = true
 
+[package.metadata.unused-deps]
+# Selects the Kubernetes API version that `k8s-controller` builds against.
+ignore = ["k8s-openapi"]
+
 [dev-dependencies]
 mz-ore = { path = "../ore", default-features = false, features = ["test"] }

src/metrics-catalog/src/main.rs

@@ -11,22 +11,25 @@
 //! source tree for the user-facing metrics documentation.
 //!
 //! It walks the Rust AST with `syn`, visiting every `metric!` invocation
-//! outside of test code.
+//! outside of test code, and every construction of the reconciliation metrics
+//! that `k8s_controller::PrometheusMetrics` defines.
 //!
 //! Regenerate with `bin/gen-metrics-catalog`.
 
 use std::process;
 
 use anyhow::{Context, bail};
 use mz_ore::metrics::{MetricTag, MetricVisibility};
+use prometheus::core::Collector as _;
+use prometheus::proto::MetricType;
 use quote::ToTokens;
 use serde::Serialize;
 use syn::parse::{Parse, ParseStream};
 use syn::punctuated::Punctuated;
 use syn::visit::Visit;
 use syn::{
-    Attribute, Expr, Ident, ImplItemFn, ItemFn, ItemImpl, ItemMod, Lit, Macro, Token, braced,
-    bracketed,
+    Attribute, Expr, ExprCall, Ident, ImplItemFn, ItemFn, ItemImpl, ItemMod, Lit, Macro, Token,
+    braced, bracketed,
 };
 use walkdir::WalkDir;
 
@@ -269,6 +272,70 @@ impl MetricArgs {
     }
 }
 
+/// Documents the metrics that `k8s_controller::PrometheusMetrics` defines
+/// when constructed with `namespace`, by constructing it and reading back its
+/// metrics' descriptions.
+fn k8s_controller_docs(namespace: &str, source: &str) -> Vec<MetricDoc> {
+    let metrics = match k8s_controller::PrometheusMetrics::new(namespace) {
+        Ok(metrics) => metrics,
+        Err(e) => {
+            eprintln!("warn: failed to construct PrometheusMetrics in {source}: {e}");
+            return Vec::new();
+        }
+    };
+    let descs = metrics.desc();
+    metrics
+        .collect()
+        .into_iter()
+        .flat_map(|family| {
+            let mut labels = descs
+                .iter()
+                .find(|desc| desc.fq_name == family.name())
+                .map(|desc| desc.variable_labels.clone())
+                .unwrap_or_default();
+            labels.sort();
+            if family.get_field_type() == MetricType::HISTOGRAM {
+                into_histogram_docs(
+                    family.name(),
+                    family.help(),
+                    labels,
+                    source,
+                    MetricVisibility::Internal,
+                    Vec::new(),
+                )
+            } else {
+                vec![MetricDoc {
+                    name: family.name().to_owned(),
+                    help: family.help().to_owned(),
+                    labels,
+                    source: source.to_owned(),
+                    visibility: MetricVisibility::Internal,
+                    tags: Vec::new(),
+                }]
+            }
+        })
+        .collect()
+}
+
+/// If `call` is `PrometheusMetrics::new(<string literal>)`, returns the
+/// literal.
+fn prometheus_metrics_namespace(call: &ExprCall) -> Option<String> {
+    let Expr::Path(func) = &*call.func else {
+        return None;
+    };
+    let mut segments = func.path.segments.iter().rev();
+    if segments.next()?.ident != "new" || segments.next()?.ident != "PrometheusMetrics" {
+        return None;
+    }
+    match call.args.first()? {
+        Expr::Lit(lit) => match &lit.lit {
+            Lit::Str(s) => Some(s.value()),
+            _ => None,
+        },
+        _ => None,
+    }
+}
+
 /// Renders a `subsystem` for use as a metric-name prefix: a string literal is
 /// used verbatim, while anything else (a runtime value like `component`) globs
 /// to `*`, since its concrete value is only known at runtime.
@@ -429,6 +496,14 @@ impl<'ast> Visit<'ast> for Collector<'_> {
         }
         syn::visit::visit_macro(self, mac);
     }
+
+    fn visit_expr_call(&mut self, call: &'ast ExprCall) {
+        if let Some(namespace) = prometheus_metrics_namespace(call) {
+            self.out
+                .extend(k8s_controller_docs(&namespace, self.source));
+        }
+        syn::visit::visit_expr_call(self, call);
+    }
 }
 
 fn run() -> anyhow::Result<()> {
@@ -449,7 +524,7 @@ fn run() -> anyhow::Result<()> {
         if path.extension().map(|e| e == "rs").unwrap_or(false) {
             let content =
                 std::fs::read_to_string(path).with_context(|| format!("reading {path:?}"))?;
-            if !content.contains("metric!") {
+            if !content.contains("metric!") && !content.contains("PrometheusMetrics::new") {
                 continue;
             }
             let ast = match syn::parse_file(&content) {
@@ -998,4 +1073,52 @@ mod tests {
             "macro_rules! body must not be scraped, got {names:?}"
         );
     }
+
+    #[mz_ore::test]
+    fn collects_k8s_controller_metrics() {
+        let file: syn::File = syn::parse_str(
+            r#"
+            fn register(registry: &MetricsRegistry) {
+                let reconcile = k8s_controller::PrometheusMetrics::new("operator").unwrap();
+                registry.register_collector(reconcile);
+            }
+            "#,
+        )
+        .expect("valid source");
+        let mut docs = Vec::new();
+        Collector {
+            source: "test.rs",
+            out: &mut docs,
+        }
+        .visit_file(&file);
+        let entries: Vec<_> = docs
+            .iter()
+            .map(|doc| (doc.name.as_str(), doc.labels.join(",")))
+            .collect();
+        for expected in [
+            ("operator_reconciliations_total", "controller,outcome,phase"),
+            (
+                "operator_reconciliation_duration_seconds_bucket",
+                "controller,le,phase",
+            ),
+            (
+                "operator_reconciliation_duration_seconds_count",
+                "controller,phase",
+            ),
+            (
+                "operator_reconciliation_steps_total",
+                "controller,outcome,step",
+            ),
+            (
+                "operator_reconciliation_step_duration_seconds_sum",
+                "controller,step",
+            ),
+        ] {
+            assert!(
+                entries.contains(&(expected.0, expected.1.to_owned())),
+                "missing {expected:?} in {entries:?}"
+            );
+        }
+        assert!(docs.iter().all(|doc| doc.source == "test.rs"));
+    }
 }

src/orchestratord/Cargo.toml

@@ -20,7 +20,7 @@ gcp_auth.workspace = true
 hex.workspace = true
 http.workspace = true
 humantime.workspace = true
-k8s-controller.workspace = true
+k8s-controller = { workspace = true, features = ["prometheus"] }
 k8s-openapi.workspace = true
 kube.workspace = true
 maplit.workspace = true