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

mz_orchestratord/controller/
materialize.rs
