CLO-188 Add events and more metrics to orchestratord

ci/nightly/pipeline.template.yml

@@ -2763,6 +2763,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

@@ -4381,6 +4381,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. This covers only the reconciler itself, not the finalizer bookkeeping around it, and a pass that waits on a rollout returns promptly rather than blocking, so this measures work done and not the wall-clock length of a rollout.
+  labels:
+  - controller
+  - event_type
+  - le
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_count
+  help: Time spent in one reconciliation pass. This covers only the reconciler itself, not the finalizer bookkeeping around it, and a pass that waits on a rollout returns promptly rather than blocking, so this measures work done and not the wall-clock length of a rollout.
+  labels:
+  - controller
+  - event_type
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_duration_seconds_sum
+  help: Time spent in one reconciliation pass. This covers only the reconciler itself, not the finalizer bookkeeping around it, and a pass that waits on a rollout returns promptly rather than blocking, so this measures work done and not the wall-clock length of a rollout.
+  labels:
+  - controller
+  - event_type
+  source: src/orchestratord/src/reconcile.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/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_count
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - step
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_step_duration_seconds_sum
+  help: Time spent in one reconciliation step.
+  labels:
+  - controller
+  - step
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliation_steps_total
+  help: Count of reconciliation steps, by controller, by step, and by what the step concluded. Steps are the named phases a reconciliation pass moves through, so this is where a pass that fails or stalls identifies which phase it was in. An `outcome` of `abandoned` means the step did not conclude, which covers both an error propagating out of it and the pass being cancelled by a leadership handoff, so alert on orchestratord_reconciliations_total instead and read this to locate the step.
+  labels:
+  - controller
+  - outcome
+  - step
+  source: src/orchestratord/src/reconcile.rs
+  visibility: internal
+- name: orchestratord_reconciliations_total
+  help: Count of reconciliation passes, by controller, by which of the controller's entry points ran, and by what the pass concluded. An `outcome` of `waiting` means the pass asked to be retried, which is normal during a rollout; `failed` means it returned an error, and is also reported as a Kubernetes event on the resource.
+  labels:
+  - controller
+  - event_type
+  - outcome
+  source: src/orchestratord/src/reconcile.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,16 @@ rules:
   - get
   - list
   - watch
+# The controllers publish a warning event on the Materialize, Balancer, or
+# Console resource whose reconciliation failed, so that `kubectl describe`
+# explains a stuck object without needing the operator's logs. Patch is for
+# the aggregation of a repeated failure into one event with a count.
+- 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,15 @@ tests:
             apiGroups: ["rbac.authorization.k8s.io"]
             resources: ["roles", "rolebindings"]
             verbs: ["create", "update", "patch", "delete", "get", "list", "watch"]
+      # Without this the operator reconciles fine but cannot report a failed
+      # reconciliation on the resource it failed on, which is the only place a
+      # user without log access sees it.
+      - 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/orchestratord/src/bin/orchestratord.rs

@@ -45,6 +45,7 @@ use mz_orchestratord::{
     controller, gcp_node_upgrade,
     k8s::{ConversionWebhookConfig, register_crds},
     metrics::{self, Metrics},
+    reconcile::{Observed, Publisher},
     tls::DefaultCertificateSpecs,
     webhook,
 };
@@ -631,6 +632,13 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
         &leader_election_identity,
     );
 
