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..436c52d2c92da 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,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, @@ -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, Error> { if let Some(conditions) = balancer .status @@ -977,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}", + ); + } + } +} 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 = [