From 6b71418508507de675f59bb77d71191a9d5c16b5 Mon Sep 17 00:00:00 2001 From: Heather Lapointe Date: Wed, 30 Sep 2026 18:14:46 -0400 Subject: [PATCH] orchestratord: add reconciliation metrics and Kubernetes events Adopt k8s-controller 0.13.0, which provides reconciliation metrics and events, and use it in place of orchestratord's own wrapper: - report every controller's passes and named steps to k8s-controller's Prometheus metrics, under the orchestratord prefix - give each controller its own event recorder, so a failed pass is published as a ReconcileFailed warning event whose reporting controller names the controller that failed - publish each UpToDate transition of a Materialize as an event - grant the operator create and patch on events.k8s.io events - teach mz-metrics-catalog to document metrics constructed with PrometheusMetrics::new, which it could not see as metric! invocations Part of CLO-188. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 8 +- Cargo.toml | 2 +- ci/nightly/pipeline.template.yml | 14 ++ doc/user/data/metrics.yml | 60 +++++++ .../operator/templates/clusterrole.yaml | 6 + .../operator/tests/clusterrole_test.yaml | 6 + src/metrics-catalog/Cargo.toml | 7 + src/metrics-catalog/src/main.rs | 131 +++++++++++++- src/orchestratord/Cargo.toml | 2 +- src/orchestratord/src/bin/orchestratord.rs | 24 +++ src/orchestratord/src/controller.rs | 12 ++ src/orchestratord/src/controller/balancer.rs | 20 ++- src/orchestratord/src/controller/console.rs | 28 ++- .../src/controller/materialize.rs | 170 +++++++++++++++--- src/orchestratord/src/metrics.rs | 12 ++ test/orchestratord/mzcompose.py | 148 +++++++++++++++ 16 files changed, 617 insertions(+), 33 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 54fe729c39047..e010e3ab40814 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5305,15 +5305,16 @@ dependencies = [ [[package]] name = "k8s-controller" -version = "0.12.0" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "548329de3a20cd4f697d9d6f99512f8a1b3b1b9526ed05186c70a7bb24a06b41" +checksum = "25b4b06f56a445e20f50c52894e412b2a3c5f2be97d03dd4326c6c768558dee1" dependencies = [ "async-trait", "futures", "k8s-openapi", "kube", "kube-runtime", + "prometheus", "rand 0.9.4", "serde", "serde_json", @@ -7538,8 +7539,11 @@ name = "mz-metrics-catalog" version = "0.0.0" dependencies = [ "anyhow", + "k8s-controller", + "k8s-openapi", "mz-metrics", "mz-ore", + "prometheus", "quote", "serde", "serde_yaml", diff --git a/Cargo.toml b/Cargo.toml index 00c9e583cf159..e788792e5c0d2 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/ci/nightly/pipeline.template.yml b/ci/nightly/pipeline.template.yml index 6efdadea8bf7c..1d93bf326ce60 100644 --- a/ci/nightly/pipeline.template.yml +++ b/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" diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index dbd4a84a4191c..24a227facfe13 100644 --- a/doc/user/data/metrics.yml +++ b/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: diff --git a/misc/helm-charts/operator/templates/clusterrole.yaml b/misc/helm-charts/operator/templates/clusterrole.yaml index 23cc4c105112a..770d57aad44e7 100644 --- a/misc/helm-charts/operator/templates/clusterrole.yaml +++ b/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 diff --git a/misc/helm-charts/operator/tests/clusterrole_test.yaml b/misc/helm-charts/operator/tests/clusterrole_test.yaml index 1e23f034d2558..35d82a3df4d6b 100644 --- a/misc/helm-charts/operator/tests/clusterrole_test.yaml +++ b/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: diff --git a/src/metrics-catalog/Cargo.toml b/src/metrics-catalog/Cargo.toml index 0ebaf772e1ad3..2f35e02f8fa6e 100644 --- a/src/metrics-catalog/Cargo.toml +++ b/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"] } diff --git a/src/metrics-catalog/src/main.rs b/src/metrics-catalog/src/main.rs index 77f29c8d5b2bb..827bb207905e4 100644 --- a/src/metrics-catalog/src/main.rs +++ b/src/metrics-catalog/src/main.rs @@ -11,7 +11,8 @@ //! 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`. @@ -19,14 +20,16 @@ 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 { + 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()`, returns the +/// literal. +fn prometheus_metrics_namespace(call: &ExprCall) -> Option { + 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")); + } } diff --git a/src/orchestratord/Cargo.toml b/src/orchestratord/Cargo.toml index a670518d29a3c..475e818649610 100644 --- a/src/orchestratord/Cargo.toml +++ b/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 diff --git a/src/orchestratord/src/bin/orchestratord.rs b/src/orchestratord/src/bin/orchestratord.rs index 1f464c483a399..4cd16f7064fce 100644 --- a/src/orchestratord/src/bin/orchestratord.rs +++ b/src/orchestratord/src/bin/orchestratord.rs @@ -18,6 +18,7 @@ use std::{ use axum_server::tls_rustls::RustlsConfig; use futures::future::{Either, join4, select}; use http::HeaderValue; +use k8s_controller::{PrometheusMetrics, ReconcileObserver, events::EventRecorder}; use k8s_openapi::{ api::{ apps::v1::Deployment, @@ -637,12 +638,22 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { &leader_election_identity, ); + let observer: Arc = Arc::::clone(&metrics.reconcile); + let event_recorder = |controller: &str| { + Arc::new(EventRecorder::new( + client.clone(), + controller::event_reporter(controller, leader_election_identity.clone()), + )) + }; + // Each of these is rebuilt every time this replica wins the election, since // running a controller consumes it and we rejoin the election after losing // the lease. let make_materialize_controller = { let client = client.clone(); let metrics = Arc::clone(&metrics); + let observer = Arc::clone(&observer); + let events = event_recorder(controller::materialize::CONTROLLER_NAME); let config = controller::materialize::Config { cloud_provider: args.cloud_provider, region: args.region, @@ -722,6 +733,9 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { controller::materialize::Context::new(config.clone(), Arc::clone(&metrics)), watcher::Config::default().timeout(29), ) + .with_name(controller::materialize::CONTROLLER_NAME) + .with_observer(Arc::clone(&observer)) + .with_event_recorder(Arc::clone(&events)) .with_controller(|controller| { let controller = controller .owns( @@ -763,6 +777,8 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { }; let make_balancer_controller = { let client = client.clone(); + let observer = Arc::clone(&observer); + let events = event_recorder(controller::balancer::CONTROLLER_NAME); let config = controller::balancer::Config { enable_security_context: args.enable_security_context, enable_prometheus_scrape_annotations: args.enable_prometheus_scrape_annotations, @@ -785,6 +801,9 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { controller::balancer::Context::new(config.clone()), watcher::Config::default().timeout(29), ) + .with_name(controller::balancer::CONTROLLER_NAME) + .with_observer(Arc::clone(&observer)) + .with_event_recorder(Arc::clone(&events)) .with_controller(|controller| { let controller = controller .owns( @@ -814,6 +833,8 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { }; let make_console_controller = { let client = client.clone(); + let observer = Arc::clone(&observer); + let events = event_recorder(controller::console::CONTROLLER_NAME); let config = controller::console::Config { enable_security_context: args.enable_security_context, enable_prometheus_scrape_annotations: args.enable_prometheus_scrape_annotations, @@ -835,6 +856,9 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { controller::console::Context::new(config.clone()), watcher::Config::default().timeout(29), ) + .with_name(controller::console::CONTROLLER_NAME) + .with_observer(Arc::clone(&observer)) + .with_event_recorder(Arc::clone(&events)) .with_controller(|controller| { let controller = controller .owns( diff --git a/src/orchestratord/src/controller.rs b/src/orchestratord/src/controller.rs index a4d0b8c23bb4e..22d3172d49397 100644 --- a/src/orchestratord/src/controller.rs +++ b/src/orchestratord/src/controller.rs @@ -7,6 +7,18 @@ // the Business Source License, use of this software will be governed // by the Apache License, Version 2.0. +use k8s_controller::events::Reporter; + pub mod balancer; pub mod console; pub mod materialize; + +/// The reporter that the events published by the controller named +/// `controller` (one of the `CONTROLLER_NAME`s) carry. `instance` identifies +/// this replica. +pub fn event_reporter(controller: &str, instance: String) -> Reporter { + Reporter { + controller: format!("orchestratord.materialize.cloud/{controller}"), + instance: Some(instance), + } +} diff --git a/src/orchestratord/src/controller/balancer.rs b/src/orchestratord/src/controller/balancer.rs index 6bd20261e25e0..8d826c503046b 100644 --- a/src/orchestratord/src/controller/balancer.rs +++ b/src/orchestratord/src/controller/balancer.rs @@ -8,7 +8,7 @@ // by the Apache License, Version 2.0. use anyhow::bail; -use k8s_controller::TraceMetadata; +use k8s_controller::{Outcome, TraceMetadata}; use k8s_openapi::{ api::{ apps::v1::{Deployment, DeploymentSpec, DeploymentStrategy, RollingUpdateDeployment}, @@ -46,6 +46,10 @@ use mz_cloud_resources::crd::{ use mz_orchestrator_kubernetes::KubernetesImagePullPolicy; use mz_ore::{cli::KeyValueArg, instrument}; +/// The name identifying this controller in its reconciliation metrics and in +/// the reporter of the events it publishes. +pub const CONTROLLER_NAME: &str = "balancer"; + #[derive(Clone)] pub struct Config { pub enable_security_context: bool, @@ -587,9 +591,10 @@ impl k8s_controller::Context for Context { &self, client: Client, balancer: &Self::Resource, - _metadata: &mut TraceMetadata, + metadata: &mut TraceMetadata, ) -> Result, Self::Error> { if balancer.status.is_none() { + let step = metadata.step("initialize_status"); let balancer_api: Api = Api::namespaced(client.clone(), &balancer.meta().namespace.clone().unwrap()); let mut new_balancer = balancer.clone(); @@ -601,6 +606,7 @@ impl k8s_controller::Context for Context { &new_balancer, ) .await?; + step.finish(Outcome::Completed); // Updating the status should trigger a reconciliation // which will include a status this time. return Ok(None); @@ -611,21 +617,31 @@ impl k8s_controller::Context for Context { let deployment_api: Api = Api::namespaced(client.clone(), &namespace); let service_api: Api = Api::namespaced(client.clone(), &namespace); + let step = metadata.step("certificate"); if let Some(external_certificate) = self.create_external_certificate_object(balancer)? { trace!("creating new balancerd external certificate"); apply_resource(&certificate_api, &external_certificate).await?; + step.finish(Outcome::Completed); + } else { + step.finish(Outcome::Skipped); } + let step = metadata.step("deployment"); let deployment = self.create_deployment_object(balancer)?; self.fix_deployment(&deployment_api, &deployment).await?; trace!("creating new balancerd deployment"); apply_resource(&deployment_api, &deployment).await?; + step.finish(Outcome::Completed); + let step = metadata.step("service"); let service = self.create_service_object(balancer); trace!("creating new balancerd service"); apply_resource(&service_api, &service).await?; + step.finish(Outcome::Completed); + let step = metadata.step("sync_status"); self.sync_deployment_status(&client, balancer).await?; + step.finish(Outcome::Completed); Ok(None) } diff --git a/src/orchestratord/src/controller/console.rs b/src/orchestratord/src/controller/console.rs index 173bef895c2e1..f94a9cdf38564 100644 --- a/src/orchestratord/src/controller/console.rs +++ b/src/orchestratord/src/controller/console.rs @@ -7,7 +7,7 @@ // the Business Source License, use of this software will be governed // by the Apache License, Version 2.0. -use k8s_controller::TraceMetadata; +use k8s_controller::{Outcome, TraceMetadata}; use k8s_openapi::{ api::{ apps::v1::{Deployment, DeploymentSpec}, @@ -51,6 +51,10 @@ use mz_orchestrator_kubernetes::KubernetesImagePullPolicy; use mz_ore::{cli::KeyValueArg, instrument}; use mz_server_core::listeners::AuthenticatorKind; +/// The name identifying this controller in its reconciliation metrics and in +/// the reporter of the events it publishes. +pub const CONTROLLER_NAME: &str = "console"; + #[derive(Clone)] pub struct Config { pub enable_security_context: bool, @@ -587,9 +591,10 @@ impl k8s_controller::Context for Context { &self, client: Client, console: &Self::Resource, - _metadata: &mut TraceMetadata, + metadata: &mut TraceMetadata, ) -> Result, Self::Error> { if console.status.is_none() { + let step = metadata.step("initialize_status"); let console_api: Api = Api::namespaced(client.clone(), &console.meta().namespace.clone().unwrap()); let mut new_console = console.clone(); @@ -601,6 +606,7 @@ impl k8s_controller::Context for Context { &new_console, ) .await?; + step.finish(Outcome::Completed); // Updating the status should trigger a reconciliation // which will include a status this time. return Ok(None); @@ -613,33 +619,51 @@ impl k8s_controller::Context for Context { let service_api: Api = Api::namespaced(client.clone(), &namespace); let certificate_api: Api = Api::namespaced(client.clone(), &namespace); + let step = metadata.step("network_policies"); trace!("creating new network policies"); let network_policies = self.create_network_policies(console); for network_policy in &network_policies { apply_resource(&network_policy_api, network_policy).await?; } + step.finish(if network_policies.is_empty() { + Outcome::Skipped + } else { + Outcome::Completed + }); + let step = metadata.step("configmap"); trace!("creating new console configmap"); let console_configmap = self.create_console_app_configmap_object(console); apply_resource(&configmap_api, &console_configmap).await?; + step.finish(Outcome::Completed); + let step = metadata.step("deployment"); trace!("creating new console deployment"); let console_deployment = self.create_console_deployment_object(console); self.fix_deployment(&deployment_api, &console_deployment) .await?; apply_resource(&deployment_api, &console_deployment).await?; + step.finish(Outcome::Completed); + let step = metadata.step("service"); trace!("creating new console service"); let console_service = self.create_console_service_object(console); apply_resource(&service_api, &console_service).await?; + step.finish(Outcome::Completed); + let step = metadata.step("certificate"); let console_external_certificate = self.create_console_external_certificate(console)?; if let Some(certificate) = &console_external_certificate { trace!("creating new console external certificate"); apply_resource(&certificate_api, certificate).await?; + step.finish(Outcome::Completed); + } else { + step.finish(Outcome::Skipped); } + let step = metadata.step("sync_status"); self.sync_deployment_status(&client, console).await?; + step.finish(Outcome::Completed); Ok(None) } diff --git a/src/orchestratord/src/controller/materialize.rs b/src/orchestratord/src/controller/materialize.rs index 4edb99d638a9a..adb21e4cd3545 100644 --- a/src/orchestratord/src/controller/materialize.rs +++ b/src/orchestratord/src/controller/materialize.rs @@ -15,7 +15,10 @@ use std::{ use anyhow::Context as _; use http::HeaderValue; -use k8s_controller::TraceMetadata; +use k8s_controller::{ + Outcome, TraceMetadata, + events::{Event, EventType}, +}; use k8s_openapi::{ api::core::v1::{Affinity, ResourceRequirements, Secret, Toleration}, apimachinery::pkg::apis::meta::v1::{Condition, Time}, @@ -54,6 +57,10 @@ use mz_ore::{cast::CastFrom, cli::KeyValueArg, instrument}; pub mod generation; pub mod global; +/// The name identifying this controller in its reconciliation metrics and in +/// the reporter of the events it publishes. +pub const CONTROLLER_NAME: &str = "materialize"; + #[derive(Clone)] pub struct Config { pub cloud_provider: CloudProvider, @@ -166,8 +173,12 @@ impl Context { .set(u64::cast_from(needs_update_set.len())); } + /// Writes `status`, if it differs from the current status in anything but + /// its timestamps, and publishes the new `UpToDate` condition as an event + /// once it has been written. async fn update_status( &self, + metadata: &TraceMetadata, mz_api: &Api, mz: &Materialize, status: MaterializeStatus, @@ -184,14 +195,48 @@ impl Context { return Ok(new_mz); } + let condition = status.conditions.first().cloned(); new_mz.status = Some(status); - mz_api + let new_mz = mz_api .replace_status(&mz.name_unchecked(), &PostParams::default(), &new_mz) - .await + .await?; + + if let Some(condition) = condition { + metadata + .publish_event(Event { + type_: transition_event_type(&condition), + reason: condition.reason, + action: "Reconcile".into(), + note: Some(condition.message), + related: None, + }) + .await; + } + + Ok(new_mz) + } + + /// Deletes `generation`'s resources, releasing the read holds its + /// environmentd was keeping. + async fn teardown_generation( + &self, + metadata: &TraceMetadata, + client: &Client, + mz: &Materialize, + resources: &generation::Resources, + generation: u64, + ) -> Result<(), Error> { + let step = metadata.step("teardown_generation"); + resources + .teardown_generation(client, mz, generation) + .await?; + step.finish(Outcome::Completed); + Ok(()) } async fn promote( &self, + metadata: &TraceMetadata, client: &Client, mz: &Materialize, resources: generation::Resources, @@ -199,14 +244,18 @@ impl Context { desired_generation: u64, resources_hash: String, ) -> Result, Error> { + let step = metadata.step("promote"); if let Some(action) = resources.promote_services(client, &mz.namespace()).await? { + step.finish(Outcome::Waiting); return Ok(Some(action)); } - resources - .teardown_generation(client, mz, active_generation) + step.finish(Outcome::Completed); + + self.teardown_generation(metadata, client, mz, &resources, active_generation) .await?; let mz_api: Api = Api::namespaced(client.clone(), &mz.namespace()); self.update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -282,8 +331,9 @@ impl k8s_controller::Context for Context { &self, client: Client, mz: &Self::Resource, - _metadata: &mut TraceMetadata, + metadata: &mut TraceMetadata, ) -> Result, Self::Error> { + let metadata = &*metadata; let mz_api: Api = Api::namespaced(client.clone(), &mz.namespace()); let balancer_api: Api = Api::namespaced(client.clone(), &mz.namespace()); let console_api: Api = Api::namespaced(client.clone(), &mz.namespace()); @@ -291,12 +341,16 @@ impl k8s_controller::Context for Context { let status = mz.status(); if mz.status.is_none() { - self.update_status(&mz_api, mz, status, true).await?; + let step = metadata.step("initialize_status"); + self.update_status(metadata, &mz_api, mz, status, true) + .await?; + step.finish(Outcome::Completed); // Updating the status should trigger a reconciliation // which will include a status this time. return Ok(None); } + let step = metadata.step("resolve_environment_id"); let backend_secret = secret_api.get(&mz.spec.backend_secret_name).await?; let license_key_environment_id: Option = if let Some(license_key) = backend_secret .data @@ -352,6 +406,7 @@ impl k8s_controller::Context for Context { mz_api .replace(&mz.name_unchecked(), &PostParams::default(), &mz) .await?; + step.finish(Outcome::Completed); // Updating the spec should also trigger a reconciliation. // We can't do that as part of the above check because you can't // update both the spec and the status in a single api call. @@ -372,10 +427,13 @@ impl k8s_controller::Context for Context { } self.check_environment_id_conflicts(&client, mz).await?; + step.finish(Outcome::Completed); + let step = metadata.step("global_resources"); global::Resources::new(&self.config, mz)? .apply(&client, &mz.namespace()) .await?; + step.finish(Outcome::Completed); // we compare the hash against the environment resources generated // for the current active generation, since that's what we expect to @@ -407,6 +465,7 @@ impl k8s_controller::Context for Context { // We don't know if we successfully promoted or not yet. (true, _, _) => { self.promote( + metadata, &client, mz, resources, @@ -441,10 +500,16 @@ impl k8s_controller::Context for Context { ); // Tear down the un-promoted generation to release // its read holds. - resources - .teardown_generation(&client, mz, next_generation) - .await?; + self.teardown_generation( + metadata, + &client, + mz, + &resources, + next_generation, + ) + .await?; self.update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -488,6 +553,7 @@ impl k8s_controller::Context for Context { status.last_completed_rollout_environmentd_image_ref; self.update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -539,6 +605,7 @@ impl k8s_controller::Context for Context { } else { &self .update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -577,16 +644,17 @@ impl k8s_controller::Context for Context { // The only reason someone would choose this strategy is if they didn't have // space for the two generations of pods. // Lets make room for the new ones by deleting the old generation. - resources - .teardown_generation(&client, mz, active_generation) + self.teardown_generation(metadata, &client, mz, &resources, active_generation) .await?; } trace!("applying environment resources"); - match resources + let step = metadata.step("generation_resources"); + let applied = resources .apply(&client, mz.should_force_promote(), &mz.namespace()) - .await - { + .await; + step.finish_with(&applied); + match applied { Ok(Some(action)) => { trace!("new environment is not yet ready"); Ok(Some(action)) @@ -599,6 +667,7 @@ impl k8s_controller::Context for Context { "Ready to promote, but not promoting because the instance is configured with ManuallyPromote rollout strategy." ); self.update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -642,6 +711,7 @@ impl k8s_controller::Context for Context { // we've crossed the point of no return. // Once we see this status, we must promote without taking other actions. self.update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -672,6 +742,7 @@ impl k8s_controller::Context for Context { ) .await?; self.promote( + metadata, &client, mz, resources, @@ -683,6 +754,7 @@ impl k8s_controller::Context for Context { } Err(e) => { self.update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -721,14 +793,12 @@ impl k8s_controller::Context for Context { (false, true, false) => { let mut needs_update = mz.conditions_need_update(); if mz.update_in_progress() { - resources - .teardown_generation(&client, mz, next_generation) + self.teardown_generation(metadata, &client, mz, &resources, next_generation) .await?; needs_update = true; } if needs_update { - self.update_status( - &mz_api, + self.update_status(metadata, &mz_api, mz, MaterializeStatus { active_generation, @@ -764,13 +834,13 @@ impl k8s_controller::Context for Context { // WaitingForApproval. let mut needs_update = mz.conditions_need_update() || mz.rollout_requested(); if mz.update_in_progress() { - resources - .teardown_generation(&client, mz, next_generation) + self.teardown_generation(metadata, &client, mz, &resources, next_generation) .await?; needs_update = true; } if needs_update { self.update_status( + metadata, &mz_api, mz, MaterializeStatus { @@ -809,6 +879,7 @@ impl k8s_controller::Context for Context { // enforced by the environmentd rollout process being able to call // into the promotion endpoint + let step = metadata.step("balancer"); if self.config.create_balancers { let balancer = Balancer { metadata: mz.managed_resource_meta(mz.name_unchecked()), @@ -838,8 +909,13 @@ impl k8s_controller::Context for Context { }; let balancer = apply_resource(&balancer_api, &balancer).await?; result = wait_for_balancer(&balancer)?; + step.finish(match result { + Some(_) => Outcome::Waiting, + None => Outcome::Completed, + }); } else { delete_resource(&balancer_api, &mz.name_unchecked()).await?; + step.finish(Outcome::Skipped); } if let Some(action) = result { @@ -849,6 +925,7 @@ impl k8s_controller::Context for Context { // and the console relies on the balancer service existing, which is // enforced by wait_for_balancer + let step = metadata.step("console"); if self.config.create_console { let active_environmentd_image_ref = mz.active_environmentd_image_ref(); let environmentd_image_tag = @@ -895,8 +972,10 @@ impl k8s_controller::Context for Context { status: None, }; apply_resource(&console_api, &console).await?; + step.finish(Outcome::Completed); } else { delete_resource(&console_api, &mz.name_unchecked()).await?; + step.finish(Outcome::Skipped); } Ok(result) @@ -915,6 +994,18 @@ impl k8s_controller::Context for Context { } } +/// The type of the event reporting that the `UpToDate` condition became +/// `condition`: a warning if the environment is not up to date for any reason +/// other than waiting for a rollout to be approved, which is the operator +/// following its configuration. +fn transition_event_type(condition: &Condition) -> EventType { + match condition.reason.as_str() { + "WaitingForApproval" => EventType::Normal, + _ if condition.status == "False" => EventType::Warning, + _ => EventType::Normal, + } +} + fn wait_for_balancer(balancer: &Balancer) -> Result, Error> { if let Some(conditions) = balancer .status @@ -931,3 +1022,40 @@ fn wait_for_balancer(balancer: &Balancer) -> Result, Error> { Ok(Some(Action::requeue(Duration::from_secs(1)))) } + +#[cfg(test)] +mod tests { + use super::*; + + fn condition(status: &str, reason: &str) -> Condition { + Condition { + type_: "UpToDate".into(), + status: status.into(), + reason: reason.into(), + message: String::new(), + last_transition_time: Time(Timestamp::now()), + observed_generation: None, + } + } + + #[mz_ore::test] + fn test_transition_event_type() { + for (status, reason, expected) in [ + ("True", "Applied", EventType::Normal), + ("Unknown", "Applying", EventType::Normal), + ("Unknown", "ReadyToPromote", EventType::Normal), + ("Unknown", "Promoting", EventType::Normal), + ("False", "WaitingForApproval", EventType::Normal), + ("False", "FailedDeploy", EventType::Warning), + ("False", "RolloutTimeout", EventType::Warning), + ("False", "SomeFutureFailure", EventType::Warning), + ("Unknown", "SomeFuturePhase", EventType::Normal), + ] { + assert_eq!( + transition_event_type(&condition(status, reason)), + expected, + "status={status} reason={reason}", + ); + } + } +} diff --git a/src/orchestratord/src/metrics.rs b/src/orchestratord/src/metrics.rs index 3125d7f7f41a1..58dc3d0f59236 100644 --- a/src/orchestratord/src/metrics.rs +++ b/src/orchestratord/src/metrics.rs @@ -8,10 +8,12 @@ // by the Apache License, Version 2.0. use std::collections::BTreeMap; +use std::sync::Arc; use std::time::Duration; use axum::{Extension, Router, body::Body, routing::get}; use http::{HeaderMap, Method, Request, Response, StatusCode}; +use k8s_controller::PrometheusMetrics; use prometheus::{Encoder, TextEncoder}; use tower_http::{classify::ServerErrorsFailureClass, trace::TraceLayer}; use tracing::{Level, Span}; @@ -23,6 +25,10 @@ use mz_ore::metrics::{MetricsRegistry, UIntGauge}; pub struct Metrics { pub is_leader: UIntGauge, pub environmentd_needs_update: UIntGauge, + /// The reconciliation metrics of every controller, as defined by + /// `k8s_controller`: counts and durations of reconciliation passes, and + /// of the steps within them. + pub reconcile: Arc, } impl Metrics { @@ -38,6 +44,12 @@ impl Metrics { name: "environmentd_needs_update", help: "Count of organizations in this cluster which are running outdated pod templates. Only the operator replica holding the leadership lease reconciles, so the others report zero.", }), + reconcile: { + let reconcile = PrometheusMetrics::new("orchestratord") + .expect("valid reconciliation metric definitions"); + registry.register_collector(reconcile.clone()); + Arc::new(reconcile) + }, } } diff --git a/test/orchestratord/mzcompose.py b/test/orchestratord/mzcompose.py index ef152ee8415ac..324aac76ad727 100644 --- a/test/orchestratord/mzcompose.py +++ b/test/orchestratord/mzcompose.py @@ -3630,6 +3630,118 @@ def ca_bundle_refreshed() -> None: print("webhook cert rotation test PASSED") +# The reporter of the events the Materialize controller publishes. +MATERIALIZE_EVENT_REPORTER = "orchestratord.materialize.cloud/materialize" + + +def get_materialize_events(definition: dict[str, Any]) -> list[dict[str, Any]]: + """Return the events published about the test's Materialize resource.""" + name = definition["materialize"]["metadata"]["name"] + events = json.loads( + spawn.capture( + [ + "kubectl", + "get", + # `kubectl get events` serves the core v1 view of events, which + # renames the events.k8s.io fields these tests read. + "events.events.k8s.io", + "-n", + "materialize-environment", + "-o", + "json", + ], + ) + )["items"] + return [ + event + for event in events + if event.get("regarding", {}).get("kind") == "Materialize" + and event.get("regarding", {}).get("name") == name + ] + + +def wait_for_failure_note( + definition: dict[str, Any], + matches: Callable[[str], bool], +) -> str: + """Wait for a ReconcileFailed event on the Materialize resource whose note + satisfies `matches`, and return that note.""" + found: list[str] = [] + + def check() -> None: + failures = [ + event + for event in get_materialize_events(definition) + if event.get("reason") == "ReconcileFailed" + ] + matching = [event for event in failures if matches(event.get("note") or "")] + assert ( + matching + ), f"No matching ReconcileFailed event; notes were {[e.get('note') for e in failures]}" + assert matching[0]["type"] == "Warning", matching[0] + found.append(matching[0]["note"]) + + retry(check, 300) + return found[0] + + +def workflow_failure_events( + c: Composition, + parser: WorkflowArgumentParser, +) -> None: + """Test that the ReconcileFailed event on a Materialize resource follows + the current cause of the failure, rather than the first one. + + The environment first fails because its backend secret is missing, then, + once the secret exists with an invalid license key, because the key fails + to validate. + """ + parser.add_argument( + "--recreate-cluster", + action=argparse.BooleanOptionalAction, + help="Recreate cluster if it exists already", + ) + parser.add_argument( + "--tag", + type=str, + help="Custom version tag to use", + ) + parser.add_argument( + "--orchestratord-override", + default=True, + action=argparse.BooleanOptionalAction, + help="Override orchestratord tag", + ) + args = parser.parse_args() + + definition = setup(c, args) + init(definition) + + spawn.runv( + ["kubectl", "apply", "-f", "-"], + stdin=yaml.dump_all( + [definition["namespace"], definition["materialize"]] + ).encode(), + ) + first = wait_for_failure_note(definition, lambda note: "not found" in note.lower()) + print(f"First failure reported: {first}") + + secret = copy.deepcopy(definition["secret"]) + secret["stringData"]["license_key"] = "not-a-real-license-key" + spawn.runv( + ["kubectl", "apply", "-f", "-"], + stdin=yaml.dump_all([secret]).encode(), + ) + second = wait_for_failure_note(definition, lambda note: note != first) + print(f"Second failure reported: {second}") + + reporters = { + event["reportingController"] for event in get_materialize_events(definition) + } + assert reporters == {MATERIALIZE_EVENT_REPORTER}, reporters + print("workflow_failure_events PASSED") + + def workflow_manually_promote( c: Composition, parser: WorkflowArgumentParser, @@ -3735,9 +3847,23 @@ def workflow_manually_promote( wait_for_rollout_complete() print("Test 2 PASSED: Promotion via v1 requestedRolloutHash succeeded") + check_lifecycle_events(definition) print("workflow_manually_promote PASSED") +def check_lifecycle_events(definition: dict[str, Any]) -> None: + """Assert that each phase of a ManuallyPromote rollout was published as an + event on the Materialize resource, with the condition's reason and + message.""" + by_reason = {event["reason"]: event for event in get_materialize_events(definition)} + for reason in ("Applying", "ReadyToPromote", "Promoting", "Applied"): + assert reason in by_reason, f"No {reason} event, found {sorted(by_reason)}" + event = by_reason[reason] + assert event["type"] == "Normal", event + assert event["reportingController"] == MATERIALIZE_EVENT_REPORTER, event + assert "generation" in (event.get("note") or ""), event + + def apply_materialize(definition: dict[str, Any]) -> None: """Apply the materialize resource definition via kubectl.""" defs = [ @@ -5364,6 +5490,28 @@ def post_run_check(definition: dict[str, Any], expect_fail: bool) -> None: ) raise ValueError("Never completed") + if expect_fail: + wait_for_reconcile_failed_event(definition) + + +def wait_for_reconcile_failed_event(definition: dict[str, Any]) -> None: + """Assert that the operator published a ReconcileFailed warning event on + the Materialize resource, with the error as its note.""" + + def check() -> None: + failures = [ + event + for event in get_materialize_events(definition) + if event.get("reason") == "ReconcileFailed" + ] + assert failures, "No ReconcileFailed event on the Materialize resource" + event = failures[0] + assert event["type"] == "Warning", event + assert event["reportingController"] == MATERIALIZE_EVENT_REPORTER, event + assert event.get("note"), event + + retry(check, 120) + def run_balancer(definition: dict[str, Any], expect_fail: bool) -> None: defs = [