Skip to main content

mz_cloud_resources/crd/
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::collections::BTreeMap;
11use std::time::Duration;
12
13use k8s_openapi::{
14    api::core::v1::{EnvVar, ResourceRequirements},
15    apimachinery::pkg::{
16        api::resource::Quantity,
17        apis::meta::v1::{Condition, Time},
18    },
19    jiff::Timestamp,
20};
21use kube::{CustomResource, Resource, ResourceExt};
22use schemars::JsonSchema;
23use semver::Version;
24use serde::{Deserialize, Serialize};
25use sha2::{Digest, Sha256};
26use uuid::Uuid;
27
28use crate::crd::{ManagedResource, MaterializeCertSpec, new_resource_id};
29use mz_server_core::listeners::AuthenticatorKind;
30
31pub const LAST_KNOWN_ACTIVE_GENERATION_ANNOTATION: &str =
32    "materialize.cloud/last-known-active-generation";
33pub const FORCE_ROLLOUT_ANNOTATION: &str = "materialize.cloud/force-rollout";
34
35#[derive(Clone, Debug, Default, PartialEq, Deserialize, Serialize, JsonSchema)]
36pub enum MaterializeRolloutStrategy {
37    /// Create a new generation of pods, leaving the old generation around until the
38    /// new ones are ready to take over.
39    /// This minimizes downtime, and is what almost everyone should use.
40    #[default]
41    WaitUntilReady,
42
43    /// Create a new generation of pods, leaving the old generation as the serving generation
44    /// until the user manually promotes the new generation.
45    ///
46    /// When using `ManuallyPromote`, the new generation can be promoted at any
47    /// time, even if it has dataflows that are not fully caught up, by setting
48    /// `forcePromote` to the current rollout identifier: in `v1`, the value of
49    /// `status.requestedRolloutHash`; in `v1alpha1`, the `requestRollout` value
50    /// in the spec.
51    ///
52    /// To minimize downtime, promotion should occur when the new generation
53    /// has caught up to the prior generation. To determine if the new
54    /// generation has caught up, consult the `UpToDate` condition in the
55    /// status of the Materialize Resource. If the condition's reason is
56    /// `ReadyToPromote` the new generation is ready to promote.
57    ///
58    /// {{<warning>}}
59    /// Do not leave new generations unpromoted indefinitely.
60    ///
61    /// The new generation keeps open read holds which prevent compaction. Once promoted or
62    /// cancelled, those read holds are released. If left unpromoted for an extended time, this
63    /// data can build up, and can cause extreme deletion load on the metadata backend database
64    /// when finally promoted or cancelled.
65    ///
66    /// To guard against this, a rollout that remains in progress longer
67    /// than `rolloutRequestTimeout` (default 24h) is automatically
68    /// cancelled.
69    /// {{</warning>}}
70    ManuallyPromote,
71
72    /// {{<warning>}}
73    /// THIS WILL CAUSE YOUR MATERIALIZE INSTANCE TO BE UNAVAILABLE FOR SOME TIME!!!
74    ///
75    /// This strategy should ONLY be used by customers with physical hardware who do not have
76    /// enough hardware for the `WaitUntilReady` strategy. If you think you want this, please
77    /// consult with Materialize engineering to discuss your situation.
78    /// {{</warning>}}
79    ///
80    /// Tear down the old generation of pods and promote the new generation of pods immediately,
81    /// without waiting for the new generation of pods to be ready.
82    ImmediatelyPromoteCausingDowntime,
83}
84
85/// Default for [`RolloutRequestTimeout`]. A new generation that sits
86/// un-promoted holds back compaction via read holds, and promoting it
87/// after a long delay can cause incident-inducing load; 24h is a
88/// conservative upper bound on how long any rollout should take.
89pub const DEFAULT_ROLLOUT_REQUEST_TIMEOUT: &str = "24h";
90
91/// The maximum time [`v1alpha1::MaterializeSpec::rollout_request_timeout`] allows a
92/// rollout to remain in progress.
93///
94/// A transparent wrapper around the duration string whose [`Default`] is
95/// [`DEFAULT_ROLLOUT_REQUEST_TIMEOUT`]. Routing the default through `Default`
96/// keeps a single source of truth: the derived `Default` for
97/// [`v1alpha1::MaterializeSpec`], serde's `#[serde(default)]` (applied when the field
98/// is omitted on deserialize), and the schema default surfaced in the
99/// generated CRD (so the API server fills it in and `kubectl explain` shows
100/// it) all resolve to the same value.
101#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
102#[serde(transparent)]
103pub struct RolloutRequestTimeout(pub String);
104
105impl Default for RolloutRequestTimeout {
106    fn default() -> Self {
107        RolloutRequestTimeout(DEFAULT_ROLLOUT_REQUEST_TIMEOUT.to_owned())
108    }
109}
110
111pub mod v1alpha1 {
112    use super::*;
113
114    #[derive(
115        CustomResource,
116        Clone,
117        Debug,
118        Default,
119        PartialEq,
120        Deserialize,
121        Serialize,
122        JsonSchema
123    )]
124    #[serde(rename_all = "camelCase")]
125    #[kube(
126        namespaced,
127        group = "materialize.cloud",
128        version = "v1alpha1",
129        kind = "Materialize",
130        singular = "materialize",
131        plural = "materializes",
132        shortname = "mzs",
133        status = "MaterializeStatus",
134        printcolumn = r#"{"name": "ImageRefRunning", "type": "string", "description": "Reference to the Docker image that is currently in use.", "jsonPath": ".status.lastCompletedRolloutEnvironmentdImageRef", "priority": 1}"#,
135        printcolumn = r#"{"name": "ImageRefToDeploy", "type": "string", "description": "Reference to the Docker image which will be deployed on the next rollout.", "jsonPath": ".spec.environmentdImageRef", "priority": 1}"#,
136        printcolumn = r#"{"name": "UpToDate", "type": "string", "description": "Whether the spec has been applied", "jsonPath": ".status.conditions[?(@.type==\"UpToDate\")].status", "priority": 1}"#
137    )]
138    pub struct MaterializeSpec {
139        /// The environmentd image to run.
140        pub environmentd_image_ref: String,
141        /// Extra args to pass to the environmentd binary.
142        pub environmentd_extra_args: Option<Vec<String>>,
143        /// Extra environment variables to pass to the environmentd binary.
144        pub environmentd_extra_env: Option<Vec<EnvVar>>,
145        /// {{<warning>}}
146        /// Deprecated.
147        ///
148        /// Use `service_account_annotations` to set "eks.amazonaws.com/role-arn" instead.
149        /// {{</warning>}}
150        ///
151        /// If running in AWS, override the IAM role to use to give
152        /// environmentd access to the persist S3 bucket.
153        #[kube(deprecated)]
154        pub environmentd_iam_role_arn: Option<String>,
155        /// If running in AWS, override the IAM role to use to support
156        /// the CREATE CONNECTION feature.
157        pub environmentd_connection_role_arn: Option<String>,
158        /// Resource requirements for the environmentd pod.
159        pub environmentd_resource_requirements: Option<ResourceRequirements>,
160        /// Amount of disk to allocate, if a storage class is provided.
161        pub environmentd_scratch_volume_storage_requirement: Option<Quantity>,
162        /// Resource requirements for the balancerd pod.
163        pub balancerd_resource_requirements: Option<ResourceRequirements>,
164        /// Resource requirements for the console pod.
165        pub console_resource_requirements: Option<ResourceRequirements>,
166        /// Number of balancerd pods to create.
167        pub balancerd_replicas: Option<i32>,
168        /// Number of console pods to create.
169        pub console_replicas: Option<i32>,
170
171        /// Name of the kubernetes service account to use.
172        /// If not set, we will create one with the same name as this Materialize object.
173        pub service_account_name: Option<String>,
174        /// Annotations to apply to the service account.
175        ///
176        /// Annotations on service accounts are commonly used by cloud providers for IAM.
177        /// AWS uses "eks.amazonaws.com/role-arn".
178        /// Azure uses "azure.workload.identity/client-id", but
179        /// additionally requires "azure.workload.identity/use": "true" on the pods.
180        pub service_account_annotations: Option<BTreeMap<String, String>>,
181        /// Labels to apply to the service account.
182        pub service_account_labels: Option<BTreeMap<String, String>>,
183        /// Annotations to apply to the pods.
184        pub pod_annotations: Option<BTreeMap<String, String>>,
185        /// Labels to apply to the pods.
186        pub pod_labels: Option<BTreeMap<String, String>>,
187
188        /// When changes are made to the environmentd resources (either via
189        /// modifying fields in the spec here or by deploying a new
190        /// orchestratord version which changes how resources are generated),
191        /// existing environmentd processes won't be automatically restarted.
192        /// In order to trigger a restart, the request_rollout field should be
193        /// set to a new (random) value. Once the rollout completes, the value
194        /// of `status.lastCompletedRolloutRequest` will be set to this value
195        /// to indicate completion.
196        ///
197        /// Defaults to a random value in order to ensure that the first
198        /// generation rollout is automatically triggered.
199        #[serde(default)]
200        pub request_rollout: Uuid,
201        /// If `forcePromote` is set to the same value as `requestRollout`, the
202        /// current rollout will skip waiting for clusters in the new
203        /// generation to rehydrate before promoting the new environmentd to
204        /// leader.
205        #[serde(default)]
206        pub force_promote: String,
207        /// This value will be written to an annotation in the generated
208        /// environmentd statefulset, in order to force the controller to
209        /// detect the generated resources as changed even if no other changes
210        /// happened. This can be used to force a rollout to a new generation
211        /// even without making any meaningful changes, by setting it to the
212        /// same value as `requestRollout`.
213        #[serde(default)]
214        pub force_rollout: Uuid,
215        /// {{<warning>}}
216        /// Deprecated and ignored. Use `rolloutStrategy` instead.
217        /// {{</warning>}}
218        #[kube(deprecated)]
219        #[serde(default)]
220        pub in_place_rollout: bool,
221        /// Rollout strategy to use when upgrading this Materialize instance.
222        #[serde(default)]
223        pub rollout_strategy: MaterializeRolloutStrategy,
224        /// The maximum amount of time a rollout may remain in progress before
225        /// it is automatically cancelled.
226        ///
227        /// While a rollout is in progress, the new generation of `environmentd`
228        /// runs in a read-only, un-promoted state and holds back compaction via
229        /// read holds. Leaving it in this state for too long can cause
230        /// incident-inducing load when it is eventually promoted, so the
231        /// operator cancels the rollout once this timeout is exceeded: the new
232        /// generation is torn down and the previously-active generation
233        /// continues serving. A new rollout can then be triggered by setting
234        /// `requestRollout` to a new value.
235        ///
236        /// This does not apply to the `ImmediatelyPromoteCausingDowntime`
237        /// rollout strategy or to force-promoted rollouts, since by the time
238        /// those are in progress the old generation may already be gone.
239        ///
240        /// The value is parsed as a human-readable duration, e.g. `24h`,
241        /// `90m`, or `1h 30m`. Defaults to [`DEFAULT_ROLLOUT_REQUEST_TIMEOUT`]
242        /// when omitted (the API server fills it in); an unparseable value also
243        /// falls back to that default.
244        #[serde(default)]
245        pub rollout_request_timeout: RolloutRequestTimeout,
246        /// The name of a secret containing `metadata_backend_url` and `persist_backend_url`.
247        /// It may also contain `external_login_password_mz_system`, which will be used as
248        /// the password for the `mz_system` user if `authenticatorKind` is `Password`,
249        /// `Sasl`, or `Oidc`.
250        pub backend_secret_name: String,
251        /// How to authenticate with Materialize.
252        #[serde(default)]
253        pub authenticator_kind: AuthenticatorKind,
254        /// Whether to enable role based access control. Defaults to false.
255        #[serde(default)]
256        pub enable_rbac: bool,
257
258        /// The value used by environmentd (via the --environment-id flag) to
259        /// uniquely identify this instance. Must be globally unique, and
260        /// is required if a license key is not provided.
261        /// NOTE: This value MUST NOT be changed in an existing instance,
262        /// since it affects things like the way data is stored in the persist
263        /// backend.
264        #[serde(default)]
265        pub environment_id: Uuid,
266
267        /// The name of a ConfigMap containing system parameters in JSON format.
268        /// The ConfigMap must contain a `system-params.json` key whose value
269        /// is a valid JSON object containing valid system parameters.
270        ///
271        /// Run `SHOW ALL` in SQL to see a subset of configurable system parameters.
272        ///
273        /// Example ConfigMap:
274        /// ```yaml
275        /// data:
276        ///   system-params.json: |
277        ///     {
278        ///       "max_connections": 1000
279        ///     }
280        /// ```
281        pub system_parameter_configmap_name: Option<String>,
282
283        /// The configuration for generating an x509 certificate using cert-manager for balancerd
284        /// to present to incoming connections.
285        /// The `dnsNames` and `issuerRef` fields are required.
286        pub balancerd_external_certificate_spec: Option<MaterializeCertSpec>,
287        /// The configuration for generating an x509 certificate using cert-manager for the console
288        /// to present to incoming connections.
289        /// The `dnsNames` and `issuerRef` fields are required.
290        /// Not yet implemented.
291        pub console_external_certificate_spec: Option<MaterializeCertSpec>,
292        /// The cert-manager Issuer or ClusterIssuer to use for database internal communication.
293        /// The `issuerRef` field is required.
294        /// This currently is only used for environmentd, but will eventually support clusterd.
295        /// Not yet implemented.
296        pub internal_certificate_spec: Option<MaterializeCertSpec>,
297    }
298
299    impl Materialize {
300        pub fn backend_secret_name(&self) -> String {
301            self.spec.backend_secret_name.clone()
302        }
303
304        pub fn namespace(&self) -> String {
305            self.meta().namespace.clone().unwrap()
306        }
307
308        pub fn create_service_account(&self) -> bool {
309            self.spec.service_account_name.is_none()
310        }
311
312        pub fn service_account_name(&self) -> String {
313            self.spec
314                .service_account_name
315                .clone()
316                .unwrap_or_else(|| self.name_unchecked())
317        }
318
319        pub fn role_name(&self) -> String {
320            self.name_unchecked()
321        }
322
323        pub fn role_binding_name(&self) -> String {
324            self.name_unchecked()
325        }
326
327        pub fn environmentd_statefulset_name(&self, generation: u64) -> String {
328            self.name_prefixed(&format!("environmentd-{generation}"))
329        }
330
331        pub fn environmentd_app_name(&self) -> String {
332            "environmentd".to_owned()
333        }
334
335        pub fn environmentd_service_name(&self) -> String {
336            self.name_prefixed("environmentd")
337        }
338
339        pub fn environmentd_service_internal_fqdn(&self) -> String {
340            format!(
341                "{}.{}.svc.cluster.local",
342                self.environmentd_service_name(),
343                self.meta().namespace.as_ref().unwrap()
344            )
345        }
346
347        pub fn environmentd_generation_service_name(&self, generation: u64) -> String {
348            self.name_prefixed(&format!("environmentd-{generation}"))
349        }
350
351        pub fn balancerd_app_name(&self) -> String {
352            "balancerd".to_owned()
353        }
354
355        pub fn environmentd_certificate_name(&self) -> String {
356            self.name_prefixed("environmentd-external")
357        }
358
359        pub fn environmentd_certificate_secret_name(&self) -> String {
360            self.name_prefixed("environmentd-tls")
361        }
362
363        pub fn balancerd_deployment_name(&self) -> String {
364            self.name_prefixed("balancerd")
365        }
366
367        pub fn balancerd_service_name(&self) -> String {
368            self.name_prefixed("balancerd")
369        }
370
371        pub fn console_app_name(&self) -> String {
372            "console".to_owned()
373        }
374
375        pub fn balancerd_external_certificate_name(&self) -> String {
376            self.name_prefixed("balancerd-external")
377        }
378
379        pub fn balancerd_external_certificate_secret_name(&self) -> String {
380            self.name_prefixed("balancerd-external-tls")
381        }
382
383        pub fn balancerd_replicas(&self) -> i32 {
384            self.spec.balancerd_replicas.unwrap_or(2)
385        }
386
387        pub fn console_replicas(&self) -> i32 {
388            self.spec.console_replicas.unwrap_or(2)
389        }
390
391        pub fn console_configmap_name(&self) -> String {
392            self.name_prefixed("console")
393        }
394
395        pub fn console_deployment_name(&self) -> String {
396            self.name_prefixed("console")
397        }
398
399        pub fn console_service_name(&self) -> String {
400            self.name_prefixed("console")
401        }
402
403        pub fn console_external_certificate_name(&self) -> String {
404            self.name_prefixed("console-external")
405        }
406
407        pub fn console_external_certificate_secret_name(&self) -> String {
408            self.name_prefixed("console-external-tls")
409        }
410
411        pub fn persist_pubsub_service_name(&self, generation: u64) -> String {
412            self.name_prefixed(&format!("persist-pubsub-{generation}"))
413        }
414
415        pub fn listeners_configmap_name(&self, generation: u64) -> String {
416            self.name_prefixed(&format!("listeners-{generation}"))
417        }
418
419        pub fn name_prefixed(&self, suffix: &str) -> String {
420            format!("mz{}-{}", self.resource_id(), suffix)
421        }
422
423        pub fn resource_id(&self) -> &str {
424            &self.status.as_ref().unwrap().resource_id
425        }
426
427        pub fn system_parameter_configmap_name(&self) -> Option<String> {
428            self.spec.system_parameter_configmap_name.clone()
429        }
430
431        pub fn environmentd_scratch_volume_storage_requirement(&self) -> Quantity {
432            self.spec
433                .environmentd_scratch_volume_storage_requirement
434                .clone()
435                .unwrap_or_else(|| {
436                    self.spec
437                        .environmentd_resource_requirements
438                        .as_ref()
439                        .and_then(|requirements| {
440                            requirements
441                                .requests
442                                .as_ref()
443                                .or(requirements.limits.as_ref())
444                        })
445                        // TODO: in cloud, we've been defaulting to twice the
446                        // memory limit, but k8s-openapi doesn't seem to
447                        // provide any way to parse Quantity values, so there
448                        // isn't an easy way to do arithmetic on it
449                        .and_then(|requirements| requirements.get("memory").cloned())
450                        // TODO: is there a better default to use here?
451                        .unwrap_or_else(|| Quantity("4096Mi".to_string()))
452                })
453        }
454
455        pub fn environment_id(&self, cloud_provider: &str, region: &str) -> String {
456            format!(
457                "{}-{}-{}-0",
458                cloud_provider, region, self.spec.environment_id,
459            )
460        }
461
462        pub fn requested_reconciliation_id(&self) -> Uuid {
463            self.spec.request_rollout
464        }
465
466        pub fn rollout_requested(&self) -> bool {
467            self.requested_reconciliation_id()
468                != self
469                    .status
470                    .as_ref()
471                    .map_or_else(Uuid::nil, |status| status.last_completed_rollout_request)
472        }
473
474        /// The maximum amount of time a rollout may remain in progress before
475        /// it is automatically cancelled. Parsed from
476        /// [`MaterializeSpec::rollout_request_timeout`], falling back to
477        /// [`DEFAULT_ROLLOUT_REQUEST_TIMEOUT`] when unset or unparseable.
478        pub fn rollout_request_timeout(&self) -> Duration {
479            let timeout = &self.spec.rollout_request_timeout.0;
480            humantime::parse_duration(timeout)
481                .or_else(|e| {
482                    tracing::warn!(
483                        rollout_request_timeout = %timeout,
484                        "failed to parse rolloutRequestTimeout, using default: {e}",
485                    );
486                    humantime::parse_duration(DEFAULT_ROLLOUT_REQUEST_TIMEOUT)
487                })
488                .expect("DEFAULT_ROLLOUT_REQUEST_TIMEOUT must be a valid duration")
489        }
490
491        /// If a timeout-eligible rollout is currently in progress, returns the
492        /// time at which it entered the in-progress (`Unknown`) state. Used to
493        /// enforce the rollout timeout.
494        ///
495        /// The `Applying` and `ReadyToPromote` phases are both reported as a
496        /// single in-progress window: [`Self::up_to_date_transition_time`]
497        /// carries the timestamp forward across them (they share the `Unknown`
498        /// status), so the timeout spans the whole pre-promotion rollout rather
499        /// than resetting at each phase.
500        ///
501        /// The `Promoting` phase is deliberately excluded even though it is
502        /// also `Unknown`: once a rollout has reached promotion it must never
503        /// be cancelled by the timeout, since the previously-active generation
504        /// may already be torn down, leaving nothing to fall back to. (The
505        /// controller also never reaches the timeout check while promoting,
506        /// because `is_promoting` takes priority; this is belt-and-suspenders.)
507        pub fn rollout_in_progress_since(&self) -> Option<Timestamp> {
508            self.status
509                .as_ref()?
510                .conditions
511                .iter()
512                .find_map(|condition| {
513                    if condition.type_ == "UpToDate"
514                        && condition.status == "Unknown"
515                        && condition.reason != "Promoting"
516                    {
517                        Some(condition.last_transition_time.0)
518                    } else {
519                        None
520                    }
521                })
522        }
523
524        /// The `last_transition_time` to record for a new `UpToDate` condition
525        /// with `new_status`, following the Kubernetes convention that
526        /// `last_transition_time` marks when the condition's *status* last
527        /// changed — not its reason or message. While the status is unchanged
528        /// the existing timestamp is carried forward; it only resets to `now`
529        /// when the status actually changes (or there is no prior condition).
530        ///
531        /// This is what lets a rollout that moves through several same-status
532        /// phases (`Applying` -> `ReadyToPromote`, both `Unknown`) be measured
533        /// from when it first entered that status, so the rollout timeout
534        /// covers the phases together instead of restarting at each one.
535        pub fn up_to_date_transition_time(&self, new_status: &str, now: Timestamp) -> Timestamp {
536            self.status
537                .as_ref()
538                .and_then(|status| {
539                    status
540                        .conditions
541                        .iter()
542                        .find(|condition| condition.type_ == "UpToDate")
543                })
544                .filter(|condition| condition.status == new_status)
545                .map_or(now, |condition| condition.last_transition_time.0)
546        }
547
548        /// Returns the environmentd image ref of the currently-active
549        /// generation: the image of the last completed rollout, falling back
550        /// to the spec image when no rollout has completed yet. Downstream
551        /// resources (balancerd, console) should track this rather than
552        /// [`MaterializeSpec::environmentd_image_ref`] so they stay aligned
553        /// with the running environmentd when the spec is mid-rollout or has
554        /// been partially reverted (DEP-42).
555        pub fn active_environmentd_image_ref(&self) -> &str {
556            self.status
557                .as_ref()
558                .and_then(|s| s.last_completed_rollout_environmentd_image_ref.as_deref())
559                .unwrap_or(&self.spec.environmentd_image_ref)
560        }
561
562        pub fn set_force_promote(&mut self) {
563            self.spec.force_promote = self.spec.request_rollout.hyphenated().to_string();
564        }
565
566        pub fn should_force_promote(&self) -> bool {
567            self.spec.force_promote == self.spec.request_rollout.hyphenated().to_string()
568                || self.spec.force_promote
569                    == super::v1::Materialize::from(self.clone()).generate_rollout_hash()
570                || self.spec.rollout_strategy
571                    == MaterializeRolloutStrategy::ImmediatelyPromoteCausingDowntime
572        }
573
574        pub fn conditions_need_update(&self) -> bool {
575            let Some(status) = self.status.as_ref() else {
576                return true;
577            };
578            if status.conditions.is_empty() {
579                return true;
580            }
581            for condition in &status.conditions {
582                if condition.observed_generation != self.meta().generation {
583                    return true;
584                }
585            }
586            false
587        }
588
589        pub fn is_ready_to_promote(&self, resources_hash: &str) -> bool {
590            let Some(status) = self.status.as_ref() else {
591                return false;
592            };
593            if status.conditions.is_empty() {
594                return false;
595            }
596            status
597                .conditions
598                .iter()
599                .any(|condition| condition.reason == "ReadyToPromote")
600                && &status.resources_hash == resources_hash
601        }
602
603        pub fn is_promoting(&self) -> bool {
604            let Some(status) = self.status.as_ref() else {
605                return false;
606            };
607            if status.conditions.is_empty() {
608                return false;
609            }
610            status
611                .conditions
612                .iter()
613                .any(|condition| condition.reason == "Promoting")
614        }
615
616        pub fn update_in_progress(&self) -> bool {
617            let Some(status) = self.status.as_ref() else {
618                return false;
619            };
620            if status.conditions.is_empty() {
621                return false;
622            }
623            for condition in &status.conditions {
624                if condition.type_ == "UpToDate" && condition.status == "Unknown" {
625                    return true;
626                }
627            }
628            false
629        }
630
631        /// Checks that the given version is greater than or equal
632        /// to the existing version, if the existing version
633        /// can be parsed.
634        pub fn meets_minimum_version(&self, minimum: &Version) -> bool {
635            let version = parse_image_ref(&self.spec.environmentd_image_ref);
636            match version {
637                // Use cmp_precedence() to ignore build metadata per SemVer 2.0.0 spec
638                Some(version) => version.cmp_precedence(minimum).is_ge(),
639                // In the rare case that we see an image reference
640                // that we can't parse, we assume that it satisfies all
641                // version checks. Usually these are custom images that have
642                // been by a developer on a branch forked from a recent copy
643                // of main, and so this works out reasonably well in practice.
644                None => {
645                    tracing::warn!(
646                        image_ref = %self.spec.environmentd_image_ref,
647                        "failed to parse image ref",
648                    );
649                    true
650                }
651            }
652        }
653
654        /// This check isn't strictly required since environmentd will still be able to determine
655        /// if the upgrade is allowed or not. However, doing this check allows us to provide
656        /// the error as soon as possible and in a more user friendly way.
657        pub fn is_valid_upgrade_version(active_version: &Version, next_version: &Version) -> bool {
658            // Don't allow rolling back
659            // Note: semver comparison handles RC versions correctly:
660            // v26.0.0-rc.1 < v26.0.0-rc.2 < v26.0.0
661            // Use cmp_precedence() to ignore build metadata
662            if next_version.cmp_precedence(active_version) == std::cmp::Ordering::Less {
663                return false;
664            }
665
666            if active_version.major == 0 {
667                if next_version.major != active_version.major {
668                    if next_version.major == 26 {
669                        // We require customers to upgrade from 0.147.20 (Self Managed 25.2) or v0.164.X (Cloud)
670                        // before upgrading to 26.0.0
671                        return (active_version.minor == 147 && active_version.patch >= 20)
672                            || active_version.minor >= 164;
673                    } else {
674                        return false;
675                    }
676                }
677                // Self managed 25.1 to 25.2
678                if next_version.minor == 147 && active_version.minor == 130 {
679                    return true;
680                }
681                // only allow upgrading a single minor version at a time
682                return next_version.minor <= active_version.minor + 1;
683            } else if active_version.major >= 26 {
684                // For versions 26.X.X and onwards, we deny upgrades past 1 major version of the active version
685                return next_version.major <= active_version.major + 1;
686            }
687
688            true
689        }
690
691        /// Checks if the current environmentd image ref is within the upgrade window of the last
692        /// successful rollout.
693        pub fn within_upgrade_window(&self) -> bool {
694            let active_environmentd_version = self
695                .status
696                .as_ref()
697                .and_then(|status| {
698                    status
699                        .last_completed_rollout_environmentd_image_ref
700                        .as_ref()
701                })
702                .and_then(|image_ref| parse_image_ref(image_ref));
703
704            if let (Some(next_environmentd_version), Some(active_environmentd_version)) = (
705                parse_image_ref(&self.spec.environmentd_image_ref),
706                active_environmentd_version,
707            ) {
708                Self::is_valid_upgrade_version(
709                    &active_environmentd_version,
710                    &next_environmentd_version,
711                )
712            } else {
713                // If we fail to parse either version,
714                // we still allow the upgrade since environmentd will still error if the upgrade is not allowed.
715                true
716            }
717        }
718
719        pub fn status(&self) -> MaterializeStatus {
720            self.status.clone().unwrap_or_else(|| {
721                let mut status = MaterializeStatus::default();
722
723                status.resource_id = new_resource_id();
724
725                // If we're creating the initial status on an un-soft-deleted
726                // Environment we need to ensure that the last active generation
727                // is restored, otherwise the env will crash loop indefinitely
728                // as its catalog would have durably recorded a greater generation
729                if let Some(last_active_generation) = self
730                    .annotations()
731                    .get(LAST_KNOWN_ACTIVE_GENERATION_ANNOTATION)
732                {
733                    status.active_generation = last_active_generation
734                        .parse()
735                        .expect("valid int generation");
736                }
737
738                // Initialize the last completed rollout environmentd image ref to
739                // the current image ref if not already set.
740                status.last_completed_rollout_environmentd_image_ref =
741                    Some(self.spec.environmentd_image_ref.clone());
742
743                status
744            })
745        }
746    }
747
748    #[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema, PartialEq)]
749    #[serde(rename_all = "camelCase")]
750    pub struct MaterializeStatus {
751        /// Resource identifier used as a name prefix to avoid pod name collisions.
752        pub resource_id: String,
753        /// The generation of Materialize pods actively capable of servicing requests.
754        pub active_generation: u64,
755        /// The UUID of the last successfully completed rollout.
756        pub last_completed_rollout_request: Uuid,
757        /// The image ref of the environmentd image that was last successfully rolled out.
758        /// Used to deny upgrades past 1 major version from the last successful rollout.
759        /// When None, we upgrade anyways.
760        pub last_completed_rollout_environmentd_image_ref: Option<String>,
761        /// A hash calculated from the spec of resources to be created based on this Materialize
762        /// spec. This is used for detecting when the existing resources are up to date.
763        /// If you want to trigger a rollout without making other changes that would cause this
764        /// hash to change, you must set forceRollout to the same UUID as requestRollout.
765        pub resources_hash: String,
766        /// The last completed rollout hash from v1.
767        /// This exists on this older version only for round-trip conversion support.
768        pub last_completed_rollout_hash: Option<String>,
769        pub conditions: Vec<Condition>,
770    }
771
772    impl MaterializeStatus {
773        pub fn needs_update(&self, other: &Self) -> bool {
774            let now = Timestamp::now();
775            let mut a = self.clone();
776            for condition in &mut a.conditions {
777                condition.last_transition_time = Time(now);
778            }
779            let mut b = other.clone();
780            for condition in &mut b.conditions {
781                condition.last_transition_time = Time(now);
782            }
783            a != b
784        }
785    }
786
787    impl ManagedResource for Materialize {
788        fn default_labels(&self) -> BTreeMap<String, String> {
789            BTreeMap::from_iter([
790                (
791                    "materialize.cloud/organization-name".to_owned(),
792                    self.name_unchecked(),
793                ),
794                (
795                    "materialize.cloud/organization-namespace".to_owned(),
796                    self.namespace(),
797                ),
798                (
799                    "materialize.cloud/mz-resource-id".to_owned(),
800                    self.resource_id().to_owned(),
801                ),
802            ])
803        }
804
805        fn app_name(&self) -> Option<&str> {
806            Some("environmentd")
807        }
808    }
809
810    impl From<v1::Materialize> for Materialize {
811        fn from(value: v1::Materialize) -> Self {
812            let rollout_hash = value.generate_rollout_hash();
813            // Derive a deterministic UUID from the rollout hash so that the
814            // same v1 spec always produces the same requestRollout,
815            // making re-applies of an unchanged spec idempotent.
816            let request_rollout = Uuid::new_v5(&Uuid::NAMESPACE_OID, rollout_hash.as_bytes());
817            Materialize {
818                metadata: value.metadata,
819                spec: MaterializeSpec {
820                    environmentd_image_ref: value.spec.environmentd_image_ref,
821                    environmentd_extra_args: value.spec.environmentd_extra_args,
822                    environmentd_extra_env: value.spec.environmentd_extra_env,
823                    environmentd_iam_role_arn: None,
824                    environmentd_connection_role_arn: value.spec.environmentd_connection_role_arn,
825                    environmentd_resource_requirements: value
826                        .spec
827                        .environmentd_resource_requirements,
828                    environmentd_scratch_volume_storage_requirement: value
829                        .spec
830                        .environmentd_scratch_volume_storage_requirement,
831                    balancerd_resource_requirements: value.spec.balancerd_resource_requirements,
832                    console_resource_requirements: value.spec.console_resource_requirements,
833                    balancerd_replicas: value.spec.balancerd_replicas,
834                    console_replicas: value.spec.console_replicas,
835                    service_account_name: value.spec.service_account_name,
836                    service_account_annotations: value.spec.service_account_annotations,
837                    service_account_labels: value.spec.service_account_labels,
838                    pod_annotations: value.spec.pod_annotations,
839                    pod_labels: value.spec.pod_labels,
840                    force_promote: value.spec.force_promote.unwrap_or_default(),
841                    force_rollout: value.spec.force_rollout,
842                    rollout_strategy: value.spec.rollout_strategy,
843                    rollout_request_timeout: value.spec.rollout_request_timeout,
844                    backend_secret_name: value.spec.backend_secret_name,
845                    authenticator_kind: value.spec.authenticator_kind,
846                    enable_rbac: value.spec.enable_rbac,
847                    environment_id: value.spec.environment_id,
848                    system_parameter_configmap_name: value.spec.system_parameter_configmap_name,
849                    balancerd_external_certificate_spec: value
850                        .spec
851                        .balancerd_external_certificate_spec,
852                    console_external_certificate_spec: value.spec.console_external_certificate_spec,
853                    internal_certificate_spec: value.spec.internal_certificate_spec,
854                    request_rollout,
855                    in_place_rollout: false,
856                },
857                status: value.status.map(|status| MaterializeStatus {
858                    resource_id: status.resource_id,
859                    active_generation: status.active_generation,
860                    last_completed_rollout_environmentd_image_ref: status
861                        .last_completed_rollout_environmentd_image_ref,
862                    conditions: status.conditions,
863                    // Derive the same deterministic UUID from the last
864                    // completed hash so that request_rollout == this value
865                    // when the spec hasn't changed (no rollout needed).
866                    last_completed_rollout_request: status
867                        .last_completed_rollout_hash
868                        .as_ref()
869                        .map(|hash| Uuid::new_v5(&Uuid::NAMESPACE_OID, hash.as_bytes()))
870                        .unwrap_or(Uuid::nil()),
871                    last_completed_rollout_hash: status.last_completed_rollout_hash,
872                    resources_hash: "".to_owned(),
873                }),
874            }
875        }
876    }
877}
878
879pub mod v1 {
880    use super::*;
881
882    #[derive(
883        CustomResource,
884        Clone,
885        Debug,
886        Default,
887        PartialEq,
888        Deserialize,
889        Serialize,
890        JsonSchema
891    )]
892    #[serde(rename_all = "camelCase")]
893    #[kube(
894        namespaced,
895        group = "materialize.cloud",
896        version = "v1",
897        kind = "Materialize",
898        singular = "materialize",
899        plural = "materializes",
900        shortname = "mzs",
901        status = "MaterializeStatus",
902        printcolumn = r#"{"name": "ImageRefRunning", "type": "string", "description": "Reference to the Docker image that is currently in use.", "jsonPath": ".status.lastCompletedRolloutEnvironmentdImageRef", "priority": 1}"#,
903        printcolumn = r#"{"name": "ImageRefToDeploy", "type": "string", "description": "Reference to the Docker image which will be deployed on the next rollout.", "jsonPath": ".spec.environmentdImageRef", "priority": 1}"#,
904        printcolumn = r#"{"name": "UpToDate", "type": "string", "description": "Whether the spec has been applied", "jsonPath": ".status.conditions[?(@.type==\"UpToDate\")].status", "priority": 1}"#
905    )]
906    pub struct MaterializeSpec {
907        /// The environmentd image to run.
908        pub environmentd_image_ref: String,
909        /// Extra args to pass to the environmentd binary.
910        pub environmentd_extra_args: Option<Vec<String>>,
911        /// Extra environment variables to pass to the environmentd binary.
912        pub environmentd_extra_env: Option<Vec<EnvVar>>,
913        /// If running in AWS, override the IAM role to use to support
914        /// the CREATE CONNECTION feature.
915        pub environmentd_connection_role_arn: Option<String>,
916        /// Resource requirements for the environmentd pod.
917        pub environmentd_resource_requirements: Option<ResourceRequirements>,
918        /// Amount of disk to allocate, if a storage class is provided.
919        pub environmentd_scratch_volume_storage_requirement: Option<Quantity>,
920        /// Resource requirements for the balancerd pod.
921        ///
922        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
923        pub balancerd_resource_requirements: Option<ResourceRequirements>,
924        /// Resource requirements for the console pod.
925        ///
926        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
927        pub console_resource_requirements: Option<ResourceRequirements>,
928        /// Number of balancerd pods to create.
929        ///
930        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
931        pub balancerd_replicas: Option<i32>,
932        /// Number of console pods to create.
933        ///
934        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
935        pub console_replicas: Option<i32>,
936
937        /// Name of the kubernetes service account to use.
938        /// If not set, we will create one with the same name as this Materialize object.
939        pub service_account_name: Option<String>,
940        /// Annotations to apply to the service account.
941        ///
942        /// Annotations on service accounts are commonly used by cloud providers for IAM.
943        /// AWS uses "eks.amazonaws.com/role-arn".
944        /// Azure uses "azure.workload.identity/client-id", but
945        /// additionally requires "azure.workload.identity/use": "true" on the pods.
946        pub service_account_annotations: Option<BTreeMap<String, String>>,
947        /// Labels to apply to the service account.
948        pub service_account_labels: Option<BTreeMap<String, String>>,
949        /// Annotations to apply to the pods.
950        pub pod_annotations: Option<BTreeMap<String, String>>,
951        /// Labels to apply to the pods.
952        pub pod_labels: Option<BTreeMap<String, String>>,
953
954        /// If `forcePromote` is set to the same value as the `status.requestedRolloutHash`,
955        /// current rollout will skip waiting for clusters in the new
956        /// generation to rehydrate before promoting the new environmentd to
957        /// leader.
958        ///
959        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
960        pub force_promote: Option<String>,
961        /// This value will force the controller to detect the spec as changed
962        /// even if no other changes happened. This can be used to force a rollout
963        /// to a new generation even without making any meaningful changes.
964        #[serde(default)]
965        pub force_rollout: Uuid,
966        /// Rollout strategy to use when upgrading this Materialize instance.
967        #[serde(default)]
968        pub rollout_strategy: MaterializeRolloutStrategy,
969        /// The maximum amount of time a rollout may remain in progress before
970        /// it is automatically cancelled.
971        ///
972        /// While a rollout is in progress, the new generation of `environmentd`
973        /// runs in a read-only, un-promoted state and holds back compaction via
974        /// read holds. Leaving it in this state for too long can cause
975        /// incident-inducing load when it is eventually promoted, so the
976        /// operator cancels the rollout once this timeout is exceeded: the new
977        /// generation is torn down and the previously-active generation
978        /// continues serving. A new rollout can then be triggered by setting
979        /// `forceRollout` to a new value.
980        ///
981        /// This does not apply to the `ImmediatelyPromoteCausingDowntime`
982        /// rollout strategy or to force-promoted rollouts, since by the time
983        /// those are in progress the old generation may already be gone.
984        ///
985        /// The value is parsed as a human-readable duration, e.g. `24h`,
986        /// `90m`, or `1h 30m`. Defaults to [`DEFAULT_ROLLOUT_REQUEST_TIMEOUT`]
987        /// when omitted (the API server fills it in); an unparseable value also
988        /// falls back to that default.
989        #[serde(default)]
990        pub rollout_request_timeout: RolloutRequestTimeout,
991        /// The name of a secret containing `metadata_backend_url` and `persist_backend_url`.
992        /// It may also contain `external_login_password_mz_system`, which will be used as
993        /// the password for the `mz_system` user if `authenticatorKind` is `Password`.
994        pub backend_secret_name: String,
995        /// How to authenticate with Materialize.
996        #[serde(default)]
997        pub authenticator_kind: AuthenticatorKind,
998        /// Whether to enable role based access control. Defaults to false.
999        #[serde(default)]
1000        pub enable_rbac: bool,
1001
1002        /// The value used by environmentd (via the --environment-id flag) to
1003        /// uniquely identify this instance. Must be globally unique, and
1004        /// is required if a license key is not provided.
1005        /// NOTE: This value MUST NOT be changed in an existing instance,
1006        /// since it affects things like the way data is stored in the persist
1007        /// backend.
1008        #[serde(default)]
1009        pub environment_id: Uuid,
1010
1011        /// The name of a ConfigMap containing system parameters in JSON format.
1012        /// The ConfigMap must contain a `system-params.json` key whose value
1013        /// is a valid JSON object containing valid system parameters.
1014        ///
1015        /// Run `SHOW ALL` in SQL to see a subset of configurable system parameters.
1016        ///
1017        /// Example ConfigMap:
1018        /// ```yaml
1019        /// data:
1020        ///   system-params.json: |
1021        ///     {
1022        ///       "max_connections": 1000
1023        ///     }
1024        /// ```
1025        pub system_parameter_configmap_name: Option<String>,
1026
1027        /// The configuration for generating an x509 certificate using cert-manager for balancerd
1028        /// to present to incoming connections.
1029        /// The `dnsNames` and `issuerRef` fields are required.
1030        ///
1031        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1032        pub balancerd_external_certificate_spec: Option<MaterializeCertSpec>,
1033        /// The configuration for generating an x509 certificate using cert-manager for the console
1034        /// to present to incoming connections.
1035        /// The `dnsNames` and `issuerRef` fields are required.
1036        /// Not yet implemented.
1037        ///
1038        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1039        pub console_external_certificate_spec: Option<MaterializeCertSpec>,
1040        /// The cert-manager Issuer or ClusterIssuer to use for database internal communication.
1041        /// The `issuerRef` field is required.
1042        /// This currently is only used for environmentd, but will eventually support clusterd.
1043        /// Not yet implemented.
1044        pub internal_certificate_spec: Option<MaterializeCertSpec>,
1045    }
1046
1047    impl Materialize {
1048        pub fn generate_rollout_hash(&self) -> String {
1049            let mut hasher = Sha256::new();
1050            // Remove fields that don't affect the resources generated per generation,
1051            // and we don't want to trigger a rollout from.
1052            let spec = MaterializeSpec {
1053                environmentd_image_ref: self.spec.environmentd_image_ref.clone(),
1054                environmentd_extra_args: self.spec.environmentd_extra_args.clone(),
1055                environmentd_extra_env: self.spec.environmentd_extra_env.clone(),
1056                environmentd_connection_role_arn: self
1057                    .spec
1058                    .environmentd_connection_role_arn
1059                    .clone(),
1060                environmentd_resource_requirements: self
1061                    .spec
1062                    .environmentd_resource_requirements
1063                    .clone(),
1064                environmentd_scratch_volume_storage_requirement: self
1065                    .spec
1066                    .environmentd_scratch_volume_storage_requirement
1067                    .clone(),
1068                balancerd_resource_requirements: None,
1069                console_resource_requirements: None,
1070                balancerd_replicas: None,
1071                console_replicas: None,
1072                service_account_name: self.spec.service_account_name.clone(),
1073                service_account_annotations: self.spec.service_account_annotations.clone(),
1074                service_account_labels: self.spec.service_account_labels.clone(),
1075                pod_annotations: self.spec.pod_annotations.clone(),
1076                pod_labels: self.spec.pod_labels.clone(),
1077                force_promote: None,
1078                force_rollout: self.spec.force_rollout,
1079                rollout_strategy: self.spec.rollout_strategy.clone(),
1080                rollout_request_timeout: self.spec.rollout_request_timeout.clone(),
1081                backend_secret_name: self.spec.backend_secret_name.clone(),
1082                authenticator_kind: self.spec.authenticator_kind,
1083                enable_rbac: self.spec.enable_rbac,
1084                environment_id: self.spec.environment_id,
1085                system_parameter_configmap_name: self.spec.system_parameter_configmap_name.clone(),
1086                balancerd_external_certificate_spec: None,
1087                console_external_certificate_spec: None,
1088                internal_certificate_spec: self.spec.internal_certificate_spec.clone(),
1089            };
1090            hasher.update(&serde_json::to_vec(&spec).unwrap());
1091            if let Some(annotation) = self
1092                .metadata
1093                .annotations
1094                .as_ref()
1095                .and_then(|annotations| annotations.get(FORCE_ROLLOUT_ANNOTATION))
1096            {
1097                hasher.update(annotation);
1098            }
1099            format!("{:x}", hasher.finalize())
1100        }
1101
1102        pub fn backend_secret_name(&self) -> String {
1103            self.spec.backend_secret_name.clone()
1104        }
1105
1106        pub fn namespace(&self) -> String {
1107            self.meta().namespace.clone().unwrap()
1108        }
1109
1110        pub fn create_service_account(&self) -> bool {
1111            self.spec.service_account_name.is_none()
1112        }
1113
1114        pub fn service_account_name(&self) -> String {
1115            self.spec
1116                .service_account_name
1117                .clone()
1118                .unwrap_or_else(|| self.name_unchecked())
1119        }
1120
1121        pub fn role_name(&self) -> String {
1122            self.name_unchecked()
1123        }
1124
1125        pub fn role_binding_name(&self) -> String {
1126            self.name_unchecked()
1127        }
1128
1129        pub fn environmentd_statefulset_name(&self, generation: u64) -> String {
1130            self.name_prefixed(&format!("environmentd-{generation}"))
1131        }
1132
1133        pub fn environmentd_app_name(&self) -> String {
1134            "environmentd".to_owned()
1135        }
1136
1137        pub fn environmentd_service_name(&self) -> String {
1138            self.name_prefixed("environmentd")
1139        }
1140
1141        pub fn environmentd_service_internal_fqdn(&self) -> String {
1142            format!(
1143                "{}.{}.svc.cluster.local",
1144                self.environmentd_service_name(),
1145                self.meta().namespace.as_ref().unwrap()
1146            )
1147        }
1148
1149        pub fn environmentd_generation_service_name(&self, generation: u64) -> String {
1150            self.name_prefixed(&format!("environmentd-{generation}"))
1151        }
1152
1153        pub fn balancerd_app_name(&self) -> String {
1154            "balancerd".to_owned()
1155        }
1156
1157        pub fn environmentd_certificate_name(&self) -> String {
1158            self.name_prefixed("environmentd-external")
1159        }
1160
1161        pub fn environmentd_certificate_secret_name(&self) -> String {
1162            self.name_prefixed("environmentd-tls")
1163        }
1164
1165        pub fn balancerd_deployment_name(&self) -> String {
1166            self.name_prefixed("balancerd")
1167        }
1168
1169        pub fn balancerd_service_name(&self) -> String {
1170            self.name_prefixed("balancerd")
1171        }
1172
1173        pub fn console_app_name(&self) -> String {
1174            "console".to_owned()
1175        }
1176
1177        pub fn balancerd_external_certificate_name(&self) -> String {
1178            self.name_prefixed("balancerd-external")
1179        }
1180
1181        pub fn balancerd_external_certificate_secret_name(&self) -> String {
1182            self.name_prefixed("balancerd-external-tls")
1183        }
1184
1185        pub fn balancerd_replicas(&self) -> i32 {
1186            self.spec.balancerd_replicas.unwrap_or(2)
1187        }
1188
1189        pub fn console_replicas(&self) -> i32 {
1190            self.spec.console_replicas.unwrap_or(2)
1191        }
1192
1193        pub fn console_configmap_name(&self) -> String {
1194            self.name_prefixed("console")
1195        }
1196
1197        pub fn console_deployment_name(&self) -> String {
1198            self.name_prefixed("console")
1199        }
1200
1201        pub fn console_service_name(&self) -> String {
1202            self.name_prefixed("console")
1203        }
1204
1205        pub fn console_external_certificate_name(&self) -> String {
1206            self.name_prefixed("console-external")
1207        }
1208
1209        pub fn console_external_certificate_secret_name(&self) -> String {
1210            self.name_prefixed("console-external-tls")
1211        }
1212
1213        pub fn persist_pubsub_service_name(&self, generation: u64) -> String {
1214            self.name_prefixed(&format!("persist-pubsub-{generation}"))
1215        }
1216
1217        pub fn listeners_configmap_name(&self, generation: u64) -> String {
1218            self.name_prefixed(&format!("listeners-{generation}"))
1219        }
1220
1221        pub fn name_prefixed(&self, suffix: &str) -> String {
1222            format!("mz{}-{}", self.resource_id(), suffix)
1223        }
1224
1225        pub fn resource_id(&self) -> &str {
1226            &self.status.as_ref().unwrap().resource_id
1227        }
1228
1229        pub fn system_parameter_configmap_name(&self) -> Option<String> {
1230            self.spec.system_parameter_configmap_name.clone()
1231        }
1232
1233        pub fn environmentd_scratch_volume_storage_requirement(&self) -> Quantity {
1234            self.spec
1235                .environmentd_scratch_volume_storage_requirement
1236                .clone()
1237                .unwrap_or_else(|| {
1238                    self.spec
1239                        .environmentd_resource_requirements
1240                        .as_ref()
1241                        .and_then(|requirements| {
1242                            requirements
1243                                .requests
1244                                .as_ref()
1245                                .or(requirements.limits.as_ref())
1246                        })
1247                        // TODO: in cloud, we've been defaulting to twice the
1248                        // memory limit, but k8s-openapi doesn't seem to
1249                        // provide any way to parse Quantity values, so there
1250                        // isn't an easy way to do arithmetic on it
1251                        .and_then(|requirements| requirements.get("memory").cloned())
1252                        // TODO: is there a better default to use here?
1253                        .unwrap_or_else(|| Quantity("4096Mi".to_string()))
1254                })
1255        }
1256
1257        pub fn environment_id(&self, cloud_provider: &str, region: &str) -> String {
1258            format!(
1259                "{}-{}-{}-0",
1260                cloud_provider, region, self.spec.environment_id,
1261            )
1262        }
1263
1264        pub fn rollout_requested(&self) -> bool {
1265            self.status
1266                .as_ref()
1267                .map(|status| status.last_completed_rollout_hash != status.requested_rollout_hash)
1268                .unwrap_or(false)
1269        }
1270
1271        pub fn set_force_promote(&mut self) {
1272            self.spec.force_promote = Some(self.generate_rollout_hash());
1273        }
1274
1275        pub fn should_force_promote(&self) -> bool {
1276            self.spec.force_promote.as_ref()
1277                == self
1278                    .status
1279                    .as_ref()
1280                    .and_then(|status| status.requested_rollout_hash.as_ref())
1281                || self.spec.rollout_strategy
1282                    == MaterializeRolloutStrategy::ImmediatelyPromoteCausingDowntime
1283        }
1284
1285        pub fn conditions_need_update(&self) -> bool {
1286            let Some(status) = self.status.as_ref() else {
1287                return true;
1288            };
1289            if status.conditions.is_empty() {
1290                return true;
1291            }
1292            for condition in &status.conditions {
1293                if condition.observed_generation != self.meta().generation {
1294                    return true;
1295                }
1296            }
1297            false
1298        }
1299
1300        pub fn is_ready_to_promote(&self, rollout_hash: &str) -> bool {
1301            let Some(status) = self.status.as_ref() else {
1302                return false;
1303            };
1304            if status.conditions.is_empty() {
1305                return false;
1306            }
1307            status
1308                .conditions
1309                .iter()
1310                .any(|condition| condition.reason == "ReadyToPromote")
1311                && status.requested_rollout_hash.as_deref() == Some(rollout_hash)
1312        }
1313
1314        pub fn is_promoting(&self) -> bool {
1315            let Some(status) = self.status.as_ref() else {
1316                return false;
1317            };
1318            if status.conditions.is_empty() {
1319                return false;
1320            }
1321            status
1322                .conditions
1323                .iter()
1324                .any(|condition| condition.reason == "Promoting")
1325        }
1326
1327        pub fn update_in_progress(&self) -> bool {
1328            let Some(status) = self.status.as_ref() else {
1329                return false;
1330            };
1331            if status.conditions.is_empty() {
1332                return false;
1333            }
1334            for condition in &status.conditions {
1335                if condition.type_ == "UpToDate" && condition.status == "Unknown" {
1336                    return true;
1337                }
1338            }
1339            false
1340        }
1341
1342        /// Checks that the given version is greater than or equal
1343        /// to the existing version, if the existing version
1344        /// can be parsed.
1345        pub fn meets_minimum_version(&self, minimum: &Version) -> bool {
1346            let version = parse_image_ref(&self.spec.environmentd_image_ref);
1347            match version {
1348                // Use cmp_precedence() to ignore build metadata per SemVer 2.0.0 spec
1349                Some(version) => version.cmp_precedence(minimum).is_ge(),
1350                // In the rare case that we see an image reference
1351                // that we can't parse, we assume that it satisfies all
1352                // version checks. Usually these are custom images that have
1353                // been by a developer on a branch forked from a recent copy
1354                // of main, and so this works out reasonably well in practice.
1355                None => {
1356                    tracing::warn!(
1357                        image_ref = %self.spec.environmentd_image_ref,
1358                        "failed to parse image ref",
1359                    );
1360                    true
1361                }
1362            }
1363        }
1364
1365        /// This check isn't strictly required since environmentd will still be able to determine
1366        /// if the upgrade is allowed or not. However, doing this check allows us to provide
1367        /// the error as soon as possible and in a more user friendly way.
1368        pub fn is_valid_upgrade_version(active_version: &Version, next_version: &Version) -> bool {
1369            // Don't allow rolling back
1370            // Note: semver comparison handles RC versions correctly:
1371            // v26.0.0-rc.1 < v26.0.0-rc.2 < v26.0.0
1372            // Use cmp_precedence() to ignore build metadata
1373            if next_version.cmp_precedence(active_version) == std::cmp::Ordering::Less {
1374                return false;
1375            }
1376
1377            if active_version.major == 0 {
1378                if next_version.major != active_version.major {
1379                    if next_version.major == 26 {
1380                        // We require customers to upgrade from 0.147.20 (Self Managed 25.2) or v0.164.X (Cloud)
1381                        // before upgrading to 26.0.0
1382
1383                        return (active_version.minor == 147 && active_version.patch >= 20)
1384                            || active_version.minor >= 164;
1385                    } else {
1386                        return false;
1387                    }
1388                }
1389                // Self managed 25.1 to 25.2
1390                if next_version.minor == 147 && active_version.minor == 130 {
1391                    return true;
1392                }
1393                // only allow upgrading a single minor version at a time
1394                return next_version.minor <= active_version.minor + 1;
1395            } else if active_version.major >= 26 {
1396                // For versions 26.X.X and onwards, we deny upgrades past 1 major version of the active version
1397                return next_version.major <= active_version.major + 1;
1398            }
1399
1400            true
1401        }
1402
1403        /// Checks if the current environmentd image ref is within the upgrade window of the last
1404        /// successful rollout.
1405        pub fn within_upgrade_window(&self) -> bool {
1406            let active_environmentd_version = self
1407                .status
1408                .as_ref()
1409                .and_then(|status| {
1410                    status
1411                        .last_completed_rollout_environmentd_image_ref
1412                        .as_ref()
1413                })
1414                .and_then(|image_ref| parse_image_ref(image_ref));
1415
1416            if let (Some(next_environmentd_version), Some(active_environmentd_version)) = (
1417                parse_image_ref(&self.spec.environmentd_image_ref),
1418                active_environmentd_version,
1419            ) {
1420                Self::is_valid_upgrade_version(
1421                    &active_environmentd_version,
1422                    &next_environmentd_version,
1423                )
1424            } else {
1425                // If we fail to parse either version,
1426                // we still allow the upgrade since environmentd will still error if the upgrade is not allowed.
1427                true
1428            }
1429        }
1430
1431        pub fn status(&self) -> MaterializeStatus {
1432            self.status.clone().unwrap_or_else(|| {
1433                let mut status = MaterializeStatus::default();
1434
1435                status.resource_id = new_resource_id();
1436
1437                // If we're creating the initial status on an un-soft-deleted
1438                // Environment we need to ensure that the last active generation
1439                // is restored, otherwise the env will crash loop indefinitely
1440                // as its catalog would have durably recorded a greater generation
1441                if let Some(last_active_generation) = self
1442                    .annotations()
1443                    .get(LAST_KNOWN_ACTIVE_GENERATION_ANNOTATION)
1444                {
1445                    status.active_generation = last_active_generation
1446                        .parse()
1447                        .expect("valid int generation");
1448                }
1449
1450                // Initialize the last completed rollout environmentd image ref to
1451                // the current image ref if not already set.
1452                status.last_completed_rollout_environmentd_image_ref =
1453                    Some(self.spec.environmentd_image_ref.clone());
1454
1455                status
1456            })
1457        }
1458    }
1459
1460    #[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema, PartialEq)]
1461    #[serde(rename_all = "camelCase")]
1462    pub struct MaterializeStatus {
1463        /// Resource identifier used as a name prefix to avoid pod name collisions.
1464        pub resource_id: String,
1465        /// The generation of Materialize pods actively capable of servicing requests.
1466        pub active_generation: u64,
1467        /// The image ref of the environmentd image that was last successfully rolled out.
1468        /// Used to deny upgrades past 1 major version from the last successful rollout.
1469        /// When None, we upgrade anyways.
1470        pub last_completed_rollout_environmentd_image_ref: Option<String>,
1471        /// The last completed rollout's requestedRolloutHash.
1472        pub last_completed_rollout_hash: Option<String>,
1473        /// Hash of a subset of the Materialize spec and other fields.
1474        /// This is used to determine when the spec has changed and we need to rollout.
1475        pub requested_rollout_hash: Option<String>,
1476        pub conditions: Vec<Condition>,
1477    }
1478
1479    impl MaterializeStatus {
1480        pub fn needs_update(&self, other: &Self) -> bool {
1481            let now = Timestamp::now();
1482            let mut a = self.clone();
1483            for condition in &mut a.conditions {
1484                condition.last_transition_time = Time(now);
1485            }
1486            let mut b = other.clone();
1487            for condition in &mut b.conditions {
1488                condition.last_transition_time = Time(now);
1489            }
1490            a != b
1491        }
1492    }
1493
1494    impl ManagedResource for Materialize {
1495        fn default_labels(&self) -> BTreeMap<String, String> {
1496            BTreeMap::from_iter([
1497                (
1498                    "materialize.cloud/organization-name".to_owned(),
1499                    self.name_unchecked(),
1500                ),
1501                (
1502                    "materialize.cloud/organization-namespace".to_owned(),
1503                    self.namespace(),
1504                ),
1505                (
1506                    "materialize.cloud/mz-resource-id".to_owned(),
1507                    self.resource_id().to_owned(),
1508                ),
1509            ])
1510        }
1511
1512        fn app_name(&self) -> Option<&str> {
1513            Some("environmentd")
1514        }
1515    }
1516
1517    impl From<v1alpha1::Materialize> for Materialize {
1518        fn from(value: v1alpha1::Materialize) -> Self {
1519            let is_promoting = value.is_promoting();
1520            let service_account_annotations = if let Some(environmentd_iam_role_arn) =
1521                value.spec.environmentd_iam_role_arn
1522            {
1523                let mut annotations = value.spec.service_account_annotations.unwrap_or_default();
1524                annotations
1525                    .entry("eks.amazonaws.com/role-arn".to_owned())
1526                    .or_insert(environmentd_iam_role_arn);
1527                Some(annotations)
1528            } else {
1529                value.spec.service_account_annotations
1530            };
1531            let mut mz = Materialize {
1532                metadata: value.metadata,
1533                spec: MaterializeSpec {
1534                    environmentd_image_ref: value.spec.environmentd_image_ref,
1535                    environmentd_extra_args: value.spec.environmentd_extra_args,
1536                    environmentd_extra_env: value.spec.environmentd_extra_env,
1537                    environmentd_connection_role_arn: value.spec.environmentd_connection_role_arn,
1538                    environmentd_resource_requirements: value
1539                        .spec
1540                        .environmentd_resource_requirements,
1541                    environmentd_scratch_volume_storage_requirement: value
1542                        .spec
1543                        .environmentd_scratch_volume_storage_requirement,
1544                    balancerd_resource_requirements: value.spec.balancerd_resource_requirements,
1545                    console_resource_requirements: value.spec.console_resource_requirements,
1546                    balancerd_replicas: value.spec.balancerd_replicas,
1547                    console_replicas: value.spec.console_replicas,
1548                    service_account_name: value.spec.service_account_name,
1549                    service_account_annotations,
1550                    service_account_labels: value.spec.service_account_labels,
1551                    pod_annotations: value.spec.pod_annotations,
1552                    pod_labels: value.spec.pod_labels,
1553                    force_promote: if value.spec.force_promote.is_empty()
1554                        || &value.spec.force_promote == "00000000-0000-0000-0000-000000000000"
1555                    {
1556                        None
1557                    } else {
1558                        Some(value.spec.force_promote.to_string())
1559                    },
1560                    force_rollout: value.spec.force_rollout,
1561                    rollout_strategy: value.spec.rollout_strategy,
1562                    rollout_request_timeout: value.spec.rollout_request_timeout,
1563                    backend_secret_name: value.spec.backend_secret_name,
1564                    authenticator_kind: value.spec.authenticator_kind,
1565                    enable_rbac: value.spec.enable_rbac,
1566                    environment_id: value.spec.environment_id,
1567                    system_parameter_configmap_name: value.spec.system_parameter_configmap_name,
1568                    balancerd_external_certificate_spec: value
1569                        .spec
1570                        .balancerd_external_certificate_spec,
1571                    console_external_certificate_spec: value.spec.console_external_certificate_spec,
1572                    internal_certificate_spec: value.spec.internal_certificate_spec,
1573                },
1574                status: None,
1575            };
1576            let calculated_rollout_hash = mz.generate_rollout_hash();
1577            let last_completed_rollout_hash = match value
1578                .status
1579                .as_ref()
1580                .and_then(|status| status.last_completed_rollout_hash.to_owned())
1581            {
1582                Some(last_completed_rollout_hash) => Some(last_completed_rollout_hash),
1583                None => {
1584                    let currently_rolling_out = value
1585                        .status
1586                        .as_ref()
1587                        .map(|status| {
1588                            status.last_completed_rollout_request != value.spec.request_rollout
1589                                // If this is the first apply,
1590                                // these could both be nil and we still need to do a rollout.
1591                                || status.last_completed_rollout_request.is_nil()
1592                        })
1593                        .unwrap_or(true);
1594                    if currently_rolling_out {
1595                        // If they store a change, we're going to start over on a new rollout.
1596                        None
1597                    } else {
1598                        Some(calculated_rollout_hash.clone())
1599                    }
1600                }
1601            };
1602            let requested_rollout_hash = if is_promoting {
1603                None
1604            } else {
1605                Some(calculated_rollout_hash)
1606            };
1607            mz.status = value.status.map(|status| MaterializeStatus {
1608                resource_id: status.resource_id,
1609                active_generation: status.active_generation,
1610                last_completed_rollout_environmentd_image_ref: status
1611                    .last_completed_rollout_environmentd_image_ref,
1612                last_completed_rollout_hash,
1613                requested_rollout_hash,
1614                conditions: status.conditions,
1615            });
1616            mz
1617        }
1618    }
1619}
1620
1621fn parse_image_ref(image_ref: &str) -> Option<Version> {
1622    image_ref
1623        .rsplit_once(':')
1624        .and_then(|(_repo, tag)| tag.strip_prefix('v'))
1625        .and_then(|tag| {
1626            // To work around Docker tag restrictions, build metadata in
1627            // a Docker tag is delimited by `--` rather than the SemVer
1628            // `+` delimiter. So we need to swap the delimiter back to
1629            // `+` before parsing it as SemVer.
1630            let tag = tag.replace("--", "+");
1631            Version::parse(&tag).ok()
1632        })
1633}
1634
1635#[cfg(test)]
1636mod tests {
1637    use std::time::Duration;
1638
1639    use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
1640    use k8s_openapi::jiff::Timestamp;
1641    use kube::core::ObjectMeta;
1642    use semver::Version;
1643
1644    use super::v1alpha1::{Materialize, MaterializeSpec, MaterializeStatus};
1645    use super::{DEFAULT_ROLLOUT_REQUEST_TIMEOUT, RolloutRequestTimeout};
1646
1647    #[mz_ore::test]
1648    fn meets_minimum_version() {
1649        let mut mz = Materialize {
1650            spec: MaterializeSpec {
1651                environmentd_image_ref:
1652                    "materialize/environmentd:devel-47116c24b8d0df33d3f60a9ee476aa8d7bce5953"
1653                        .to_owned(),
1654                ..Default::default()
1655            },
1656            metadata: ObjectMeta {
1657                ..Default::default()
1658            },
1659            status: None,
1660        };
1661
1662        // true cases
1663        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1664        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0".to_owned();
1665        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1666        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.35.0".to_owned();
1667        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1668        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.3".to_owned();
1669        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1670        mz.spec.environmentd_image_ref = "materialize/environmentd@41af286dc0b172ed2f1ca934fd2278de4a1192302ffa07087cea2682e7d372e3".to_owned();
1671        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1672        mz.spec.environmentd_image_ref = "my.private.registry:5000:v0.34.3".to_owned();
1673        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1674        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.asdf.0".to_owned();
1675        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1676        mz.spec.environmentd_image_ref =
1677            "materialize/environmentd:v0.146.0-dev.0--pr.g5a05a9e4ba873be8adaa528644aaae6e4c7cd29b"
1678                .to_owned();
1679        assert!(mz.meets_minimum_version(&Version::parse("0.146.0-dev.0").unwrap()));
1680
1681        // false cases
1682        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0-dev".to_owned();
1683        assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1684        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.33.0".to_owned();
1685        assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1686        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0".to_owned();
1687        assert!(!mz.meets_minimum_version(&Version::parse("1.0.0").unwrap()));
1688        mz.spec.environmentd_image_ref = "my.private.registry:5000:v0.33.3".to_owned();
1689        assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1690    }
1691
1692    #[mz_ore::test]
1693    fn within_upgrade_window() {
1694        let mut mz = Materialize {
1695            spec: MaterializeSpec {
1696                environmentd_image_ref: "materialize/environmentd:v26.0.0".to_owned(),
1697                ..Default::default()
1698            },
1699            metadata: ObjectMeta {
1700                ..Default::default()
1701            },
1702            status: Some(MaterializeStatus {
1703                last_completed_rollout_environmentd_image_ref: Some(
1704                    "materialize/environmentd:v26.0.0".to_owned(),
1705                ),
1706                ..Default::default()
1707            }),
1708        };
1709
1710        // Pass: upgrading from 26.0.0 to 27.7.3 (within 1 major version)
1711        mz.spec.environmentd_image_ref = "materialize/environmentd:v27.7.3".to_owned();
1712        assert!(mz.within_upgrade_window());
1713
1714        // Pass: upgrading from 26.0.0 to 27.7.8-dev.0 (within 1 major version, pre-release)
1715        mz.spec.environmentd_image_ref = "materialize/environmentd:v27.7.8-dev.0".to_owned();
1716        assert!(mz.within_upgrade_window());
1717
1718        // Fail: upgrading from 26.0.0 to 28.0.1 (more than 1 major version)
1719        mz.spec.environmentd_image_ref = "materialize/environmentd:v28.0.1".to_owned();
1720        assert!(!mz.within_upgrade_window());
1721
1722        // Pass: upgrading from 26.0.0 to 28.0.1.not_a_valid_version (invalid version, defaults to true)
1723        mz.spec.environmentd_image_ref =
1724            "materialize/environmentd:v28.0.1.not_a_valid_version".to_owned();
1725        assert!(mz.within_upgrade_window());
1726
1727        // Pass: upgrading from 0.164.0 to 26.1.0 (self managed 25.2 to 26.0)
1728        mz.status
1729            .as_mut()
1730            .unwrap()
1731            .last_completed_rollout_environmentd_image_ref =
1732            Some("materialize/environmentd:v0.147.20".to_owned());
1733        mz.spec.environmentd_image_ref = "materialize/environmentd:v26.1.0".to_owned();
1734        assert!(mz.within_upgrade_window());
1735
1736        // Pass: upgrading from 26.11.0-dev.0+b to 26.11.0-dev.0+a (same major.minor.patch.prerelease, different build metadata)
1737        mz.status
1738            .as_mut()
1739            .unwrap()
1740            .last_completed_rollout_environmentd_image_ref =
1741            Some("materialize/environmentd:v26.11.0-dev.0+b".to_owned());
1742        mz.spec.environmentd_image_ref = "materialize/environmentd:v26.11.0-dev.0+a".to_owned();
1743        assert!(mz.within_upgrade_window());
1744    }
1745
1746    #[mz_ore::test]
1747    fn is_valid_upgrade_version() {
1748        let success_tests = [
1749            (Version::new(0, 83, 0), Version::new(0, 83, 0)),
1750            (Version::new(0, 83, 0), Version::new(0, 84, 0)),
1751            (Version::new(0, 9, 0), Version::new(0, 10, 0)),
1752            (Version::new(0, 99, 0), Version::new(0, 100, 0)),
1753            (Version::new(0, 83, 0), Version::new(0, 83, 1)),
1754            (Version::new(0, 83, 0), Version::new(0, 83, 2)),
1755            (Version::new(0, 83, 2), Version::new(0, 83, 10)),
1756            // 0.147.20 to 26.0.0 represents the Self Managed 25.2 to 26.0 upgrade
1757            (Version::new(0, 147, 20), Version::new(26, 0, 0)),
1758            (Version::new(0, 164, 0), Version::new(26, 0, 0)),
1759            (Version::new(26, 0, 0), Version::new(26, 1, 0)),
1760            (Version::new(26, 5, 3), Version::new(26, 10, 0)),
1761            (Version::new(0, 130, 0), Version::new(0, 147, 0)),
1762        ];
1763        for (active_version, next_version) in success_tests {
1764            assert!(
1765                Materialize::is_valid_upgrade_version(&active_version, &next_version),
1766                "v{active_version} can upgrade to v{next_version}"
1767            );
1768        }
1769
1770        let failure_tests = [
1771            (Version::new(0, 83, 0), Version::new(0, 82, 0)),
1772            (Version::new(0, 83, 3), Version::new(0, 83, 2)),
1773            (Version::new(0, 83, 3), Version::new(1, 83, 3)),
1774            (Version::new(0, 83, 0), Version::new(0, 85, 0)),
1775            (Version::new(26, 0, 0), Version::new(28, 0, 0)),
1776            (Version::new(0, 130, 0), Version::new(26, 1, 0)),
1777            // Disallow anything before 0.147.20 to upgrade
1778            (Version::new(0, 147, 1), Version::new(26, 0, 0)),
1779            // Disallow anything between 0.148.0 and 0.164.0 to upgrade
1780            (Version::new(0, 148, 0), Version::new(26, 0, 0)),
1781        ];
1782        for (active_version, next_version) in failure_tests {
1783            assert!(
1784                !Materialize::is_valid_upgrade_version(&active_version, &next_version),
1785                "v{active_version} can't upgrade to v{next_version}"
1786            );
1787        }
1788    }
1789
1790    #[mz_ore::test]
1791    fn rollout_request_timeout() {
1792        let mz_with = |timeout: &str| Materialize {
1793            spec: MaterializeSpec {
1794                rollout_request_timeout: RolloutRequestTimeout(timeout.to_owned()),
1795                ..Default::default()
1796            },
1797            metadata: ObjectMeta::default(),
1798            status: None,
1799        };
1800
1801        // The default const is a valid duration and resolves to 24h.
1802        let default = humantime::parse_duration(DEFAULT_ROLLOUT_REQUEST_TIMEOUT).unwrap();
1803        assert_eq!(default, Duration::from_secs(24 * 60 * 60));
1804
1805        // The field's Default (used by `MaterializeSpec::default()` and serde's
1806        // `#[serde(default)]`) is the 24h default, with no empty intermediate.
1807        assert_eq!(
1808            RolloutRequestTimeout::default().0,
1809            DEFAULT_ROLLOUT_REQUEST_TIMEOUT
1810        );
1811        assert_eq!(
1812            Materialize {
1813                spec: MaterializeSpec::default(),
1814                metadata: ObjectMeta::default(),
1815                status: None,
1816            }
1817            .rollout_request_timeout(),
1818            default
1819        );
1820
1821        // Parseable values are honored.
1822        assert_eq!(
1823            mz_with("1h").rollout_request_timeout(),
1824            Duration::from_secs(60 * 60)
1825        );
1826        assert_eq!(
1827            mz_with("90m").rollout_request_timeout(),
1828            Duration::from_secs(90 * 60)
1829        );
1830        assert_eq!(
1831            mz_with("1h 30m").rollout_request_timeout(),
1832            Duration::from_secs(90 * 60)
1833        );
1834        // Unparseable values fall back to the default.
1835        assert_eq!(mz_with("not a duration").rollout_request_timeout(), default);
1836    }
1837
1838    #[mz_ore::test]
1839    fn rollout_request_timeout_schema_default() {
1840        // The default must be surfaced in the generated CRD's OpenAPI schema
1841        // (not just in the Rust helper), so the Kubernetes API server defaults
1842        // omitted fields and `kubectl explain` shows it.
1843        let crd = serde_json::to_value(<Materialize as kube::CustomResourceExt>::crd())
1844            .expect("CRD serializes");
1845        let default = &crd["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"]
1846            ["properties"]["rolloutRequestTimeout"]["default"];
1847        assert_eq!(
1848            default,
1849            &serde_json::json!(DEFAULT_ROLLOUT_REQUEST_TIMEOUT),
1850            "rolloutRequestTimeout schema default missing/wrong in generated CRD",
1851        );
1852    }
1853
1854    #[mz_ore::test]
1855    fn rollout_in_progress_since() {
1856        let now = Timestamp::now();
1857        let condition = |type_: &str, status: &str| Condition {
1858            type_: type_.to_owned(),
1859            status: status.to_owned(),
1860            last_transition_time: Time(now),
1861            message: String::new(),
1862            observed_generation: None,
1863            reason: "Test".to_owned(),
1864        };
1865        let mz_with = |conditions: Vec<Condition>| Materialize {
1866            spec: MaterializeSpec::default(),
1867            metadata: ObjectMeta::default(),
1868            status: Some(MaterializeStatus {
1869                conditions,
1870                ..Default::default()
1871            }),
1872        };
1873
1874        // No status at all.
1875        let mz = Materialize {
1876            spec: MaterializeSpec::default(),
1877            metadata: ObjectMeta::default(),
1878            status: None,
1879        };
1880        assert_eq!(mz.rollout_in_progress_since(), None);
1881
1882        // A timeout-eligible rollout in progress is signalled by an `Unknown`
1883        // `UpToDate` condition (the Applying and ReadyToPromote phases).
1884        assert_eq!(
1885            mz_with(vec![condition("UpToDate", "Unknown")]).rollout_in_progress_since(),
1886            Some(now)
1887        );
1888
1889        // The `Promoting` phase is also `Unknown`, but must NOT be reported:
1890        // once promoting, the rollout can no longer be cancelled by the
1891        // timeout.
1892        assert_eq!(
1893            mz_with(vec![Condition {
1894                reason: "Promoting".to_owned(),
1895                ..condition("UpToDate", "Unknown")
1896            }])
1897            .rollout_in_progress_since(),
1898            None
1899        );
1900
1901        // A settled rollout (True/False) is not in progress.
1902        assert_eq!(
1903            mz_with(vec![condition("UpToDate", "True")]).rollout_in_progress_since(),
1904            None
1905        );
1906        assert_eq!(
1907            mz_with(vec![condition("UpToDate", "False")]).rollout_in_progress_since(),
1908            None
1909        );
1910    }
1911
1912    #[mz_ore::test]
1913    fn up_to_date_transition_time() {
1914        // Two distinct, fixed instants so we can tell "carried the old
1915        // timestamp" apart from "reset to now".
1916        let stored = Timestamp::from_second(1_000).unwrap();
1917        let now = Timestamp::from_second(2_000).unwrap();
1918
1919        let condition = |status: &str| Condition {
1920            type_: "UpToDate".to_owned(),
1921            status: status.to_owned(),
1922            last_transition_time: Time(stored),
1923            message: String::new(),
1924            observed_generation: None,
1925            reason: "Test".to_owned(),
1926        };
1927        let mz_with = |conditions: Vec<Condition>| Materialize {
1928            spec: MaterializeSpec::default(),
1929            metadata: ObjectMeta::default(),
1930            status: Some(MaterializeStatus {
1931                conditions,
1932                ..Default::default()
1933            }),
1934        };
1935
1936        // No prior condition: use `now`.
1937        let mz = Materialize {
1938            spec: MaterializeSpec::default(),
1939            metadata: ObjectMeta::default(),
1940            status: None,
1941        };
1942        assert_eq!(mz.up_to_date_transition_time("Unknown", now), now);
1943
1944        // Same status as the prior condition: carry its timestamp forward, so
1945        // consecutive same-status phases (Applying -> ReadyToPromote) share one
1946        // timer.
1947        assert_eq!(
1948            mz_with(vec![condition("Unknown")]).up_to_date_transition_time("Unknown", now),
1949            stored
1950        );
1951
1952        // Status changed: reset to `now`.
1953        assert_eq!(
1954            mz_with(vec![condition("Unknown")]).up_to_date_transition_time("True", now),
1955            now
1956        );
1957    }
1958
1959    #[mz_ore::test]
1960    fn active_environmentd_image_ref() {
1961        const OLD: &str = "materialize/environmentd:v26.0.0";
1962        const NEW: &str = "materialize/environmentd:v27.0.0";
1963
1964        let mz_with = |spec_image: &str, status: Option<MaterializeStatus>| Materialize {
1965            spec: MaterializeSpec {
1966                environmentd_image_ref: spec_image.to_owned(),
1967                ..Default::default()
1968            },
1969            metadata: ObjectMeta::default(),
1970            status,
1971        };
1972
1973        // No status yet (pre-initial-reconcile): fall back to spec.
1974        let mz = mz_with(NEW, None);
1975        assert_eq!(mz.active_environmentd_image_ref(), NEW);
1976
1977        // Status present but last_completed_rollout_environmentd_image_ref
1978        // unset (e.g. resource upgraded from older orchestratord that didn't
1979        // populate the field): fall back to spec.
1980        let mz = mz_with(
1981            NEW,
1982            Some(MaterializeStatus {
1983                last_completed_rollout_environmentd_image_ref: None,
1984                ..Default::default()
1985            }),
1986        );
1987        assert_eq!(mz.active_environmentd_image_ref(), NEW);
1988
1989        // Steady state: spec image == last completed image. Either source is
1990        // fine; the method must return that image.
1991        let mz = mz_with(
1992            NEW,
1993            Some(MaterializeStatus {
1994                last_completed_rollout_environmentd_image_ref: Some(NEW.to_owned()),
1995                ..Default::default()
1996            }),
1997        );
1998        assert_eq!(mz.active_environmentd_image_ref(), NEW);
1999
2000        // DEP-42 / mid-rollout: spec image == NEW but last_completed_* still
2001        // holds OLD — either because the user canceled the rollout by
2002        // reverting only requestRollout, or because the new generation has
2003        // not yet been promoted. The active environmentd is still OLD, so
2004        // downstream resources must track OLD. Without this method,
2005        // balancerd would inherit the spec's NEW image while environmentd
2006        // still runs OLD, leaving balancerd pods skewed from the running
2007        // env.
2008        let mz = mz_with(
2009            NEW,
2010            Some(MaterializeStatus {
2011                last_completed_rollout_environmentd_image_ref: Some(OLD.to_owned()),
2012                ..Default::default()
2013            }),
2014        );
2015        assert_eq!(mz.active_environmentd_image_ref(), OLD);
2016    }
2017}