CLO-188 orchestratord: report Materialize lifecycle transitions as events

src/orchestratord/src/bin/orchestratord.rs

@@ -721,7 +721,11 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             k8s_controller::Controller::namespaced_all(
                 client.clone(),
                 Observed::new(
-                    controller::materialize::Context::new(config.clone(), Arc::clone(&metrics)),
+                    controller::materialize::Context::new(
+                        config.clone(),
+                        Arc::clone(&metrics),
+                        Arc::clone(&publisher),
+                    ),
                     controller::materialize::CONTROLLER_NAME,
                     Arc::clone(&metrics.reconcile),
                     Arc::clone(&publisher),

src/orchestratord/src/controller/materialize.rs

@@ -25,6 +25,7 @@ use kube::{
     Api, Client, Resource, ResourceExt,
     api::{ListParams, PostParams},
     runtime::controller::Action,
+    runtime::events::{Event, EventType},
 };
 use tracing::{debug, trace, warn};
 use uuid::Uuid;
@@ -137,11 +138,16 @@ pub struct Config {
 pub struct Context {
     config: Config,
     metrics: Arc<Metrics>,
+    publisher: Arc<reconcile::Publisher>,
     needs_update: Arc<Mutex<BTreeSet<String>>>,
 }
 
 impl Context {
-    pub fn new(config: Config, metrics: Arc<Metrics>) -> Self {
+    pub fn new(
+        config: Config,
+        metrics: Arc<Metrics>,
+        publisher: Arc<reconcile::Publisher>,
+    ) -> Self {
         if config.cloud_provider == CloudProvider::Aws {
             assert!(
                 config.aws_account_id.is_some(),
@@ -152,6 +158,7 @@ impl Context {
         Self {
             config,
             metrics,
+            publisher,
             needs_update: Default::default(),
         }
     }
@@ -191,6 +198,39 @@ impl Context {
             .set(u64::cast_from(needs_update_set.len()));
     }
 
+    /// Reports a lifecycle transition as a Kubernetes event on the resource.
+    ///
+    /// The condition's own `reason` and `message` become the event's, so the
+    /// vocabulary an operator sees in `kubectl describe` is the one the status
+    /// already reports: `Applying`, `ReadyToPromote`, `Promoting`, `Applied`,
+    /// `WaitingForApproval`, `RolloutTimeout`, `FailedDeploy`. Keep those
+    /// stable, since they are what a dashboard groups on.
+    ///
+    /// A `FailedDeploy` reports here and again from the generic reconciliation
+    /// wrapper, which files the error itself. The two carry different halves of
+    /// the picture, the phase and the cause, and the reasons tell them apart.
+    ///
+    /// These do not aggregate the way repeated failures do, so each transition
+    /// is its own event carrying its own message. Two things make that so: a
+    /// transition only reports when the status actually moved, and a pass that
+    /// ends cleanly forgets what the resource published. A transition reported
+    /// from a pass that then fails is the exception, and there aggregating is
+    /// right, since the retry is reporting the same phase again.
+    async fn publish_transition(&self, mz: &Materialize, condition: &Condition) {
+        self.publisher
+            .publish(
+                mz,
+                Event {
+                    type_: transition_event_type(condition),
+                    reason: condition.reason.clone(),
+                    action: "Reconcile".into(),
+                    note: Some(condition.message.clone()),
+                    secondary: None,
+                },
+            )
+            .await;
+    }
+
     async fn update_status(
         &self,
         mz_api: &Api<Materialize>,
@@ -209,10 +249,26 @@ impl Context {
             return Ok(new_mz);
         }
 
+        // Reaching here is the definition of a transition: the guard above
+        // returns early unless the status moved, comparing everything but the
+        // timestamp. That is what keeps this to one event per transition in a
+        // reconciler that re-runs on every watch event.
+        //
+        // The exception is the pass that gives a new resource its first status,
+        // which carries no conditions yet and so reports no phase.
+        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?;
+
+        // Only once the status is durable, so that no event claims a transition
+        // that failed to persist.
+        if let Some(condition) = condition {
+            self.publish_transition(mz, &condition).await;
+        }
+
+        Ok(new_mz)
     }
 
     async fn promote(
@@ -961,6 +1017,26 @@ impl k8s_controller::Context for Context {
     }
 }
 
+/// The severity to report a lifecycle transition at.
+///
+/// `UpToDate=False` says the environment is not up to date, which is worth
+/// flagging for every reason that reports it but one. Waiting for approval is
+/// the operator doing exactly what it was configured to do, and a rollout
+/// nobody has requested yet is not a problem; reporting it as a warning would
+/// teach people to ignore warnings on Materialize resources. `Unknown` is a
+/// rollout under way, which is also not a problem.
+///
+/// A reason this does not recognize falls back to the condition's status, so a
+/// failure reason added later reports as a warning without having to be named
+/// here.
+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<Option<Action>, Error> {
     if let Some(conditions) = balancer
         .status
@@ -977,3 +1053,44 @@ fn wait_for_balancer(balancer: &Balancer) -> Result<Option<Action>, 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() {
+        // Every condition this controller writes, and how loudly it should be
+        // reported. Keep this in step with the `Condition`s built above.
+        for (status, reason, expected) in [
+            ("True", "Applied", EventType::Normal),
+            ("Unknown", "Applying", EventType::Normal),
+            ("Unknown", "ReadyToPromote", EventType::Normal),
+            ("Unknown", "Promoting", EventType::Normal),
+            // Not up to date, but by configuration rather than by failure.
+            ("False", "WaitingForApproval", EventType::Normal),
+            ("False", "FailedDeploy", EventType::Warning),
+            ("False", "RolloutTimeout", EventType::Warning),
+            // An unrecognized reason is judged by its status.
+            ("False", "SomeFutureFailure", EventType::Warning),
+            ("Unknown", "SomeFuturePhase", EventType::Normal),
+        ] {
+            assert_eq!(
+                transition_event_type(&condition(status, reason)),
+                expected,
+                "status={status} reason={reason}",
+            );
+        }
+    }
+}

src/orchestratord/src/reconcile.rs

@@ -20,7 +20,9 @@
 //! Reconcilers additionally report their own [steps](Metrics::step). A step is
 //! the only level at which "there was nothing to do" is distinguishable from
 //! "the desired state was reached", since the reconciler interface reports both
-//! as success.
+//! as success. A reconciler that tracks a lifecycle of its own reports it with
+//! [`Publisher::publish`], which is the same event machinery pointed at
+//! transitions rather than failures.
 
 use std::collections::BTreeMap;
 use std::sync::{Arc, Mutex};

test/orchestratord/mzcompose.py

@@ -3563,9 +3563,59 @@ def workflow_manually_promote(
     wait_for_rollout_complete()
     print("Test 2 PASSED: Promotion via v1 requestedRolloutHash succeeded")
 
+    # Two manual promotions have now driven the full lifecycle, so every phase
+    # of one should have been reported as an event.
+    check_lifecycle_events(definition)
     print("workflow_manually_promote PASSED")
 
 
+def check_lifecycle_events(definition: dict[str, Any]) -> None:
+    """Assert that the operator reported each rollout phase as an event.
+
+    The status conditions record only the phase the resource is in now, so the
+    events are the only account of the phases it passed through. Their reasons
+    are what a dashboard groups on, which is why this pins the exact vocabulary
+    rather than checking that any event at all was published.
+    """
+    name = definition["materialize"]["metadata"]["name"]
+    events = json.loads(
+        spawn.capture(
+            [
+                "kubectl",
+                "get",
+                # The events.k8s.io API explicitly, rather than the core v1 view
+                # `kubectl get events` serves, which renames these fields.
+                "events.events.k8s.io",
+                "-n",
+                "materialize-environment",
+                "-o",
+                "json",
+            ],
+        )
+    )["items"]
+    by_reason = {
+        event["reason"]: event
+        for event in events
+        if event.get("regarding", {}).get("kind") == "Materialize"
+        and event.get("regarding", {}).get("name") == name
+    }
+    # ManuallyPromote holds at ReadyToPromote until forcePromote is set, so a
+    # promotion under this strategy passes through all three.
+    for reason in ("Applying", "ReadyToPromote", "Promoting", "Applied"):
+        assert reason in by_reason, (
+            f"Expected a {reason} event on Materialize/{name}, "
+            f"found {sorted(by_reason)}"
+        )
+        event = by_reason[reason]
+        assert event["type"] == "Normal", event
+        assert event["reportingController"] == "orchestratord.materialize.cloud", event
+        assert event.get("note"), event
+    # The promotion events should name the generation they promoted, which is
+    # what makes them readable against a rollout after the fact.
+    applied = by_reason["Applied"]["note"]
+    assert "generation" in applied, applied
+
+
 def apply_materialize(definition: dict[str, Any]) -> None:
     """Apply the materialize resource definition via kubectl."""
     defs = [