Skip to main content

mz_orchestratord/controller/
materialize.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::{
11    collections::BTreeSet,
12    sync::{Arc, Mutex},
13    time::Duration,
14};
15
16use anyhow::Context as _;
17use http::HeaderValue;
18use k8s_controller::TraceMetadata;
19use k8s_openapi::{
20    api::core::v1::{Affinity, ResourceRequirements, Secret, Toleration},
21    apimachinery::pkg::apis::meta::v1::{Condition, Time},
22    jiff::{SignedDuration, Timestamp},
23};
24use kube::{
25    Api, Client, Resource, ResourceExt,
26    api::{ListParams, PostParams},
27    runtime::controller::Action,
28};
29use tracing::{debug, trace, warn};
30use uuid::Uuid;
31
32use crate::{
33    Error,
34    controller::materialize::generation::V161,
35    k8s::{apply_resource, delete_resource},
36    matching_image_from_environmentd_image_ref,
37    metrics::Metrics,
38    parse_image_tag,
39    tls::{DefaultCertificateSpecs, issuer_ref_defined, resolved_dns_names},
40};
41use mz_cloud_provider::CloudProvider;
42use mz_cloud_resources::crd::{
43    ManagedResource,
44    balancer::v1alpha1::{Balancer, BalancerSpec},
45    console::v1alpha1::{BalancerdRef, Console, ConsoleSpec, HttpConnectionScheme},
46    materialize::MaterializeRolloutStrategy,
47    materialize::v1alpha1::{Materialize, MaterializeStatus},
48};
49use mz_license_keys::validate;
50use mz_orchestrator_kubernetes::KubernetesImagePullPolicy;
51use mz_orchestrator_tracing::TracingCliArgs;
52use mz_ore::{cast::CastFrom, cli::KeyValueArg, instrument};
53
54pub mod generation;
55pub mod global;
56
57#[derive(Clone)]
58pub struct Config {
59    pub cloud_provider: CloudProvider,
60    pub region: String,
61    pub create_balancers: bool,
62    pub create_console: bool,
63    pub helm_chart_version: Option<String>,
64    pub secrets_controller: String,
65    pub collect_pod_metrics: bool,
66    pub enable_prometheus_scrape_annotations: bool,
67
68    pub segment_api_key: Option<String>,
69    pub segment_client_side: bool,
70
71    pub console_image_tag_default: String,
72    pub console_image_tag_map: Vec<KeyValueArg<String, String>>,
73
74    pub aws_account_id: Option<String>,
75    pub environmentd_iam_role_arn: Option<String>,
76    pub environmentd_connection_role_arn: Option<String>,
77    pub aws_secrets_controller_tags: Vec<String>,
78    pub environmentd_availability_zones: Option<Vec<String>>,
79
80    pub ephemeral_volume_class: Option<String>,
81    pub scheduler_name: Option<String>,
82    pub enable_security_context: bool,
83    pub enable_internal_statement_logging: bool,
84    pub disable_statement_logging: bool,
85
86    pub orchestratord_pod_selector_labels: Vec<KeyValueArg<String, String>>,
87    pub environmentd_node_selector: Vec<KeyValueArg<String, String>>,
88    pub environmentd_affinity: Option<Affinity>,
89    pub environmentd_tolerations: Option<Vec<Toleration>>,
90    pub environmentd_default_resources: Option<ResourceRequirements>,
91    pub clusterd_node_selector: Vec<KeyValueArg<String, String>>,
92    pub clusterd_affinity: Option<Affinity>,
93    pub clusterd_tolerations: Option<Vec<Toleration>>,
94    pub image_pull_policy: KubernetesImagePullPolicy,
95    pub network_policies_internal_enabled: bool,
96    pub network_policies_ingress_enabled: bool,
97    pub network_policies_ingress_cidrs: Vec<String>,
98    pub network_policies_egress_enabled: bool,
99    pub network_policies_egress_cidrs: Vec<String>,
100
101    pub environmentd_cluster_replica_sizes: Option<String>,
102    pub bootstrap_default_cluster_replica_size: Option<String>,
103    pub bootstrap_builtin_system_cluster_replica_size: Option<String>,
104    pub bootstrap_builtin_probe_cluster_replica_size: Option<String>,
105    pub bootstrap_builtin_support_cluster_replica_size: Option<String>,
106    pub bootstrap_builtin_catalog_server_cluster_replica_size: Option<String>,
107    pub bootstrap_builtin_analytics_cluster_replica_size: Option<String>,
108    pub bootstrap_builtin_system_cluster_replication_factor: Option<u32>,
109    pub bootstrap_builtin_probe_cluster_replication_factor: Option<u32>,
110    pub bootstrap_builtin_support_cluster_replication_factor: Option<u32>,
111    pub bootstrap_builtin_analytics_cluster_replication_factor: Option<u32>,
112
113    pub environmentd_allowed_origins: Vec<HeaderValue>,
114    pub internal_console_proxy_url: String,
115
116    pub environmentd_sql_port: u16,
117    pub environmentd_http_port: u16,
118    pub environmentd_internal_sql_port: u16,
119    pub environmentd_internal_http_port: u16,
120    pub environmentd_internal_persist_pubsub_port: u16,
121
122    pub default_certificate_specs: DefaultCertificateSpecs,
123
124    pub disable_license_key_checks: bool,
125
126    pub tracing: TracingCliArgs,
127    pub orchestratord_namespace: String,
128}
129
130pub struct Context {
131    config: Config,
132    metrics: Arc<Metrics>,
133    needs_update: Arc<Mutex<BTreeSet<String>>>,
134}
135
136impl Context {
137    pub fn new(config: Config, metrics: Arc<Metrics>) -> Self {
138        if config.cloud_provider == CloudProvider::Aws {
139            assert!(
140                config.aws_account_id.is_some(),
141                "--aws-account-id is required when using --cloud-provider=aws"
142            );
143        }
144
145        Self {
146            config,
147            metrics,
148            needs_update: Default::default(),
149        }
150    }
151
152    fn set_needs_update(&self, mz: &Materialize, needs_update: bool) {
153        let mut needs_update_set = self.needs_update.lock().unwrap();
154        if needs_update {
155            needs_update_set.insert(mz.name_unchecked());
156        } else {
157            needs_update_set.remove(&mz.name_unchecked());
158        }
159        self.metrics
160            .environmentd_needs_update
161            .set(u64::cast_from(needs_update_set.len()));
162    }
163
164    async fn update_status(
165        &self,
166        mz_api: &Api<Materialize>,
167        mz: &Materialize,
168        status: MaterializeStatus,
169        needs_update: bool,
170    ) -> Result<Materialize, kube::Error> {
171        self.set_needs_update(mz, needs_update);
172
173        let mut new_mz = mz.clone();
174        if !mz
175            .status
176            .as_ref()
177            .map_or(true, |mz_status| mz_status.needs_update(&status))
178        {
179            return Ok(new_mz);
180        }
181
182        new_mz.status = Some(status);
183        mz_api
184            .replace_status(&mz.name_unchecked(), &PostParams::default(), &new_mz)
185            .await
186    }
187
188    async fn promote(
189        &self,
190        client: &Client,
191        mz: &Materialize,
192        resources: generation::Resources,
193        active_generation: u64,
194        desired_generation: u64,
195        resources_hash: String,
196    ) -> Result<Option<Action>, Error> {
197        if let Some(action) = resources.promote_services(client, &mz.namespace()).await? {
198            return Ok(Some(action));
199        }
200        resources
201            .teardown_generation(client, mz, active_generation)
202            .await?;
203        let mz_api: Api<Materialize> = Api::namespaced(client.clone(), &mz.namespace());
204        self.update_status(
205            &mz_api,
206            mz,
207            MaterializeStatus {
208                active_generation: desired_generation,
209                last_completed_rollout_request: mz.requested_reconciliation_id(),
210                last_completed_rollout_environmentd_image_ref: Some(
211                    mz.spec.environmentd_image_ref.clone(),
212                ),
213                resource_id: mz.status().resource_id,
214                resources_hash,
215                last_completed_rollout_hash: None,
216                conditions: vec![Condition {
217                    type_: "UpToDate".into(),
218                    status: "True".into(),
219                    last_transition_time: Time(Timestamp::now()),
220                    message: format!(
221                        "Successfully applied changes for generation {desired_generation}"
222                    ),
223                    observed_generation: mz.meta().generation,
224                    reason: "Applied".into(),
225                }],
226            },
227            false,
228        )
229        .await?;
230        Ok(None)
231    }
232
233    async fn check_environment_id_conflicts(
234        &self,
235        client: &Client,
236        mz: &Materialize,
237    ) -> Result<(), Error> {
238        if mz.spec.environment_id.is_nil() {
239            // this is always a bug - we delay doing this check until the
240            // resource should have an environment id set, either from the
241            // license key, or explicitly given, or randomly defaulted.
242            return Err(Error::Anyhow(anyhow::anyhow!(
243                "trying to reconcile a materialize resource with no environment id - this is a bug!"
244            )));
245        }
246
247        let mz_api: Api<Materialize> = Api::all(client.clone());
248        let all_mz = mz_api.list(&ListParams::default()).await?;
249        for existing_mz in &all_mz.items {
250            if existing_mz.spec.environment_id == mz.spec.environment_id
251                && existing_mz.metadata.uid != mz.metadata.uid
252            {
253                return Err(Error::Anyhow(anyhow::anyhow!(
254                    "Materialize resources {}/{} and {}/{} have the environmentId field set to the same value. This field must be unique across environments.",
255                    mz.namespace(),
256                    mz.name_unchecked(),
257                    existing_mz.namespace(),
258                    existing_mz.name_unchecked(),
259                )));
260            }
261        }
262
263        Ok(())
264    }
265}
266
267#[async_trait::async_trait]
268impl k8s_controller::Context for Context {
269    type Resource = Materialize;
270    type Error = Error;
271
272    const FINALIZER_NAME: Option<&'static str> =
273        Some("orchestratord.materialize.cloud/materialize");
274
275    #[instrument(fields(organization_name=mz.name_unchecked()))]
276    async fn apply(
277        &self,
278        client: Client,
279        mz: &Self::Resource,
280        _metadata: &mut TraceMetadata,
281    ) -> Result<Option<Action>, Self::Error> {
282        let mz_api: Api<Materialize> = Api::namespaced(client.clone(), &mz.namespace());
283        let balancer_api: Api<Balancer> = Api::namespaced(client.clone(), &mz.namespace());
284        let console_api: Api<Console> = Api::namespaced(client.clone(), &mz.namespace());
285        let secret_api: Api<Secret> = Api::namespaced(client.clone(), &mz.namespace());
286
287        let status = mz.status();
288        if mz.status.is_none() {
289            self.update_status(&mz_api, mz, status, true).await?;
290            // Updating the status should trigger a reconciliation
291            // which will include a status this time.
292            return Ok(None);
293        }
294
295        let backend_secret = secret_api.get(&mz.spec.backend_secret_name).await?;
296        let license_key_environment_id: Option<Uuid> = if let Some(license_key) = backend_secret
297            .data
298            .as_ref()
299            .and_then(|data| data.get("license_key"))
300        {
301            let license_key = validate(
302                str::from_utf8(&license_key.0)
303                    .context("invalid utf8")?
304                    .trim(),
305            )?;
306            let environment_id = license_key
307                .environment_id
308                .parse()
309                .context("invalid environment id in license key")?;
310            Some(environment_id)
311        } else {
312            if mz.meets_minimum_version(&V161) {
313                return Err(Error::Anyhow(anyhow::anyhow!(
314                    "license_key is required when running in kubernetes",
315                )));
316            } else {
317                None
318            }
319        };
320
321        if mz.spec.request_rollout.is_nil() || mz.spec.environment_id.is_nil() {
322            let mut mz = mz.clone();
323            if mz.spec.request_rollout.is_nil() {
324                mz.spec.request_rollout = Uuid::new_v4();
325            }
326            if mz.spec.environment_id.is_nil() {
327                if let Some(environment_id) = license_key_environment_id {
328                    if environment_id.is_nil() {
329                        // this makes it easier to use a license key in
330                        // development with no environment id set
331                        mz.spec.environment_id = Uuid::new_v4();
332                    } else {
333                        mz.spec.environment_id = environment_id;
334                    }
335                } else {
336                    if mz.meets_minimum_version(&V161) {
337                        return Err(Error::Anyhow(anyhow::anyhow!(
338                            "environmentId is not set in materialize resource {}/{} but no license key was given",
339                            mz.namespace(),
340                            mz.name_unchecked()
341                        )));
342                    } else {
343                        mz.spec.environment_id = Uuid::new_v4();
344                    }
345                }
346            }
347            mz_api
348                .replace(&mz.name_unchecked(), &PostParams::default(), &mz)
349                .await?;
350            // Updating the spec should also trigger a reconciliation.
351            // We can't do that as part of the above check because you can't
352            // update both the spec and the status in a single api call.
353            return Ok(None);
354        }
355
356        if let Some(environment_id) = license_key_environment_id {
357            // we still allow a nil environment id in the license key to be
358            // accepted for any provided environment id, to support cloud
359            if !environment_id.is_nil() && mz.spec.environment_id != environment_id {
360                return Err(Error::Anyhow(anyhow::anyhow!(
361                    "environment_id is set in materialize resource {}/{} but does not match the environment_id set in the associated license key {}",
362                    mz.namespace(),
363                    mz.name_unchecked(),
364                    environment_id,
365                )));
366            }
367        }
368
369        self.check_environment_id_conflicts(&client, mz).await?;
370
371        global::Resources::new(&self.config, mz)?
372            .apply(&client, &mz.namespace())
373            .await?;
374
375        // we compare the hash against the environment resources generated
376        // for the current active generation, since that's what we expect to
377        // have been applied earlier, but we don't want to use these
378        // environment resources because when we apply them, we want to apply
379        // them with data that uses the new generation
380        let active_resources =
381            generation::Resources::new(&self.config, mz, status.active_generation);
382        let has_current_changes = status.resources_hash != active_resources.generate_hash();
383        let active_generation = status.active_generation;
384        let next_generation = active_generation + 1;
385        let desired_generation = if has_current_changes {
386            next_generation
387        } else {
388            active_generation
389        };
390
391        // here we regenerate the environment resources using the
392        // same inputs except with an updated generation
393        let resources = generation::Resources::new(&self.config, mz, desired_generation);
394        let resources_hash = resources.generate_hash();
395
396        let mut result = match (
397            mz.is_promoting(),
398            has_current_changes,
399            mz.rollout_requested(),
400        ) {
401            // If we're in status promoting, we MUST promote now.
402            // We don't know if we successfully promoted or not yet.
403            (true, _, _) => {
404                self.promote(
405                    &client,
406                    mz,
407                    resources,
408                    active_generation,
409                    desired_generation,
410                    resources_hash,
411                )
412                .await
413            }
414            // There are changes pending, and we want to apply them.
415            (false, true, true) => {
416                // If a rollout has been in progress for longer than the
417                // configured timeout, cancel it. While a rollout is in
418                // progress the new generation runs un-promoted and holds back
419                // compaction via read holds; promoting it after a long delay
420                // can cause incident-inducing load, so we abort instead and
421                // let the user retry by requesting a fresh rollout.
422                //
423                // We never cancel a force-promoting rollout (including the
424                // `ImmediatelyPromoteCausingDowntime` strategy), because by
425                // then the previously-active generation may already be torn
426                // down, leaving nothing to fall back to.
427                if !mz.should_force_promote() {
428                    if let Some(started) = mz.rollout_in_progress_since() {
429                        let timeout = mz.rollout_request_timeout();
430                        let elapsed = Timestamp::now().duration_since(started);
431                        let timed_out = SignedDuration::try_from(timeout)
432                            .is_ok_and(|timeout| elapsed >= timeout);
433                        if timed_out {
434                            warn!(
435                                "rollout to generation {desired_generation} exceeded timeout, cancelling"
436                            );
437                            // Tear down the un-promoted generation to release
438                            // its read holds.
439                            resources
440                                .teardown_generation(&client, mz, next_generation)
441                                .await?;
442                            self.update_status(
443                                &mz_api,
444                                mz,
445                                MaterializeStatus {
446                                    active_generation,
447                                    // Mark this rollout request as completed so
448                                    // that we don't immediately retry it; the
449                                    // user must request a new rollout to try
450                                    // again.
451                                    last_completed_rollout_request: mz
452                                        .requested_reconciliation_id(),
453                                    last_completed_rollout_environmentd_image_ref: status
454                                        .last_completed_rollout_environmentd_image_ref
455                                        .clone(),
456                                    resource_id: status.resource_id.clone(),
457                                    resources_hash: status.resources_hash.clone(),
458                                    last_completed_rollout_hash: None,
459                                    conditions: vec![Condition {
460                                        type_: "UpToDate".into(),
461                                        status: "False".into(),
462                                        last_transition_time: Time(Timestamp::now()),
463                                        message: format!(
464                                            "Cancelled rollout to generation \
465                                             {desired_generation} after it \
466                                             exceeded the rollout timeout of {}",
467                                            humantime::format_duration(timeout),
468                                        ),
469                                        observed_generation: mz.meta().generation,
470                                        reason: "RolloutTimeout".into(),
471                                    }],
472                                },
473                                active_generation != desired_generation,
474                            )
475                            .await?;
476                            return Ok(None);
477                        }
478                    }
479                }
480
481                if !mz.within_upgrade_window() {
482                    let last_completed_rollout_environmentd_image_ref =
483                        status.last_completed_rollout_environmentd_image_ref;
484
485                    self.update_status(
486                        &mz_api,
487                        mz,
488                        MaterializeStatus {
489                            active_generation,
490                            last_completed_rollout_request: status.last_completed_rollout_request,
491                            last_completed_rollout_environmentd_image_ref:
492                                last_completed_rollout_environmentd_image_ref.clone(),
493                            resource_id: status.resource_id,
494                            resources_hash: status.resources_hash,
495                            last_completed_rollout_hash: None,
496                            conditions: vec![Condition {
497                                type_: "UpToDate".into(),
498                                status: "False".into(),
499                                last_transition_time: Time(Timestamp::now()),
500                                message: format!(
501                                    "Refusing to upgrade from {} to {}. \
502                                     More than one major version from \
503                                     last successful rollout. If coming \
504                                     from Self Managed 25.2, upgrade to \
505                                     materialize/environmentd:v0.147.20 \
506                                     first.",
507                                    last_completed_rollout_environmentd_image_ref
508                                        .expect("should be set if upgrade window check fails"),
509                                    &mz.spec.environmentd_image_ref,
510                                ),
511                                observed_generation: mz.meta().generation,
512                                reason: "FailedDeploy".into(),
513                            }],
514                        },
515                        active_generation != desired_generation,
516                    )
517                    .await?;
518                    return Ok(None);
519                }
520
521                // we remove the environment resources hash annotation here
522                // because if we fail halfway through applying the resources,
523                // things will be in an inconsistent state, and we don't want
524                // to allow the possibility of the user making a second
525                // change which reverts to the original state and then
526                // skipping retrying this apply, since that would leave
527                // things in a permanently inconsistent state.
528                // note that environment.spec will be empty here after
529                // replace_status, but this is fine because we already
530                // extracted all of the information we want from the spec
531                // earlier.
532                let mz = if mz.is_ready_to_promote(&resources_hash) {
533                    mz
534                } else {
535                    &self
536                        .update_status(
537                            &mz_api,
538                            mz,
539                            MaterializeStatus {
540                                active_generation,
541                                // don't update the reconciliation id yet,
542                                // because the rollout hasn't yet completed. if
543                                // we fail later on, we want to ensure that the
544                                // rollout gets retried.
545                                last_completed_rollout_request: status
546                                    .last_completed_rollout_request,
547                                last_completed_rollout_environmentd_image_ref: status
548                                    .last_completed_rollout_environmentd_image_ref,
549                                resource_id: status.resource_id.clone(),
550                                resources_hash: String::new(),
551                                last_completed_rollout_hash: None,
552                                conditions: vec![Condition {
553                                    type_: "UpToDate".into(),
554                                    status: "Unknown".into(),
555                                    last_transition_time: Time(Timestamp::now()),
556                                    message: format!(
557                                        "Applying changes for generation {desired_generation}"
558                                    ),
559                                    observed_generation: mz.meta().generation,
560                                    reason: "Applying".into(),
561                                }],
562                            },
563                            active_generation != desired_generation,
564                        )
565                        .await?
566                };
567                let status = mz.status();
568
569                if mz.spec.rollout_strategy
570                    == MaterializeRolloutStrategy::ImmediatelyPromoteCausingDowntime
571                {
572                    // The only reason someone would choose this strategy is if they didn't have
573                    // space for the two generations of pods.
574                    // Lets make room for the new ones by deleting the old generation.
575                    resources
576                        .teardown_generation(&client, mz, active_generation)
577                        .await?;
578                }
579
580                trace!("applying environment resources");
581                match resources
582                    .apply(&client, mz.should_force_promote(), &mz.namespace())
583                    .await
584                {
585                    Ok(Some(action)) => {
586                        trace!("new environment is not yet ready");
587                        Ok(Some(action))
588                    }
589                    Ok(None) => {
590                        if mz.spec.rollout_strategy == MaterializeRolloutStrategy::ManuallyPromote
591                            && !mz.should_force_promote()
592                        {
593                            trace!(
594                                "Ready to promote, but not promoting because the instance is configured with ManuallyPromote rollout strategy."
595                            );
596                            self.update_status(
597                                &mz_api,
598                                mz,
599                                MaterializeStatus {
600                                    active_generation,
601                                    last_completed_rollout_request: status
602                                        .last_completed_rollout_request,
603                                    last_completed_rollout_environmentd_image_ref: status
604                                        .last_completed_rollout_environmentd_image_ref,
605                                    resource_id: status.resource_id,
606                                    resources_hash,
607                                    last_completed_rollout_hash: None,
608                                    conditions: vec![Condition {
609                                        type_: "UpToDate".into(),
610                                        status: "Unknown".into(),
611                                        // Carry the `Applying` phase's
612                                        // timestamp forward (both phases are
613                                        // `Unknown`) so the rollout timeout
614                                        // spans Applying + ReadyToPromote
615                                        // rather than resetting here.
616                                        last_transition_time: Time(mz.up_to_date_transition_time(
617                                            "Unknown",
618                                            Timestamp::now(),
619                                        )),
620                                        message: format!(
621                                            "Ready to promote generation {desired_generation}"
622                                        ),
623                                        observed_generation: mz.meta().generation,
624                                        reason: "ReadyToPromote".into(),
625                                    }],
626                                },
627                                active_generation != desired_generation,
628                            )
629                            .await?;
630                            return Ok(None);
631                        }
632                        // do this last, so that we keep traffic pointing at
633                        // the previous environmentd until the new one is
634                        // fully ready
635
636                        // Update the status before calling promote, so that we know
637                        // we've crossed the point of no return.
638                        // Once we see this status, we must promote without taking other actions.
639                        self.update_status(
640                            &mz_api,
641                            mz,
642                            MaterializeStatus {
643                                active_generation,
644                                // don't update the reconciliation id yet,
645                                // because the rollout hasn't yet completed. if
646                                // we fail later on, we want to ensure that the
647                                // rollout gets retried.
648                                last_completed_rollout_request: status
649                                    .last_completed_rollout_request,
650                                last_completed_rollout_environmentd_image_ref: status
651                                    .last_completed_rollout_environmentd_image_ref,
652                                resource_id: status.resource_id,
653                                resources_hash: resources_hash.clone(),
654                                last_completed_rollout_hash: None,
655                                conditions: vec![Condition {
656                                    type_: "UpToDate".into(),
657                                    status: "Unknown".into(),
658                                    last_transition_time: Time(Timestamp::now()),
659                                    message: format!(
660                                        "Attempting to promote generation {desired_generation}"
661                                    ),
662                                    observed_generation: mz.meta().generation,
663                                    reason: "Promoting".into(),
664                                }],
665                            },
666                            active_generation != desired_generation,
667                        )
668                        .await?;
669                        self.promote(
670                            &client,
671                            mz,
672                            resources,
673                            active_generation,
674                            desired_generation,
675                            resources_hash,
676                        )
677                        .await
678                    }
679                    Err(e) => {
680                        self.update_status(
681                            &mz_api,
682                            mz,
683                            MaterializeStatus {
684                                active_generation,
685                                // also don't update the reconciliation id
686                                // here, because there was an error during
687                                // the rollout and we want to ensure it gets
688                                // retried.
689                                last_completed_rollout_request: status
690                                    .last_completed_rollout_request,
691                                last_completed_rollout_environmentd_image_ref: status
692                                    .last_completed_rollout_environmentd_image_ref,
693                                resource_id: status.resource_id,
694                                resources_hash: status.resources_hash,
695                                last_completed_rollout_hash: None,
696                                conditions: vec![Condition {
697                                    type_: "UpToDate".into(),
698                                    status: "False".into(),
699                                    last_transition_time: Time(Timestamp::now()),
700                                    message: format!(
701                                        "Failed to apply changes for \
702                                         generation {desired_generation}: {e}"
703                                    ),
704                                    observed_generation: mz.meta().generation,
705                                    reason: "FailedDeploy".into(),
706                                }],
707                            },
708                            active_generation != desired_generation,
709                        )
710                        .await?;
711                        Err(e)
712                    }
713                }
714            }
715            // There are changes pending, but we don't want to apply them yet.
716            (false, true, false) => {
717                let mut needs_update = mz.conditions_need_update();
718                if mz.update_in_progress() {
719                    resources
720                        .teardown_generation(&client, mz, next_generation)
721                        .await?;
722                    needs_update = true;
723                }
724                if needs_update {
725                    self.update_status(
726                        &mz_api,
727                        mz,
728                        MaterializeStatus {
729                            active_generation,
730                            last_completed_rollout_request: mz.requested_reconciliation_id(),
731                            last_completed_rollout_environmentd_image_ref: status
732                                .last_completed_rollout_environmentd_image_ref,
733                            resource_id: status.resource_id.clone(),
734                            resources_hash: status.resources_hash,
735                            last_completed_rollout_hash: None,
736                            conditions: vec![Condition {
737                                type_: "UpToDate".into(),
738                                status: "False".into(),
739                                last_transition_time: Time(Timestamp::now()),
740                                message: format!(
741                                    "Changes detected, waiting for approval for generation {desired_generation}"
742                                ),
743                                observed_generation: mz.meta().generation,
744                                reason: "WaitingForApproval".into(),
745                            }],
746                        },
747                        active_generation != desired_generation,
748                    )
749                    .await?;
750                }
751                debug!("changes detected, waiting for approval");
752                Ok(None)
753            }
754            // No changes pending, but we might need to clean up a partially applied rollout.
755            (false, false, _) => {
756                // this can happen if we update the environment, but then revert
757                // that update before the update was deployed. in this case, we
758                // don't want the environment to still show up as
759                // WaitingForApproval.
760                let mut needs_update = mz.conditions_need_update() || mz.rollout_requested();
761                if mz.update_in_progress() {
762                    resources
763                        .teardown_generation(&client, mz, next_generation)
764                        .await?;
765                    needs_update = true;
766                }
767                if needs_update {
768                    self.update_status(
769                        &mz_api,
770                        mz,
771                        MaterializeStatus {
772                            active_generation,
773                            last_completed_rollout_request: mz.requested_reconciliation_id(),
774                            last_completed_rollout_environmentd_image_ref: status
775                                .last_completed_rollout_environmentd_image_ref,
776                            resource_id: status.resource_id.clone(),
777                            resources_hash: status.resources_hash,
778                            last_completed_rollout_hash: None,
779                            conditions: vec![Condition {
780                                type_: "UpToDate".into(),
781                                status: "True".into(),
782                                last_transition_time: Time(Timestamp::now()),
783                                message: format!(
784                                    "No changes found from generation {active_generation}"
785                                ),
786                                observed_generation: mz.meta().generation,
787                                reason: "Applied".into(),
788                            }],
789                        },
790                        active_generation != desired_generation,
791                    )
792                    .await?;
793                }
794                debug!("no changes");
795                Ok(None)
796            }
797        }?;
798
799        if let Some(action) = result {
800            return Ok(Some(action));
801        }
802
803        // balancers rely on the environmentd service existing, which is
804        // enforced by the environmentd rollout process being able to call
805        // into the promotion endpoint
806
807        if self.config.create_balancers {
808            let balancer = Balancer {
809                metadata: mz.managed_resource_meta(mz.name_unchecked()),
810                spec: BalancerSpec {
811                    balancerd_image_ref: matching_image_from_environmentd_image_ref(
812                        mz.active_environmentd_image_ref(),
813                        "balancerd",
814                        None,
815                    ),
816                    resource_requirements: mz.spec.balancerd_resource_requirements.clone(),
817                    replicas: Some(mz.balancerd_replicas()),
818                    external_certificate_spec: mz.spec.balancerd_external_certificate_spec.clone(),
819                    internal_certificate_spec: mz.spec.internal_certificate_spec.clone(),
820                    pod_annotations: mz.spec.pod_annotations.clone(),
821                    pod_labels: mz.spec.pod_labels.clone(),
822                    static_routing: Some(
823                        mz_cloud_resources::crd::balancer::v1alpha1::StaticRoutingConfig {
824                            environmentd_namespace: mz.namespace(),
825                            environmentd_service_name: mz.environmentd_service_name(),
826                        },
827                    ),
828                    frontegg_routing: None,
829                    resource_id: Some(status.resource_id.clone()),
830                },
831                status: None,
832            };
833            let balancer = apply_resource(&balancer_api, &balancer).await?;
834            result = wait_for_balancer(&balancer)?;
835        } else {
836            delete_resource(&balancer_api, &mz.name_unchecked()).await?;
837        }
838
839        if let Some(action) = result {
840            return Ok(Some(action));
841        }
842
843        // and the console relies on the balancer service existing, which is
844        // enforced by wait_for_balancer
845
846        if self.config.create_console {
847            let active_environmentd_image_ref = mz.active_environmentd_image_ref();
848            let environmentd_image_tag =
849                parse_image_tag(active_environmentd_image_ref).unwrap_or("latest");
850            let console_image_tag = self
851                .config
852                .console_image_tag_map
853                .iter()
854                .find(|kv| kv.key == environmentd_image_tag)
855                .map(|kv| kv.value.clone())
856                .unwrap_or_else(|| self.config.console_image_tag_default.clone());
857            let console = Console {
858                metadata: mz.managed_resource_meta(mz.name_unchecked()),
859                spec: ConsoleSpec {
860                    console_image_ref: matching_image_from_environmentd_image_ref(
861                        active_environmentd_image_ref,
862                        "console",
863                        Some(&console_image_tag),
864                    ),
865                    resource_requirements: mz.spec.console_resource_requirements.clone(),
866                    replicas: Some(mz.console_replicas()),
867                    external_certificate_spec: mz.spec.console_external_certificate_spec.clone(),
868                    pod_annotations: mz.spec.pod_annotations.clone(),
869                    pod_labels: mz.spec.pod_labels.clone(),
870                    balancerd: BalancerdRef {
871                        service_name: mz.balancerd_service_name(),
872                        namespace: mz.namespace(),
873                        scheme: if issuer_ref_defined(
874                            &self.config.default_certificate_specs.balancerd_external,
875                            &mz.spec.balancerd_external_certificate_spec,
876                        ) {
877                            HttpConnectionScheme::Https
878                        } else {
879                            HttpConnectionScheme::Http
880                        },
881                        dns_names: resolved_dns_names(
882                            &self.config.default_certificate_specs.balancerd_external,
883                            &mz.spec.balancerd_external_certificate_spec,
884                        ),
885                    },
886                    authenticator_kind: mz.spec.authenticator_kind,
887                    resource_id: Some(status.resource_id),
888                },
889                status: None,
890            };
891            apply_resource(&console_api, &console).await?;
892        } else {
893            delete_resource(&console_api, &mz.name_unchecked()).await?;
894        }
895
896        Ok(result)
897    }
898
899    #[instrument(fields(organization_name=mz.name_unchecked()))]
900    async fn cleanup(
901        &self,
902        _client: Client,
903        mz: &Self::Resource,
904        _metadata: &mut TraceMetadata,
905    ) -> Result<Option<Action>, Self::Error> {
906        self.set_needs_update(mz, false);
907
908        Ok(None)
909    }
910}
911
912fn wait_for_balancer(balancer: &Balancer) -> Result<Option<Action>, Error> {
913    if let Some(conditions) = balancer
914        .status
915        .as_ref()
916        .map(|status| status.conditions.as_slice())
917    {
918        if conditions
919            .iter()
920            .any(|condition| condition.type_ == "Ready" && condition.status == "True")
921        {
922            return Ok(None);
923        }
924    }
925
926    Ok(Some(Action::requeue(Duration::from_secs(1))))
927}