+    // Shared by all of the controllers, so that the aggregation of repeated
+    // events is too.
+    let publisher = Arc::new(Publisher::new(
+        client.clone(),
+        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.
@@ -708,10 +716,16 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             tracing: args.tracing,
             orchestratord_namespace: namespace,
         };
+        let publisher = Arc::clone(&publisher);
         move || {
             k8s_controller::Controller::namespaced_all(
                 client.clone(),
-                controller::materialize::Context::new(config.clone(), Arc::clone(&metrics)),
+                Observed::new(
+                    controller::materialize::Context::new(config.clone(), Arc::clone(&metrics)),
+                    controller::materialize::CONTROLLER_NAME,
+                    Arc::clone(&metrics.reconcile),
+                    Arc::clone(&publisher),
+                ),
                 watcher::Config::default().timeout(29),
             )
             .with_controller(|controller| {
@@ -771,10 +785,20 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             balancerd_http_port: args.balancerd_http_port,
             balancerd_internal_http_port: args.balancerd_internal_http_port,
         };
+        let reconcile_metrics = Arc::clone(&metrics.reconcile);
+        let publisher = Arc::clone(&publisher);
         move || {
             k8s_controller::Controller::namespaced_all(
                 client.clone(),
-                controller::balancer::Context::new(config.clone()),
+                Observed::new(
+                    controller::balancer::Context::new(
+                        config.clone(),
+                        Arc::clone(&reconcile_metrics),
+                    ),
+                    controller::balancer::CONTROLLER_NAME,
+                    Arc::clone(&reconcile_metrics),
+                    Arc::clone(&publisher),
+                ),
                 watcher::Config::default().timeout(29),
             )
             .with_controller(|controller| {
@@ -821,10 +845,20 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             console_http_port: args.console_http_port,
             balancerd_http_port: args.balancerd_http_port,
         };
+        let reconcile_metrics = Arc::clone(&metrics.reconcile);
+        let publisher = Arc::clone(&publisher);
         move || {
             k8s_controller::Controller::namespaced_all(
                 client.clone(),
-                controller::console::Context::new(config.clone()),
+                Observed::new(
+                    controller::console::Context::new(
+                        config.clone(),
+                        Arc::clone(&reconcile_metrics),
+                    ),
+                    controller::console::CONTROLLER_NAME,
+                    Arc::clone(&reconcile_metrics),
+                    Arc::clone(&publisher),
+                ),
                 watcher::Config::default().timeout(29),
             )
             .with_controller(|controller| {

src/orchestratord/src/controller/balancer.rs

@@ -7,6 +7,8 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::sync::Arc;
+
 use anyhow::bail;
 use k8s_controller::TraceMetadata;
 use k8s_openapi::{
@@ -36,6 +38,7 @@ use tracing::{trace, warn};
 use crate::{
     Error,
     k8s::{apply_resource, get_resource, recommended_k8s_labels, replace_resource},
+    reconcile::{self, Outcome},
     tls::{DefaultCertificateSpecs, create_certificate, issuer_ref_defined},
 };
 use mz_cloud_resources::crd::{
@@ -46,6 +49,11 @@ use mz_cloud_resources::crd::{
 use mz_orchestrator_kubernetes::KubernetesImagePullPolicy;
 use mz_ore::{cli::KeyValueArg, instrument};
 
+/// The `controller` label that this controller's metrics and Kubernetes events
+/// carry. Shared by `Observed` and by every step this controller records, which
+/// must agree for a pass and its steps to line up.
+pub const CONTROLLER_NAME: &str = "balancer";
+
 #[derive(Clone)]
 pub struct Config {
     pub enable_security_context: bool,
@@ -69,11 +77,18 @@ pub struct Config {
 
 pub struct Context {
     config: Config,
+    metrics: Arc<reconcile::Metrics>,
 }
 
 impl Context {
-    pub fn new(config: Config) -> Self {
-        Self { config }
+    pub fn new(config: Config, metrics: Arc<reconcile::Metrics>) -> Self {
+        Self { config, metrics }
+    }
+
+    /// Starts recording the step named `step` of this controller's
+    /// reconciliation.
+    fn step(&self, step: &'static str) -> reconcile::Step<'_> {
+        self.metrics.step(CONTROLLER_NAME, step)
     }
 
     async fn sync_deployment_status(
@@ -562,6 +577,7 @@ impl k8s_controller::Context for Context {
         _metadata: &mut TraceMetadata,
     ) -> Result<Option<Action>, Self::Error> {
         if balancer.status.is_none() {
+            let step = self.step("initialize_status");
             let balancer_api: Api<Balancer> =
                 Api::namespaced(client.clone(), &balancer.meta().namespace.clone().unwrap());
             let mut new_balancer = balancer.clone();
@@ -573,6 +589,7 @@ impl k8s_controller::Context for Context {
                     &new_balancer,
                 )
                 .await?;
+            step.finish(Outcome::Applied);
             // Updating the status should trigger a reconciliation
             // which will include a status this time.
             return Ok(None);
@@ -583,21 +600,31 @@ impl k8s_controller::Context for Context {
         let deployment_api: Api<Deployment> = Api::namespaced(client.clone(), &namespace);
         let service_api: Api<Service> = Api::namespaced(client.clone(), &namespace);
 
+        let step = self.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::Applied);
+        } else {
+            step.finish(Outcome::Skipped);
         }
 
+        let step = self.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::Applied);
 
+        let step = self.step("service");
         let service = self.create_service_object(balancer);
         trace!("creating new balancerd service");
         apply_resource(&service_api, &service).await?;
+        step.finish(Outcome::Applied);
 
+        let step = self.step("sync_status");
         self.sync_deployment_status(&client, balancer).await?;
+        step.finish(Outcome::Applied);
 
         Ok(None)
     }