From 493ffb7256d1cb6324b277e1c845f6ce8815924e Mon Sep 17 00:00:00 2001 From: Heather Lapointe Date: Thu, 27 Aug 2026 16:14:02 -0400 Subject: [PATCH 1/4] orchestratord: report Materialize lifecycle transitions as events The status conditions record only the phase an environment is in now, so a rollout that has moved on leaves no account of the phases it passed through. Report each transition as a Kubernetes event, which the monitoring stack persists as a structured log. `update_status` is the single place a condition is written, and it already returns early unless the status substantively moved, comparing everything but the timestamp. That makes it an exact edge detector, so a reconciler that re-runs on every watch event still files one event per transition without tracking any state of its own. The event carries the condition's own reason and message, so `Applying`, `ReadyToPromote`, `Promoting`, `Applied`, `WaitingForApproval`, `RolloutTimeout` and `FailedDeploy` are one vocabulary across the status, `kubectl describe`, and a dashboard. Publishing happens after the status write lands, so no event claims a transition that failed to persist. Failures now report twice, once here with the phase and once from the reconciliation wrapper with the cause; the reasons tell them apart. `workflow_manually_promote` drives a promotion through every phase, so it asserts the vocabulary rather than that any event at all was published. Co-Authored-By: Claude Opus 5 --- src/orchestratord/src/bin/orchestratord.rs | 6 +- .../src/controller/materialize.rs | 69 ++++++++++++++++++- src/orchestratord/src/reconcile.rs | 4 +- test/orchestratord/mzcompose.py | 50 ++++++++++++++ 4 files changed, 124 insertions(+), 5 deletions(-) diff --git a/src/orchestratord/src/bin/orchestratord.rs b/src/orchestratord/src/bin/orchestratord.rs index c891a72ecd1a5..7db348972b73b 100644 --- a/src/orchestratord/src/bin/orchestratord.rs +++ b/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), diff --git a/src/orchestratord/src/controller/materialize.rs b/src/orchestratord/src/controller/materialize.rs index bf44cde278112..8483ac5e77b8b 100644 --- a/src/orchestratord/src/controller/materialize.rs +++ b/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, + publisher: Arc, needs_update: Arc>>, } impl Context { - pub fn new(config: Config, metrics: Arc) -> Self { + pub fn new( + config: Config, + metrics: Arc, + publisher: Arc, + ) -> 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,46 @@ 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 { + // A condition that went false is the environment failing to + // reach the state it was asked for, which is worth flagging. + // `Unknown` is a rollout still under way, which is not. + type_: if condition.status == "False" { + EventType::Warning + } else { + EventType::Normal + }, + reason: condition.reason.clone(), + action: "Reconcile".into(), + note: Some(condition.message.clone()), + secondary: None, + }, + ) + .await; + } + async fn update_status( &self, mz_api: &Api, @@ -209,10 +256,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( diff --git a/src/orchestratord/src/reconcile.rs b/src/orchestratord/src/reconcile.rs index 2439193c06b91..f286689ce834a 100644 --- a/src/orchestratord/src/reconcile.rs +++ b/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}; diff --git a/test/orchestratord/mzcompose.py b/test/orchestratord/mzcompose.py index 7fb5b7d82d879..57f245654ab52 100644 --- a/test/orchestratord/mzcompose.py +++ b/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 = [ From 3918cf6beeab829fe73062427a5c40b429cb4476 Mon Sep 17 00:00:00 2001 From: Heather Lapointe Date: Fri, 28 Aug 2026 09:16:13 -0400 Subject: [PATCH 2/4] Report waiting for approval as a normal event, not a warning The severity of a lifecycle event was taken from the condition's status, so every reason carrying `UpToDate=False` reported as a warning. Three reasons carry it: `FailedDeploy` and `RolloutTimeout`, which are failures, and `WaitingForApproval`, which is not. That last one is the operator doing exactly what it was configured to do, holding a detected change until someone requests the rollout. Flagging it teaches people to ignore warnings on Materialize resources, which costs them the two that mean something. The rule moves into `transition_event_type`, which special-cases that reason and otherwise falls back to the status, so a failure reason added later still reports as a warning without having to be named there. A unit test pins the severity of every condition the controller writes. Co-Authored-By: Claude Opus 5 --- .../src/controller/materialize.rs | 70 ++++++++++++++++--- 1 file changed, 62 insertions(+), 8 deletions(-) diff --git a/src/orchestratord/src/controller/materialize.rs b/src/orchestratord/src/controller/materialize.rs index 8483ac5e77b8b..436c52d2c92da 100644 --- a/src/orchestratord/src/controller/materialize.rs +++ b/src/orchestratord/src/controller/materialize.rs @@ -221,14 +221,7 @@ impl Context { .publish( mz, Event { - // A condition that went false is the environment failing to - // reach the state it was asked for, which is worth flagging. - // `Unknown` is a rollout still under way, which is not. - type_: if condition.status == "False" { - EventType::Warning - } else { - EventType::Normal - }, + type_: transition_event_type(condition), reason: condition.reason.clone(), action: "Reconcile".into(), note: Some(condition.message.clone()), @@ -1024,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, Error> { if let Some(conditions) = balancer .status @@ -1040,3 +1053,44 @@ 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() { + // 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}", + ); + } + } +} From 8698e04e4b3d0c64d0e8fcaeb97801dfdb357ec8 Mon Sep 17 00:00:00 2001 From: Heather Lapointe Date: Tue, 1 Sep 2026 09:47:00 -0400 Subject: [PATCH 3/4] Clean up publish --- src/orchestratord/src/reconcile.rs | 78 +++++++++++++++++++++++------- 1 file changed, 61 insertions(+), 17 deletions(-) diff --git a/src/orchestratord/src/reconcile.rs b/src/orchestratord/src/reconcile.rs index f286689ce834a..09da24275f36c 100644 --- a/src/orchestratord/src/reconcile.rs +++ b/src/orchestratord/src/reconcile.rs @@ -21,8 +21,8 @@ //! 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. 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. +//! [`publish`], which is the same event machinery pointed at transitions rather +//! than failures. use std::collections::BTreeMap; use std::sync::{Arc, Mutex}; @@ -310,21 +310,27 @@ where resource: &Ctx::Resource, error: &Ctx::Error, ) { - self.publisher - .publish( - resource, - Event { - type_: EventType::Warning, - reason: RECONCILIATION_FAILED.into(), - action: entry_point.action().into(), - // The cause chain, not just the outermost message: an error - // like "invalid environment id in license key" is only - // actionable alongside what it was that failed to parse. - note: Some(error.to_string_with_causes()), - secondary: None, - }, - ) - .await; + let event = Event { + type_: EventType::Warning, + reason: RECONCILIATION_FAILED.into(), + action: entry_point.action().into(), + // The cause chain, not just the outermost message: an error like + // "invalid environment id in license key" is only actionable + // alongside what it was that failed to parse. + note: Some(truncate_note(&error.to_string_with_causes())), + secondary: None, + }; + if let Err(e) = self + .recorder + .publish(&event, &resource.object_ref(&Default::default())) + .await + { + warn!( + error = %e, + controller = self.controller, + "failed to publish reconciliation failure event", + ); + } } /// Records what one reconciliation pass did, and reports a failure on the @@ -401,6 +407,44 @@ where // not, gets the crate's default backoff. } +/// Publishes `event` as a Kubernetes event on `resource`, trimming its note to +/// what the API server accepts. +/// +/// Failing to publish is only logged. An event reports something that already +/// happened, so failing to file it must not change what the caller goes on to +/// do; a missing `events.k8s.io` permission costs visibility, not +/// reconciliation. +/// +/// NOTE: `recorder` collapses repeats of the same `(type, action, reason, +/// regarding)` within six minutes into one event carrying a count, and a repeat +/// keeps the note the first one had. That is what stops a reconciliation +/// retrying on a backoff from filing an event per attempt. A caller reporting +/// transitions rather than retries pays for it in the other direction: two +/// transitions that share a reason inside that window report the earlier one's +/// note. The reason and the count stay correct, so what a dashboard groups on +/// survives; only the human-readable detail goes stale. +pub async fn publish(recorder: &Recorder, resource: &K, mut event: Event) +where + K: Resource, + K::DynamicType: Default, +{ + if let Some(note) = event.note.take() { + event.note = Some(truncate_note(¬e)); + } + let reason = event.reason.clone(); + if let Err(e) = recorder + .publish(&event, &resource.object_ref(&Default::default())) + .await + { + warn!( + error = %e, + reason = %reason, + kind = %K::kind(&Default::default()), + "failed to publish event", + ); + } +} + /// Publishes Kubernetes events on the resources a controller reconciles. /// /// Kubernetes expects repeats of an event to aggregate into one object carrying From fa572cba0178edfc989cf7a15f24823a7331da81 Mon Sep 17 00:00:00 2001 From: Heather Lapointe Date: Tue, 8 Sep 2026 11:22:54 -0400 Subject: [PATCH 4/4] Revert "Clean up publish" This reverts commit 8698e04e4b3d0c64d0e8fcaeb97801dfdb357ec8, which does not build: it points `Observed::publish_failure` back at a `recorder` field that no longer exists, since the aggregation fix replaced it with `publisher`. The commit reapplies the free `publish` function verbatim, stale doc comment included, and leaves it beside the `Publisher` that superseded it with no callers. That is the rebase conflict resolved toward the wrong side rather than an intended change, so reverting it restores the file to its parent exactly. Reinstating that path would also undo the fix for the aggregation bug QA reported on the parent branch: publishing every failure through one shared recorder pins a failing resource's event note to whichever error came first, for as long as the failure lasts. Co-Authored-By: Claude Opus 5 --- src/orchestratord/src/reconcile.rs | 78 +++++++----------------------- 1 file changed, 17 insertions(+), 61 deletions(-) diff --git a/src/orchestratord/src/reconcile.rs b/src/orchestratord/src/reconcile.rs index 09da24275f36c..f286689ce834a 100644 --- a/src/orchestratord/src/reconcile.rs +++ b/src/orchestratord/src/reconcile.rs @@ -21,8 +21,8 @@ //! 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. A reconciler that tracks a lifecycle of its own reports it with -//! [`publish`], which is the same event machinery pointed at transitions rather -//! than failures. +//! [`Publisher::publish`], which is the same event machinery pointed at +//! transitions rather than failures. use std::collections::BTreeMap; use std::sync::{Arc, Mutex}; @@ -310,27 +310,21 @@ where resource: &Ctx::Resource, error: &Ctx::Error, ) { - let event = Event { - type_: EventType::Warning, - reason: RECONCILIATION_FAILED.into(), - action: entry_point.action().into(), - // The cause chain, not just the outermost message: an error like - // "invalid environment id in license key" is only actionable - // alongside what it was that failed to parse. - note: Some(truncate_note(&error.to_string_with_causes())), - secondary: None, - }; - if let Err(e) = self - .recorder - .publish(&event, &resource.object_ref(&Default::default())) - .await - { - warn!( - error = %e, - controller = self.controller, - "failed to publish reconciliation failure event", - ); - } + self.publisher + .publish( + resource, + Event { + type_: EventType::Warning, + reason: RECONCILIATION_FAILED.into(), + action: entry_point.action().into(), + // The cause chain, not just the outermost message: an error + // like "invalid environment id in license key" is only + // actionable alongside what it was that failed to parse. + note: Some(error.to_string_with_causes()), + secondary: None, + }, + ) + .await; } /// Records what one reconciliation pass did, and reports a failure on the @@ -407,44 +401,6 @@ where // not, gets the crate's default backoff. } -/// Publishes `event` as a Kubernetes event on `resource`, trimming its note to -/// what the API server accepts. -/// -/// Failing to publish is only logged. An event reports something that already -/// happened, so failing to file it must not change what the caller goes on to -/// do; a missing `events.k8s.io` permission costs visibility, not -/// reconciliation. -/// -/// NOTE: `recorder` collapses repeats of the same `(type, action, reason, -/// regarding)` within six minutes into one event carrying a count, and a repeat -/// keeps the note the first one had. That is what stops a reconciliation -/// retrying on a backoff from filing an event per attempt. A caller reporting -/// transitions rather than retries pays for it in the other direction: two -/// transitions that share a reason inside that window report the earlier one's -/// note. The reason and the count stay correct, so what a dashboard groups on -/// survives; only the human-readable detail goes stale. -pub async fn publish(recorder: &Recorder, resource: &K, mut event: Event) -where - K: Resource, - K::DynamicType: Default, -{ - if let Some(note) = event.note.take() { - event.note = Some(truncate_note(¬e)); - } - let reason = event.reason.clone(); - if let Err(e) = recorder - .publish(&event, &resource.object_ref(&Default::default())) - .await - { - warn!( - error = %e, - reason = %reason, - kind = %K::kind(&Default::default()), - "failed to publish event", - ); - } -} - /// Publishes Kubernetes events on the resources a controller reconciles. /// /// Kubernetes expects repeats of an event to aggregate into one object carrying