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        /// The value used to force the generated per-generation resources to
467        /// be detected as changed even when nothing else in the spec changed,
468        /// so that a requested rollout actually creates a new generation of
469        /// pods rather than completing as a no-op.
470        ///
471        /// Combines `spec.forceRollout` with the
472        /// [`FORCE_ROLLOUT_ANNOTATION`] annotation; changing either forces a
473        /// new generation. The annotation exists so that automation (e.g. the
474        /// GCP node upgrade watcher in orchestratord) can force a rollout
475        /// without touching spec fields that may be managed by tools like
476        /// Terraform.
477        pub fn force_rollout_value(&self) -> String {
478            match self
479                .meta()
480                .annotations
481                .as_ref()
482                .and_then(|annotations| annotations.get(FORCE_ROLLOUT_ANNOTATION))
483            {
484                Some(annotation) => format!("{}/{}", self.spec.force_rollout, annotation),
485                None => self.spec.force_rollout.to_string(),
486            }
487        }
488
489        pub fn rollout_requested(&self) -> bool {
490            self.requested_reconciliation_id()
491                != self
492                    .status
493                    .as_ref()
494                    .map_or_else(Uuid::nil, |status| status.last_completed_rollout_request)
495        }
496
497        /// The maximum amount of time a rollout may remain in progress before
498        /// it is automatically cancelled. Parsed from
499        /// [`MaterializeSpec::rollout_request_timeout`], falling back to
500        /// [`DEFAULT_ROLLOUT_REQUEST_TIMEOUT`] when unset or unparseable.
501        pub fn rollout_request_timeout(&self) -> Duration {
502            let timeout = &self.spec.rollout_request_timeout.0;
503            humantime::parse_duration(timeout)
504                .or_else(|e| {
505                    tracing::warn!(
506                        rollout_request_timeout = %timeout,
507                        "failed to parse rolloutRequestTimeout, using default: {e}",
508                    );
509                    humantime::parse_duration(DEFAULT_ROLLOUT_REQUEST_TIMEOUT)
510                })
511                .expect("DEFAULT_ROLLOUT_REQUEST_TIMEOUT must be a valid duration")
512        }
513
514        /// If a timeout-eligible rollout is currently in progress, returns the
515        /// time at which it entered the in-progress (`Unknown`) state. Used to
516        /// enforce the rollout timeout.
517        ///
518        /// The `Applying` and `ReadyToPromote` phases are both reported as a
519        /// single in-progress window: [`Self::up_to_date_transition_time`]
520        /// carries the timestamp forward across them (they share the `Unknown`
521        /// status), so the timeout spans the whole pre-promotion rollout rather
522        /// than resetting at each phase.
523        ///
524        /// The `Promoting` phase is deliberately excluded even though it is
525        /// also `Unknown`: once a rollout has reached promotion it must never
526        /// be cancelled by the timeout, since the previously-active generation
527        /// may already be torn down, leaving nothing to fall back to. (The
528        /// controller also never reaches the timeout check while promoting,
529        /// because `is_promoting` takes priority; this is belt-and-suspenders.)
530        pub fn rollout_in_progress_since(&self) -> Option<Timestamp> {
531            self.status
532                .as_ref()?
533                .conditions
534                .iter()
535                .find_map(|condition| {
536                    if condition.type_ == "UpToDate"
537                        && condition.status == "Unknown"
538                        && condition.reason != "Promoting"
539                    {
540                        Some(condition.last_transition_time.0)
541                    } else {
542                        None
543                    }
544                })
545        }
546
547        /// The `last_transition_time` to record for a new `UpToDate` condition
548        /// with `new_status`, following the Kubernetes convention that
549        /// `last_transition_time` marks when the condition's *status* last
550        /// changed — not its reason or message. While the status is unchanged
551        /// the existing timestamp is carried forward; it only resets to `now`
552        /// when the status actually changes (or there is no prior condition).
553        ///
554        /// This is what lets a rollout that moves through several same-status
555        /// phases (`Applying` -> `ReadyToPromote`, both `Unknown`) be measured
556        /// from when it first entered that status, so the rollout timeout
557        /// covers the phases together instead of restarting at each one.
558        pub fn up_to_date_transition_time(&self, new_status: &str, now: Timestamp) -> Timestamp {
559            self.status
560                .as_ref()
561                .and_then(|status| {
562                    status
563                        .conditions
564                        .iter()
565                        .find(|condition| condition.type_ == "UpToDate")
566                })
567                .filter(|condition| condition.status == new_status)
568                .map_or(now, |condition| condition.last_transition_time.0)
569        }
570
571        /// Returns the environmentd image ref of the currently-active
572        /// generation: the image of the last completed rollout, falling back
573        /// to the spec image when no rollout has completed yet. Downstream
574        /// resources (balancerd, console) should track this rather than
575        /// [`MaterializeSpec::environmentd_image_ref`] so they stay aligned
576        /// with the running environmentd when the spec is mid-rollout or has
577        /// been partially reverted (DEP-42).
578        pub fn active_environmentd_image_ref(&self) -> &str {
579            self.status
580                .as_ref()
581                .and_then(|s| s.last_completed_rollout_environmentd_image_ref.as_deref())
582                .unwrap_or(&self.spec.environmentd_image_ref)
583        }
584
585        pub fn set_force_promote(&mut self) {
586            self.spec.force_promote = self.spec.request_rollout.hyphenated().to_string();
587        }
588
589        pub fn should_force_promote(&self) -> bool {
590            self.spec.force_promote == self.spec.request_rollout.hyphenated().to_string()
591                || self.spec.force_promote
592                    == super::v1::Materialize::from(self.clone()).generate_rollout_hash()
593                || self.spec.rollout_strategy
594                    == MaterializeRolloutStrategy::ImmediatelyPromoteCausingDowntime
595        }
596
597        pub fn conditions_need_update(&self) -> bool {
598            let Some(status) = self.status.as_ref() else {
599                return true;
600            };
601            if status.conditions.is_empty() {
602                return true;
603            }
604            for condition in &status.conditions {
605                if condition.observed_generation != self.meta().generation {
606                    return true;
607                }
608            }
609            false
610        }
611
612        pub fn is_ready_to_promote(&self, resources_hash: &str) -> bool {
613            let Some(status) = self.status.as_ref() else {
614                return false;
615            };
616            if status.conditions.is_empty() {
617                return false;
618            }
619            status
620                .conditions
621                .iter()
622                .any(|condition| condition.reason == "ReadyToPromote")
623                && &status.resources_hash == resources_hash
624        }
625
626        pub fn is_promoting(&self) -> bool {
627            let Some(status) = self.status.as_ref() else {
628                return false;
629            };
630            if status.conditions.is_empty() {
631                return false;
632            }
633            status
634                .conditions
635                .iter()
636                .any(|condition| condition.reason == "Promoting")
637        }
638
639        pub fn update_in_progress(&self) -> bool {
640            let Some(status) = self.status.as_ref() else {
641                return false;
642            };
643            if status.conditions.is_empty() {
644                return false;
645            }
646            for condition in &status.conditions {
647                if condition.type_ == "UpToDate" && condition.status == "Unknown" {
648                    return true;
649                }
650            }
651            false
652        }
653
654        /// Checks that the given version is greater than or equal
655        /// to the existing version, if the existing version
656        /// can be parsed.
657        pub fn meets_minimum_version(&self, minimum: &Version) -> bool {
658            let version = parse_image_ref(&self.spec.environmentd_image_ref);
659            match version {
660                // Use cmp_precedence() to ignore build metadata per SemVer 2.0.0 spec
661                Some(version) => version.cmp_precedence(minimum).is_ge(),
662                // In the rare case that we see an image reference
663                // that we can't parse, we assume that it satisfies all
664                // version checks. Usually these are custom images that have
665                // been by a developer on a branch forked from a recent copy
666                // of main, and so this works out reasonably well in practice.
667                None => {
668                    tracing::warn!(
669                        image_ref = %self.spec.environmentd_image_ref,
670                        "failed to parse image ref",
671                    );
672                    true
673                }
674            }
675        }
676
677        /// This check isn't strictly required since environmentd will still be able to determine
678        /// if the upgrade is allowed or not. However, doing this check allows us to provide
679        /// the error as soon as possible and in a more user friendly way.
680        pub fn is_valid_upgrade_version(active_version: &Version, next_version: &Version) -> bool {
681            // Don't allow rolling back
682            // Note: semver comparison handles RC versions correctly:
683            // v26.0.0-rc.1 < v26.0.0-rc.2 < v26.0.0
684            // Use cmp_precedence() to ignore build metadata
685            if next_version.cmp_precedence(active_version) == std::cmp::Ordering::Less {
686                return false;
687            }
688
689            if active_version.major == 0 {
690                if next_version.major != active_version.major {
691                    if next_version.major == 26 {
692                        // We require customers to upgrade from 0.147.20 (Self Managed 25.2) or v0.164.X (Cloud)
693                        // before upgrading to 26.0.0
694                        return (active_version.minor == 147 && active_version.patch >= 20)
695                            || active_version.minor >= 164;
696                    } else {
697                        return false;
698                    }
699                }
700                // Self managed 25.1 to 25.2
701                if next_version.minor == 147 && active_version.minor == 130 {
702                    return true;
703                }
704                // only allow upgrading a single minor version at a time
705                return next_version.minor <= active_version.minor + 1;
706            } else if active_version.major >= 26 {
707                // For versions 26.X.X and onwards, we deny upgrades past 1 major version of the active version
708                return next_version.major <= active_version.major + 1;
709            }
710
711            true
712        }
713
714        /// Checks if the current environmentd image ref is within the upgrade window of the last
715        /// successful rollout.
716        pub fn within_upgrade_window(&self) -> bool {
717            let active_environmentd_version = self
718                .status
719                .as_ref()
720                .and_then(|status| {
721                    status
722                        .last_completed_rollout_environmentd_image_ref
723                        .as_ref()
724                })
725                .and_then(|image_ref| parse_image_ref(image_ref));
726
727            if let (Some(next_environmentd_version), Some(active_environmentd_version)) = (
728                parse_image_ref(&self.spec.environmentd_image_ref),
729                active_environmentd_version,
730            ) {
731                Self::is_valid_upgrade_version(
732                    &active_environmentd_version,
733                    &next_environmentd_version,
734                )
735            } else {
736                // If we fail to parse either version,
737                // we still allow the upgrade since environmentd will still error if the upgrade is not allowed.
738                true
739            }
740        }
741
742        pub fn status(&self) -> MaterializeStatus {
743            self.status.clone().unwrap_or_else(|| {
744                let mut status = MaterializeStatus::default();
745
746                status.resource_id = new_resource_id();
747
748                // If we're creating the initial status on an un-soft-deleted
749                // Environment we need to ensure that the last active generation
750                // is restored, otherwise the env will crash loop indefinitely
751                // as its catalog would have durably recorded a greater generation
752                if let Some(last_active_generation) = self
753                    .annotations()
754                    .get(LAST_KNOWN_ACTIVE_GENERATION_ANNOTATION)
755                {
756                    status.active_generation = last_active_generation
757                        .parse()
758                        .expect("valid int generation");
759                }
760
761                // Initialize the last completed rollout environmentd image ref to
762                // the current image ref if not already set.
763                status.last_completed_rollout_environmentd_image_ref =
764                    Some(self.spec.environmentd_image_ref.clone());
765
766                status
767            })
768        }
769    }
770
771    #[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema, PartialEq)]
772    #[serde(rename_all = "camelCase")]
773    pub struct MaterializeStatus {
774        /// Resource identifier used as a name prefix to avoid pod name collisions.
775        pub resource_id: String,
776        /// The generation of Materialize pods actively capable of servicing requests.
777        pub active_generation: u64,
778        /// The UUID of the last successfully completed rollout.
779        pub last_completed_rollout_request: Uuid,
780        /// The image ref of the environmentd image that was last successfully rolled out.
781        /// Used to deny upgrades past 1 major version from the last successful rollout.
782        /// When None, we upgrade anyways.
783        pub last_completed_rollout_environmentd_image_ref: Option<String>,
784        /// A hash calculated from the spec of resources to be created based on this Materialize
785        /// spec. This is used for detecting when the existing resources are up to date.
786        /// If you want to trigger a rollout without making other changes that would cause this
787        /// hash to change, you must set forceRollout to the same UUID as requestRollout.
788        pub resources_hash: String,
789        /// The last completed rollout hash from v1.
790        /// This exists on this older version only for round-trip conversion support.
791        pub last_completed_rollout_hash: Option<String>,
792        pub conditions: Vec<Condition>,
793    }
794
795    impl MaterializeStatus {
796        pub fn needs_update(&self, other: &Self) -> bool {
797            let now = Timestamp::now();
798            let mut a = self.clone();
799            for condition in &mut a.conditions {
800                condition.last_transition_time = Time(now);
801            }
802            let mut b = other.clone();
803            for condition in &mut b.conditions {
804                condition.last_transition_time = Time(now);
805            }
806            a != b
807        }
808    }
809
810    impl ManagedResource for Materialize {
811        fn default_labels(&self) -> BTreeMap<String, String> {
812            BTreeMap::from_iter([
813                (
814                    "materialize.cloud/organization-name".to_owned(),
815                    self.name_unchecked(),
816                ),
817                (
818                    "materialize.cloud/organization-namespace".to_owned(),
819                    self.namespace(),
820                ),
821                (
822                    "materialize.cloud/mz-resource-id".to_owned(),
823                    self.resource_id().to_owned(),
824                ),
825            ])
826        }
827
828        fn app_name(&self) -> Option<&str> {
829            Some("environmentd")
830        }
831    }
832
833    impl From<v1::Materialize> for Materialize {
834        fn from(value: v1::Materialize) -> Self {
835            let rollout_hash = value.generate_rollout_hash();
836            // Derive a deterministic UUID from the rollout hash so that the
837            // same v1 spec always produces the same requestRollout,
838            // making re-applies of an unchanged spec idempotent.
839            let request_rollout = Uuid::new_v5(&Uuid::NAMESPACE_OID, rollout_hash.as_bytes());
840            Materialize {
841                metadata: value.metadata,
842                spec: MaterializeSpec {
843                    environmentd_image_ref: value.spec.environmentd_image_ref,
844                    environmentd_extra_args: value.spec.environmentd_extra_args,
845                    environmentd_extra_env: value.spec.environmentd_extra_env,
846                    environmentd_iam_role_arn: None,
847                    environmentd_connection_role_arn: value.spec.environmentd_connection_role_arn,
848                    environmentd_resource_requirements: value
849                        .spec
850                        .environmentd_resource_requirements,
851                    environmentd_scratch_volume_storage_requirement: value
852                        .spec
853                        .environmentd_scratch_volume_storage_requirement,
854                    balancerd_resource_requirements: value.spec.balancerd_resource_requirements,
855                    console_resource_requirements: value.spec.console_resource_requirements,
856                    balancerd_replicas: value.spec.balancerd_replicas,
857                    console_replicas: value.spec.console_replicas,
858                    service_account_name: value.spec.service_account_name,
859                    service_account_annotations: value.spec.service_account_annotations,
860                    service_account_labels: value.spec.service_account_labels,
861                    pod_annotations: value.spec.pod_annotations,
862                    pod_labels: value.spec.pod_labels,
863                    force_promote: value.spec.force_promote.unwrap_or_default(),
864                    force_rollout: value.spec.force_rollout,
865                    rollout_strategy: value.spec.rollout_strategy,
866                    rollout_request_timeout: value.spec.rollout_request_timeout,
867                    backend_secret_name: value.spec.backend_secret_name,
868                    authenticator_kind: value.spec.authenticator_kind,
869                    enable_rbac: value.spec.enable_rbac,
870                    environment_id: value.spec.environment_id,
871                    system_parameter_configmap_name: value.spec.system_parameter_configmap_name,
872                    balancerd_external_certificate_spec: value
873                        .spec
874                        .balancerd_external_certificate_spec,
875                    console_external_certificate_spec: value.spec.console_external_certificate_spec,
876                    internal_certificate_spec: value.spec.internal_certificate_spec,
877                    request_rollout,
878                    in_place_rollout: false,
879                },
880                status: value.status.map(|status| MaterializeStatus {
881                    resource_id: status.resource_id,
882                    active_generation: status.active_generation,
883                    last_completed_rollout_environmentd_image_ref: status
884                        .last_completed_rollout_environmentd_image_ref,
885                    conditions: status.conditions,
886                    // Derive the same deterministic UUID from the last
887                    // completed hash so that request_rollout == this value
888                    // when the spec hasn't changed (no rollout needed).
889                    last_completed_rollout_request: status
890                        .last_completed_rollout_hash
891                        .as_ref()
892                        .map(|hash| Uuid::new_v5(&Uuid::NAMESPACE_OID, hash.as_bytes()))
893                        .unwrap_or(Uuid::nil()),
894                    last_completed_rollout_hash: status.last_completed_rollout_hash,
895                    resources_hash: "".to_owned(),
896                }),
897            }
898        }
899    }
900
901    /// Partial mirror of [`MaterializeSpec`] for field-wise conversion of
902    /// objects that may be incomplete. See [`super::convert_v1_to_v1alpha1`].
903    ///
904    /// Each field is a [`PartialField`], distinguishing absent from explicit
905    /// null from a value, and unknown fields pass through via `extra`, so
906    /// serializing reproduces exactly what was deserialized. Adding a field
907    /// to [`MaterializeSpec`] breaks the exhaustive destructures in the
908    /// `From` impls below until its conversion is decided.
909    #[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
910    #[serde(rename_all = "camelCase")]
911    pub struct PartialMaterializeSpec {
912        #[serde(
913            default,
914            with = "double_option",
915            skip_serializing_if = "Option::is_none"
916        )]
917        pub environmentd_image_ref: PartialField,
918        #[serde(
919            default,
920            with = "double_option",
921            skip_serializing_if = "Option::is_none"
922        )]
923        pub environmentd_extra_args: PartialField,
924        #[serde(
925            default,
926            with = "double_option",
927            skip_serializing_if = "Option::is_none"
928        )]
929        pub environmentd_extra_env: PartialField,
930        #[serde(
931            default,
932            with = "double_option",
933            skip_serializing_if = "Option::is_none"
934        )]
935        pub environmentd_iam_role_arn: PartialField,
936        #[serde(
937            default,
938            with = "double_option",
939            skip_serializing_if = "Option::is_none"
940        )]
941        pub environmentd_connection_role_arn: PartialField,
942        #[serde(
943            default,
944            with = "double_option",
945            skip_serializing_if = "Option::is_none"
946        )]
947        pub environmentd_resource_requirements: PartialField,
948        #[serde(
949            default,
950            with = "double_option",
951            skip_serializing_if = "Option::is_none"
952        )]
953        pub environmentd_scratch_volume_storage_requirement: PartialField,
954        #[serde(
955            default,
956            with = "double_option",
957            skip_serializing_if = "Option::is_none"
958        )]
959        pub balancerd_resource_requirements: PartialField,
960        #[serde(
961            default,
962            with = "double_option",
963            skip_serializing_if = "Option::is_none"
964        )]
965        pub console_resource_requirements: PartialField,
966        #[serde(
967            default,
968            with = "double_option",
969            skip_serializing_if = "Option::is_none"
970        )]
971        pub balancerd_replicas: PartialField,
972        #[serde(
973            default,
974            with = "double_option",
975            skip_serializing_if = "Option::is_none"
976        )]
977        pub console_replicas: PartialField,
978        #[serde(
979            default,
980            with = "double_option",
981            skip_serializing_if = "Option::is_none"
982        )]
983        pub service_account_name: PartialField,
984        #[serde(
985            default,
986            with = "double_option",
987            skip_serializing_if = "Option::is_none"
988        )]
989        pub service_account_annotations: PartialField,
990        #[serde(
991            default,
992            with = "double_option",
993            skip_serializing_if = "Option::is_none"
994        )]
995        pub service_account_labels: PartialField,
996        #[serde(
997            default,
998            with = "double_option",
999            skip_serializing_if = "Option::is_none"
1000        )]
1001        pub pod_annotations: PartialField,
1002        #[serde(
1003            default,
1004            with = "double_option",
1005            skip_serializing_if = "Option::is_none"
1006        )]
1007        pub pod_labels: PartialField,
1008        #[serde(
1009            default,
1010            with = "double_option",
1011            skip_serializing_if = "Option::is_none"
1012        )]
1013        pub request_rollout: PartialField,
1014        #[serde(
1015            default,
1016            with = "double_option",
1017            skip_serializing_if = "Option::is_none"
1018        )]
1019        pub force_promote: PartialField,
1020        #[serde(
1021            default,
1022            with = "double_option",
1023            skip_serializing_if = "Option::is_none"
1024        )]
1025        pub force_rollout: PartialField,
1026        #[serde(
1027            default,
1028            with = "double_option",
1029            skip_serializing_if = "Option::is_none"
1030        )]
1031        pub in_place_rollout: PartialField,
1032        #[serde(
1033            default,
1034            with = "double_option",
1035            skip_serializing_if = "Option::is_none"
1036        )]
1037        pub rollout_strategy: PartialField,
1038        #[serde(
1039            default,
1040            with = "double_option",
1041            skip_serializing_if = "Option::is_none"
1042        )]
1043        pub rollout_request_timeout: PartialField,
1044        #[serde(
1045            default,
1046            with = "double_option",
1047            skip_serializing_if = "Option::is_none"
1048        )]
1049        pub backend_secret_name: PartialField,
1050        #[serde(
1051            default,
1052            with = "double_option",
1053            skip_serializing_if = "Option::is_none"
1054        )]
1055        pub authenticator_kind: PartialField,
1056        #[serde(
1057            default,
1058            with = "double_option",
1059            skip_serializing_if = "Option::is_none"
1060        )]
1061        pub enable_rbac: PartialField,
1062        #[serde(
1063            default,
1064            with = "double_option",
1065            skip_serializing_if = "Option::is_none"
1066        )]
1067        pub environment_id: PartialField,
1068        #[serde(
1069            default,
1070            with = "double_option",
1071            skip_serializing_if = "Option::is_none"
1072        )]
1073        pub system_parameter_configmap_name: PartialField,
1074        #[serde(
1075            default,
1076            with = "double_option",
1077            skip_serializing_if = "Option::is_none"
1078        )]
1079        pub balancerd_external_certificate_spec: PartialField,
1080        #[serde(
1081            default,
1082            with = "double_option",
1083            skip_serializing_if = "Option::is_none"
1084        )]
1085        pub console_external_certificate_spec: PartialField,
1086        #[serde(
1087            default,
1088            with = "double_option",
1089            skip_serializing_if = "Option::is_none"
1090        )]
1091        pub internal_certificate_spec: PartialField,
1092        #[serde(flatten)]
1093        pub extra: serde_json::Map<String, serde_json::Value>,
1094    }
1095
1096    /// Partial mirror of [`MaterializeStatus`], see [`PartialMaterializeSpec`].
1097    #[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
1098    #[serde(rename_all = "camelCase")]
1099    pub struct PartialMaterializeStatus {
1100        #[serde(
1101            default,
1102            with = "double_option",
1103            skip_serializing_if = "Option::is_none"
1104        )]
1105        pub resource_id: PartialField,
1106        #[serde(
1107            default,
1108            with = "double_option",
1109            skip_serializing_if = "Option::is_none"
1110        )]
1111        pub active_generation: PartialField,
1112        #[serde(
1113            default,
1114            with = "double_option",
1115            skip_serializing_if = "Option::is_none"
1116        )]
1117        pub last_completed_rollout_request: PartialField,
1118        #[serde(
1119            default,
1120            with = "double_option",
1121            skip_serializing_if = "Option::is_none"
1122        )]
1123        pub last_completed_rollout_environmentd_image_ref: PartialField,
1124        #[serde(
1125            default,
1126            with = "double_option",
1127            skip_serializing_if = "Option::is_none"
1128        )]
1129        pub resources_hash: PartialField,
1130        #[serde(
1131            default,
1132            with = "double_option",
1133            skip_serializing_if = "Option::is_none"
1134        )]
1135        pub last_completed_rollout_hash: PartialField,
1136        #[serde(
1137            default,
1138            with = "double_option",
1139            skip_serializing_if = "Option::is_none"
1140        )]
1141        pub conditions: PartialField,
1142        #[serde(flatten)]
1143        pub extra: serde_json::Map<String, serde_json::Value>,
1144    }
1145
1146    /// Partial mirror of [`Materialize`], see [`PartialMaterializeSpec`].
1147    #[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
1148    #[serde(rename_all = "camelCase")]
1149    pub struct PartialMaterialize {
1150        #[serde(default, skip_serializing_if = "Option::is_none")]
1151        pub api_version: Option<String>,
1152        #[serde(default, skip_serializing_if = "Option::is_none")]
1153        pub kind: Option<serde_json::Value>,
1154        #[serde(default, skip_serializing_if = "Option::is_none")]
1155        pub metadata: Option<serde_json::Value>,
1156        #[serde(default, skip_serializing_if = "Option::is_none")]
1157        pub spec: Option<PartialMaterializeSpec>,
1158        #[serde(default, skip_serializing_if = "Option::is_none")]
1159        pub status: Option<PartialMaterializeStatus>,
1160        #[serde(flatten)]
1161        pub extra: serde_json::Map<String, serde_json::Value>,
1162    }
1163
1164    impl From<MaterializeSpec> for PartialMaterializeSpec {
1165        fn from(spec: MaterializeSpec) -> Self {
1166            let MaterializeSpec {
1167                environmentd_image_ref,
1168                environmentd_extra_args,
1169                environmentd_extra_env,
1170                environmentd_iam_role_arn,
1171                environmentd_connection_role_arn,
1172                environmentd_resource_requirements,
1173                environmentd_scratch_volume_storage_requirement,
1174                balancerd_resource_requirements,
1175                console_resource_requirements,
1176                balancerd_replicas,
1177                console_replicas,
1178                service_account_name,
1179                service_account_annotations,
1180                service_account_labels,
1181                pod_annotations,
1182                pod_labels,
1183                request_rollout,
1184                force_promote,
1185                force_rollout,
1186                in_place_rollout,
1187                rollout_strategy,
1188                rollout_request_timeout,
1189                backend_secret_name,
1190                authenticator_kind,
1191                enable_rbac,
1192                environment_id,
1193                system_parameter_configmap_name,
1194                balancerd_external_certificate_spec,
1195                console_external_certificate_spec,
1196                internal_certificate_spec,
1197            } = spec;
1198            Self {
1199                environmentd_image_ref: present(environmentd_image_ref),
1200                environmentd_extra_args: present_opt(environmentd_extra_args),
1201                environmentd_extra_env: present_opt(environmentd_extra_env),
1202                environmentd_iam_role_arn: present_opt(environmentd_iam_role_arn),
1203                environmentd_connection_role_arn: present_opt(environmentd_connection_role_arn),
1204                environmentd_resource_requirements: present_opt(environmentd_resource_requirements),
1205                environmentd_scratch_volume_storage_requirement: present_opt(
1206                    environmentd_scratch_volume_storage_requirement,
1207                ),
1208                balancerd_resource_requirements: present_opt(balancerd_resource_requirements),
1209                console_resource_requirements: present_opt(console_resource_requirements),
1210                balancerd_replicas: present_opt(balancerd_replicas),
1211                console_replicas: present_opt(console_replicas),
1212                service_account_name: present_opt(service_account_name),
1213                service_account_annotations: present_opt(service_account_annotations),
1214                service_account_labels: present_opt(service_account_labels),
1215                pod_annotations: present_opt(pod_annotations),
1216                pod_labels: present_opt(pod_labels),
1217                request_rollout: present(request_rollout),
1218                force_promote: present(force_promote),
1219                force_rollout: present(force_rollout),
1220                in_place_rollout: present(in_place_rollout),
1221                rollout_strategy: present(rollout_strategy),
1222                rollout_request_timeout: present(rollout_request_timeout),
1223                backend_secret_name: present(backend_secret_name),
1224                authenticator_kind: present(authenticator_kind),
1225                enable_rbac: present(enable_rbac),
1226                environment_id: present(environment_id),
1227                system_parameter_configmap_name: present_opt(system_parameter_configmap_name),
1228                balancerd_external_certificate_spec: present_opt(
1229                    balancerd_external_certificate_spec,
1230                ),
1231                console_external_certificate_spec: present_opt(console_external_certificate_spec),
1232                internal_certificate_spec: present_opt(internal_certificate_spec),
1233                extra: serde_json::Map::new(),
1234            }
1235        }
1236    }
1237
1238    impl From<MaterializeStatus> for PartialMaterializeStatus {
1239        fn from(status: MaterializeStatus) -> Self {
1240            let MaterializeStatus {
1241                resource_id,
1242                active_generation,
1243                last_completed_rollout_request,
1244                last_completed_rollout_environmentd_image_ref,
1245                resources_hash,
1246                last_completed_rollout_hash,
1247                conditions,
1248            } = status;
1249            Self {
1250                resource_id: present(resource_id),
1251                active_generation: present(active_generation),
1252                last_completed_rollout_request: present(last_completed_rollout_request),
1253                last_completed_rollout_environmentd_image_ref: present_opt(
1254                    last_completed_rollout_environmentd_image_ref,
1255                ),
1256                resources_hash: present(resources_hash),
1257                last_completed_rollout_hash: present_opt(last_completed_rollout_hash),
1258                conditions: present(conditions),
1259                extra: serde_json::Map::new(),
1260            }
1261        }
1262    }
1263
1264    impl From<Materialize> for PartialMaterialize {
1265        fn from(mz: Materialize) -> Self {
1266            let Materialize {
1267                metadata,
1268                spec,
1269                status,
1270            } = mz;
1271            Self {
1272                api_version: Some("materialize.cloud/v1alpha1".to_owned()),
1273                kind: Some("Materialize".into()),
1274                metadata: Some(
1275                    serde_json::to_value(metadata).expect("ObjectMeta serializes to JSON"),
1276                ),
1277                spec: Some(spec.into()),
1278                status: status.map(Into::into),
1279                extra: serde_json::Map::new(),
1280            }
1281        }
1282    }
1283
1284    impl From<super::v1::PartialMaterializeSpec> for PartialMaterializeSpec {
1285        fn from(spec: super::v1::PartialMaterializeSpec) -> Self {
1286            let super::v1::PartialMaterializeSpec {
1287                environmentd_image_ref,
1288                environmentd_extra_args,
1289                environmentd_extra_env,
1290                environmentd_connection_role_arn,
1291                environmentd_resource_requirements,
1292                environmentd_scratch_volume_storage_requirement,
1293                balancerd_resource_requirements,
1294                console_resource_requirements,
1295                balancerd_replicas,
1296                console_replicas,
1297                service_account_name,
1298                service_account_annotations,
1299                service_account_labels,
1300                pod_annotations,
1301                pod_labels,
1302                force_promote,
1303                force_rollout,
1304                rollout_strategy,
1305                rollout_request_timeout,
1306                backend_secret_name,
1307                authenticator_kind,
1308                enable_rbac,
1309                environment_id,
1310                system_parameter_configmap_name,
1311                balancerd_external_certificate_spec,
1312                console_external_certificate_spec,
1313                internal_certificate_spec,
1314                extra,
1315            } = spec;
1316            Self {
1317                environmentd_image_ref,
1318                environmentd_extra_args,
1319                environmentd_extra_env,
1320                environmentd_iam_role_arn: None,
1321                environmentd_connection_role_arn,
1322                environmentd_resource_requirements,
1323                environmentd_scratch_volume_storage_requirement,
1324                balancerd_resource_requirements,
1325                console_resource_requirements,
1326                balancerd_replicas,
1327                console_replicas,
1328                service_account_name,
1329                service_account_annotations,
1330                service_account_labels,
1331                pod_annotations,
1332                pod_labels,
1333                // Derived from the complete spec. Spliced in by
1334                // `convert_v1_to_v1alpha1` when the input is complete, left
1335                // absent for partial objects so a field manager cannot gain
1336                // ownership of a field it never set.
1337                request_rollout: None,
1338                force_promote,
1339                force_rollout,
1340                in_place_rollout: None,
1341                rollout_strategy,
1342                rollout_request_timeout,
1343                backend_secret_name,
1344                authenticator_kind,
1345                enable_rbac,
1346                environment_id,
1347                system_parameter_configmap_name,
1348                balancerd_external_certificate_spec,
1349                console_external_certificate_spec,
1350                internal_certificate_spec,
1351                extra,
1352            }
1353        }
1354    }
1355
1356    impl From<super::v1::PartialMaterializeStatus> for PartialMaterializeStatus {
1357        fn from(status: super::v1::PartialMaterializeStatus) -> Self {
1358            let super::v1::PartialMaterializeStatus {
1359                resource_id,
1360                active_generation,
1361                last_completed_rollout_environmentd_image_ref,
1362                last_completed_rollout_hash,
1363                requested_rollout_hash: _,
1364                conditions,
1365                extra,
1366            } = status;
1367            Self {
1368                resource_id,
1369                active_generation,
1370                // Derived from the complete object, spliced in by
1371                // `convert_v1_to_v1alpha1` when the input is complete, left
1372                // absent for partial objects so a field manager cannot gain
1373                // ownership of fields it never set.
1374                last_completed_rollout_request: None,
1375                resources_hash: None,
1376                last_completed_rollout_environmentd_image_ref,
1377                last_completed_rollout_hash,
1378                conditions,
1379                extra,
1380            }
1381        }
1382    }
1383
1384    impl From<super::v1::PartialMaterialize> for PartialMaterialize {
1385        fn from(mz: super::v1::PartialMaterialize) -> Self {
1386            let super::v1::PartialMaterialize {
1387                api_version: _,
1388                kind,
1389                metadata,
1390                spec,
1391                status,
1392                extra,
1393            } = mz;
1394            Self {
1395                api_version: Some("materialize.cloud/v1alpha1".to_owned()),
1396                kind,
1397                metadata,
1398                spec: spec.map(Into::into),
1399                status: status.map(Into::into),
1400                extra,
1401            }
1402        }
1403    }
1404}
1405
1406pub mod v1 {
1407    use super::*;
1408
1409    #[derive(
1410        CustomResource,
1411        Clone,
1412        Debug,
1413        Default,
1414        PartialEq,
1415        Deserialize,
1416        Serialize,
1417        JsonSchema
1418    )]
1419    #[serde(rename_all = "camelCase")]
1420    #[kube(
1421        namespaced,
1422        group = "materialize.cloud",
1423        version = "v1",
1424        kind = "Materialize",
1425        singular = "materialize",
1426        plural = "materializes",
1427        shortname = "mzs",
1428        status = "MaterializeStatus",
1429        printcolumn = r#"{"name": "ImageRefRunning", "type": "string", "description": "Reference to the Docker image that is currently in use.", "jsonPath": ".status.lastCompletedRolloutEnvironmentdImageRef", "priority": 1}"#,
1430        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}"#,
1431        printcolumn = r#"{"name": "UpToDate", "type": "string", "description": "Whether the spec has been applied", "jsonPath": ".status.conditions[?(@.type==\"UpToDate\")].status", "priority": 1}"#
1432    )]
1433    pub struct MaterializeSpec {
1434        /// The environmentd image to run.
1435        pub environmentd_image_ref: String,
1436        /// Extra args to pass to the environmentd binary.
1437        pub environmentd_extra_args: Option<Vec<String>>,
1438        /// Extra environment variables to pass to the environmentd binary.
1439        pub environmentd_extra_env: Option<Vec<EnvVar>>,
1440        /// If running in AWS, override the IAM role to use to support
1441        /// the CREATE CONNECTION feature.
1442        pub environmentd_connection_role_arn: Option<String>,
1443        /// Resource requirements for the environmentd pod.
1444        pub environmentd_resource_requirements: Option<ResourceRequirements>,
1445        /// Amount of disk to allocate, if a storage class is provided.
1446        pub environmentd_scratch_volume_storage_requirement: Option<Quantity>,
1447        /// Resource requirements for the balancerd pod.
1448        ///
1449        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1450        pub balancerd_resource_requirements: Option<ResourceRequirements>,
1451        /// Resource requirements for the console pod.
1452        ///
1453        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1454        pub console_resource_requirements: Option<ResourceRequirements>,
1455        /// Number of balancerd pods to create.
1456        ///
1457        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1458        pub balancerd_replicas: Option<i32>,
1459        /// Number of console pods to create.
1460        ///
1461        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1462        pub console_replicas: Option<i32>,
1463
1464        /// Name of the kubernetes service account to use.
1465        /// If not set, we will create one with the same name as this Materialize object.
1466        pub service_account_name: Option<String>,
1467        /// Annotations to apply to the service account.
1468        ///
1469        /// Annotations on service accounts are commonly used by cloud providers for IAM.
1470        /// AWS uses "eks.amazonaws.com/role-arn".
1471        /// Azure uses "azure.workload.identity/client-id", but
1472        /// additionally requires "azure.workload.identity/use": "true" on the pods.
1473        pub service_account_annotations: Option<BTreeMap<String, String>>,
1474        /// Labels to apply to the service account.
1475        pub service_account_labels: Option<BTreeMap<String, String>>,
1476        /// Annotations to apply to the pods.
1477        pub pod_annotations: Option<BTreeMap<String, String>>,
1478        /// Labels to apply to the pods.
1479        pub pod_labels: Option<BTreeMap<String, String>>,
1480
1481        /// If `forcePromote` is set to the same value as the `status.requestedRolloutHash`,
1482        /// current rollout will skip waiting for clusters in the new
1483        /// generation to rehydrate before promoting the new environmentd to
1484        /// leader.
1485        ///
1486        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1487        pub force_promote: Option<String>,
1488        /// This value will force the controller to detect the spec as changed
1489        /// even if no other changes happened. This can be used to force a rollout
1490        /// to a new generation even without making any meaningful changes.
1491        #[serde(default)]
1492        pub force_rollout: Uuid,
1493        /// Rollout strategy to use when upgrading this Materialize instance.
1494        #[serde(default)]
1495        pub rollout_strategy: MaterializeRolloutStrategy,
1496        /// The maximum amount of time a rollout may remain in progress before
1497        /// it is automatically cancelled.
1498        ///
1499        /// While a rollout is in progress, the new generation of `environmentd`
1500        /// runs in a read-only, un-promoted state and holds back compaction via
1501        /// read holds. Leaving it in this state for too long can cause
1502        /// incident-inducing load when it is eventually promoted, so the
1503        /// operator cancels the rollout once this timeout is exceeded: the new
1504        /// generation is torn down and the previously-active generation
1505        /// continues serving. A new rollout can then be triggered by setting
1506        /// `forceRollout` to a new value.
1507        ///
1508        /// This does not apply to the `ImmediatelyPromoteCausingDowntime`
1509        /// rollout strategy or to force-promoted rollouts, since by the time
1510        /// those are in progress the old generation may already be gone.
1511        ///
1512        /// The value is parsed as a human-readable duration, e.g. `24h`,
1513        /// `90m`, or `1h 30m`. Defaults to [`DEFAULT_ROLLOUT_REQUEST_TIMEOUT`]
1514        /// when omitted (the API server fills it in); an unparseable value also
1515        /// falls back to that default.
1516        #[serde(default)]
1517        pub rollout_request_timeout: RolloutRequestTimeout,
1518        /// The name of a secret containing `metadata_backend_url` and `persist_backend_url`.
1519        /// It may also contain `external_login_password_mz_system`, which will be used as
1520        /// the password for the `mz_system` user if `authenticatorKind` is `Password`.
1521        pub backend_secret_name: String,
1522        /// How to authenticate with Materialize.
1523        #[serde(default)]
1524        pub authenticator_kind: AuthenticatorKind,
1525        /// Whether to enable role based access control. Defaults to false.
1526        #[serde(default)]
1527        pub enable_rbac: bool,
1528
1529        /// The value used by environmentd (via the --environment-id flag) to
1530        /// uniquely identify this instance. Must be globally unique, and
1531        /// is required if a license key is not provided.
1532        /// NOTE: This value MUST NOT be changed in an existing instance,
1533        /// since it affects things like the way data is stored in the persist
1534        /// backend.
1535        #[serde(default)]
1536        pub environment_id: Uuid,
1537
1538        /// The name of a ConfigMap containing system parameters in JSON format.
1539        /// The ConfigMap must contain a `system-params.json` key whose value
1540        /// is a valid JSON object containing valid system parameters.
1541        ///
1542        /// Run `SHOW ALL` in SQL to see a subset of configurable system parameters.
1543        ///
1544        /// Example ConfigMap:
1545        /// ```yaml
1546        /// data:
1547        ///   system-params.json: |
1548        ///     {
1549        ///       "max_connections": 1000
1550        ///     }
1551        /// ```
1552        pub system_parameter_configmap_name: Option<String>,
1553
1554        /// The configuration for generating an x509 certificate using cert-manager for balancerd
1555        /// to present to incoming connections.
1556        /// The `dnsNames` and `issuerRef` fields are required.
1557        ///
1558        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1559        pub balancerd_external_certificate_spec: Option<MaterializeCertSpec>,
1560        /// The configuration for generating an x509 certificate using cert-manager for the console
1561        /// to present to incoming connections.
1562        /// The `dnsNames` and `issuerRef` fields are required.
1563        /// Not yet implemented.
1564        ///
1565        /// This field is excluded from the rollout hash and changes will not trigger a rollout.
1566        pub console_external_certificate_spec: Option<MaterializeCertSpec>,
1567        /// The cert-manager Issuer or ClusterIssuer to use for database internal communication.
1568        /// The `issuerRef` field is required.
1569        /// This currently is only used for environmentd, but will eventually support clusterd.
1570        /// Not yet implemented.
1571        pub internal_certificate_spec: Option<MaterializeCertSpec>,
1572    }
1573
1574    impl Materialize {
1575        pub fn generate_rollout_hash(&self) -> String {
1576            let mut hasher = Sha256::new();
1577            // Remove fields that don't affect the resources generated per generation,
1578            // and we don't want to trigger a rollout from.
1579            let spec = MaterializeSpec {
1580                environmentd_image_ref: self.spec.environmentd_image_ref.clone(),
1581                environmentd_extra_args: self.spec.environmentd_extra_args.clone(),
1582                environmentd_extra_env: self.spec.environmentd_extra_env.clone(),
1583                environmentd_connection_role_arn: self
1584                    .spec
1585                    .environmentd_connection_role_arn
1586                    .clone(),
1587                environmentd_resource_requirements: self
1588                    .spec
1589                    .environmentd_resource_requirements
1590                    .clone(),
1591                environmentd_scratch_volume_storage_requirement: self
1592                    .spec
1593                    .environmentd_scratch_volume_storage_requirement
1594                    .clone(),
1595                balancerd_resource_requirements: None,
1596                console_resource_requirements: None,
1597                balancerd_replicas: None,
1598                console_replicas: None,
1599                service_account_name: self.spec.service_account_name.clone(),
1600                service_account_annotations: self.spec.service_account_annotations.clone(),
1601                service_account_labels: self.spec.service_account_labels.clone(),
1602                pod_annotations: self.spec.pod_annotations.clone(),
1603                pod_labels: self.spec.pod_labels.clone(),
1604                force_promote: None,
1605                force_rollout: self.spec.force_rollout,
1606                rollout_strategy: self.spec.rollout_strategy.clone(),
1607                rollout_request_timeout: self.spec.rollout_request_timeout.clone(),
1608                backend_secret_name: self.spec.backend_secret_name.clone(),
1609                authenticator_kind: self.spec.authenticator_kind,
1610                enable_rbac: self.spec.enable_rbac,
1611                environment_id: self.spec.environment_id,
1612                system_parameter_configmap_name: self.spec.system_parameter_configmap_name.clone(),
1613                balancerd_external_certificate_spec: None,
1614                console_external_certificate_spec: None,
1615                internal_certificate_spec: self.spec.internal_certificate_spec.clone(),
1616            };
1617            hasher.update(&serde_json::to_vec(&spec).unwrap());
1618            if let Some(annotation) = self
1619                .metadata
1620                .annotations
1621                .as_ref()
1622                .and_then(|annotations| annotations.get(FORCE_ROLLOUT_ANNOTATION))
1623            {
1624                hasher.update(annotation);
1625            }
1626            format!("{:x}", hasher.finalize())
1627        }
1628
1629        pub fn backend_secret_name(&self) -> String {
1630            self.spec.backend_secret_name.clone()
1631        }
1632
1633        pub fn namespace(&self) -> String {
1634            self.meta().namespace.clone().unwrap()
1635        }
1636
1637        pub fn create_service_account(&self) -> bool {
1638            self.spec.service_account_name.is_none()
1639        }
1640
1641        pub fn service_account_name(&self) -> String {
1642            self.spec
1643                .service_account_name
1644                .clone()
1645                .unwrap_or_else(|| self.name_unchecked())
1646        }
1647
1648        pub fn role_name(&self) -> String {
1649            self.name_unchecked()
1650        }
1651
1652        pub fn role_binding_name(&self) -> String {
1653            self.name_unchecked()
1654        }
1655
1656        pub fn environmentd_statefulset_name(&self, generation: u64) -> String {
1657            self.name_prefixed(&format!("environmentd-{generation}"))
1658        }
1659
1660        pub fn environmentd_app_name(&self) -> String {
1661            "environmentd".to_owned()
1662        }
1663
1664        pub fn environmentd_service_name(&self) -> String {
1665            self.name_prefixed("environmentd")
1666        }
1667
1668        pub fn environmentd_service_internal_fqdn(&self) -> String {
1669            format!(
1670                "{}.{}.svc.cluster.local",
1671                self.environmentd_service_name(),
1672                self.meta().namespace.as_ref().unwrap()
1673            )
1674        }
1675
1676        pub fn environmentd_generation_service_name(&self, generation: u64) -> String {
1677            self.name_prefixed(&format!("environmentd-{generation}"))
1678        }
1679
1680        pub fn balancerd_app_name(&self) -> String {
1681            "balancerd".to_owned()
1682        }
1683
1684        pub fn environmentd_certificate_name(&self) -> String {
1685            self.name_prefixed("environmentd-external")
1686        }
1687
1688        pub fn environmentd_certificate_secret_name(&self) -> String {
1689            self.name_prefixed("environmentd-tls")
1690        }
1691
1692        pub fn balancerd_deployment_name(&self) -> String {
1693            self.name_prefixed("balancerd")
1694        }
1695
1696        pub fn balancerd_service_name(&self) -> String {
1697            self.name_prefixed("balancerd")
1698        }
1699
1700        pub fn console_app_name(&self) -> String {
1701            "console".to_owned()
1702        }
1703
1704        pub fn balancerd_external_certificate_name(&self) -> String {
1705            self.name_prefixed("balancerd-external")
1706        }
1707
1708        pub fn balancerd_external_certificate_secret_name(&self) -> String {
1709            self.name_prefixed("balancerd-external-tls")
1710        }
1711
1712        pub fn balancerd_replicas(&self) -> i32 {
1713            self.spec.balancerd_replicas.unwrap_or(2)
1714        }
1715
1716        pub fn console_replicas(&self) -> i32 {
1717            self.spec.console_replicas.unwrap_or(2)
1718        }
1719
1720        pub fn console_configmap_name(&self) -> String {
1721            self.name_prefixed("console")
1722        }
1723
1724        pub fn console_deployment_name(&self) -> String {
1725            self.name_prefixed("console")
1726        }
1727
1728        pub fn console_service_name(&self) -> String {
1729            self.name_prefixed("console")
1730        }
1731
1732        pub fn console_external_certificate_name(&self) -> String {
1733            self.name_prefixed("console-external")
1734        }
1735
1736        pub fn console_external_certificate_secret_name(&self) -> String {
1737            self.name_prefixed("console-external-tls")
1738        }
1739
1740        pub fn persist_pubsub_service_name(&self, generation: u64) -> String {
1741            self.name_prefixed(&format!("persist-pubsub-{generation}"))
1742        }
1743
1744        pub fn listeners_configmap_name(&self, generation: u64) -> String {
1745            self.name_prefixed(&format!("listeners-{generation}"))
1746        }
1747
1748        pub fn name_prefixed(&self, suffix: &str) -> String {
1749            format!("mz{}-{}", self.resource_id(), suffix)
1750        }
1751
1752        pub fn resource_id(&self) -> &str {
1753            &self.status.as_ref().unwrap().resource_id
1754        }
1755
1756        pub fn system_parameter_configmap_name(&self) -> Option<String> {
1757            self.spec.system_parameter_configmap_name.clone()
1758        }
1759
1760        pub fn environmentd_scratch_volume_storage_requirement(&self) -> Quantity {
1761            self.spec
1762                .environmentd_scratch_volume_storage_requirement
1763                .clone()
1764                .unwrap_or_else(|| {
1765                    self.spec
1766                        .environmentd_resource_requirements
1767                        .as_ref()
1768                        .and_then(|requirements| {
1769                            requirements
1770                                .requests
1771                                .as_ref()
1772                                .or(requirements.limits.as_ref())
1773                        })
1774                        // TODO: in cloud, we've been defaulting to twice the
1775                        // memory limit, but k8s-openapi doesn't seem to
1776                        // provide any way to parse Quantity values, so there
1777                        // isn't an easy way to do arithmetic on it
1778                        .and_then(|requirements| requirements.get("memory").cloned())
1779                        // TODO: is there a better default to use here?
1780                        .unwrap_or_else(|| Quantity("4096Mi".to_string()))
1781                })
1782        }
1783
1784        pub fn environment_id(&self, cloud_provider: &str, region: &str) -> String {
1785            format!(
1786                "{}-{}-{}-0",
1787                cloud_provider, region, self.spec.environment_id,
1788            )
1789        }
1790
1791        pub fn rollout_requested(&self) -> bool {
1792            self.status
1793                .as_ref()
1794                .map(|status| status.last_completed_rollout_hash != status.requested_rollout_hash)
1795                .unwrap_or(false)
1796        }
1797
1798        pub fn set_force_promote(&mut self) {
1799            self.spec.force_promote = Some(self.generate_rollout_hash());
1800        }
1801
1802        pub fn should_force_promote(&self) -> bool {
1803            self.spec.force_promote.as_ref()
1804                == self
1805                    .status
1806                    .as_ref()
1807                    .and_then(|status| status.requested_rollout_hash.as_ref())
1808                || self.spec.rollout_strategy
1809                    == MaterializeRolloutStrategy::ImmediatelyPromoteCausingDowntime
1810        }
1811
1812        pub fn conditions_need_update(&self) -> bool {
1813            let Some(status) = self.status.as_ref() else {
1814                return true;
1815            };
1816            if status.conditions.is_empty() {
1817                return true;
1818            }
1819            for condition in &status.conditions {
1820                if condition.observed_generation != self.meta().generation {
1821                    return true;
1822                }
1823            }
1824            false
1825        }
1826
1827        pub fn is_ready_to_promote(&self, rollout_hash: &str) -> bool {
1828            let Some(status) = self.status.as_ref() else {
1829                return false;
1830            };
1831            if status.conditions.is_empty() {
1832                return false;
1833            }
1834            status
1835                .conditions
1836                .iter()
1837                .any(|condition| condition.reason == "ReadyToPromote")
1838                && status.requested_rollout_hash.as_deref() == Some(rollout_hash)
1839        }
1840
1841        pub fn is_promoting(&self) -> bool {
1842            let Some(status) = self.status.as_ref() else {
1843                return false;
1844            };
1845            if status.conditions.is_empty() {
1846                return false;
1847            }
1848            status
1849                .conditions
1850                .iter()
1851                .any(|condition| condition.reason == "Promoting")
1852        }
1853
1854        pub fn update_in_progress(&self) -> bool {
1855            let Some(status) = self.status.as_ref() else {
1856                return false;
1857            };
1858            if status.conditions.is_empty() {
1859                return false;
1860            }
1861            for condition in &status.conditions {
1862                if condition.type_ == "UpToDate" && condition.status == "Unknown" {
1863                    return true;
1864                }
1865            }
1866            false
1867        }
1868
1869        /// Checks that the given version is greater than or equal
1870        /// to the existing version, if the existing version
1871        /// can be parsed.
1872        pub fn meets_minimum_version(&self, minimum: &Version) -> bool {
1873            let version = parse_image_ref(&self.spec.environmentd_image_ref);
1874            match version {
1875                // Use cmp_precedence() to ignore build metadata per SemVer 2.0.0 spec
1876                Some(version) => version.cmp_precedence(minimum).is_ge(),
1877                // In the rare case that we see an image reference
1878                // that we can't parse, we assume that it satisfies all
1879                // version checks. Usually these are custom images that have
1880                // been by a developer on a branch forked from a recent copy
1881                // of main, and so this works out reasonably well in practice.
1882                None => {
1883                    tracing::warn!(
1884                        image_ref = %self.spec.environmentd_image_ref,
1885                        "failed to parse image ref",
1886                    );
1887                    true
1888                }
1889            }
1890        }
1891
1892        /// This check isn't strictly required since environmentd will still be able to determine
1893        /// if the upgrade is allowed or not. However, doing this check allows us to provide
1894        /// the error as soon as possible and in a more user friendly way.
1895        pub fn is_valid_upgrade_version(active_version: &Version, next_version: &Version) -> bool {
1896            // Don't allow rolling back
1897            // Note: semver comparison handles RC versions correctly:
1898            // v26.0.0-rc.1 < v26.0.0-rc.2 < v26.0.0
1899            // Use cmp_precedence() to ignore build metadata
1900            if next_version.cmp_precedence(active_version) == std::cmp::Ordering::Less {
1901                return false;
1902            }
1903
1904            if active_version.major == 0 {
1905                if next_version.major != active_version.major {
1906                    if next_version.major == 26 {
1907                        // We require customers to upgrade from 0.147.20 (Self Managed 25.2) or v0.164.X (Cloud)
1908                        // before upgrading to 26.0.0
1909
1910                        return (active_version.minor == 147 && active_version.patch >= 20)
1911                            || active_version.minor >= 164;
1912                    } else {
1913                        return false;
1914                    }
1915                }
1916                // Self managed 25.1 to 25.2
1917                if next_version.minor == 147 && active_version.minor == 130 {
1918                    return true;
1919                }
1920                // only allow upgrading a single minor version at a time
1921                return next_version.minor <= active_version.minor + 1;
1922            } else if active_version.major >= 26 {
1923                // For versions 26.X.X and onwards, we deny upgrades past 1 major version of the active version
1924                return next_version.major <= active_version.major + 1;
1925            }
1926
1927            true
1928        }
1929
1930        /// Checks if the current environmentd image ref is within the upgrade window of the last
1931        /// successful rollout.
1932        pub fn within_upgrade_window(&self) -> bool {
1933            let active_environmentd_version = self
1934                .status
1935                .as_ref()
1936                .and_then(|status| {
1937                    status
1938                        .last_completed_rollout_environmentd_image_ref
1939                        .as_ref()
1940                })
1941                .and_then(|image_ref| parse_image_ref(image_ref));
1942
1943            if let (Some(next_environmentd_version), Some(active_environmentd_version)) = (
1944                parse_image_ref(&self.spec.environmentd_image_ref),
1945                active_environmentd_version,
1946            ) {
1947                Self::is_valid_upgrade_version(
1948                    &active_environmentd_version,
1949                    &next_environmentd_version,
1950                )
1951            } else {
1952                // If we fail to parse either version,
1953                // we still allow the upgrade since environmentd will still error if the upgrade is not allowed.
1954                true
1955            }
1956        }
1957
1958        pub fn status(&self) -> MaterializeStatus {
1959            self.status.clone().unwrap_or_else(|| {
1960                let mut status = MaterializeStatus::default();
1961
1962                status.resource_id = new_resource_id();
1963
1964                // If we're creating the initial status on an un-soft-deleted
1965                // Environment we need to ensure that the last active generation
1966                // is restored, otherwise the env will crash loop indefinitely
1967                // as its catalog would have durably recorded a greater generation
1968                if let Some(last_active_generation) = self
1969                    .annotations()
1970                    .get(LAST_KNOWN_ACTIVE_GENERATION_ANNOTATION)
1971                {
1972                    status.active_generation = last_active_generation
1973                        .parse()
1974                        .expect("valid int generation");
1975                }
1976
1977                // Initialize the last completed rollout environmentd image ref to
1978                // the current image ref if not already set.
1979                status.last_completed_rollout_environmentd_image_ref =
1980                    Some(self.spec.environmentd_image_ref.clone());
1981
1982                status
1983            })
1984        }
1985    }
1986
1987    #[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema, PartialEq)]
1988    #[serde(rename_all = "camelCase")]
1989    pub struct MaterializeStatus {
1990        /// Resource identifier used as a name prefix to avoid pod name collisions.
1991        pub resource_id: String,
1992        /// The generation of Materialize pods actively capable of servicing requests.
1993        pub active_generation: u64,
1994        /// The image ref of the environmentd image that was last successfully rolled out.
1995        /// Used to deny upgrades past 1 major version from the last successful rollout.
1996        /// When None, we upgrade anyways.
1997        pub last_completed_rollout_environmentd_image_ref: Option<String>,
1998        /// The last completed rollout's requestedRolloutHash.
1999        pub last_completed_rollout_hash: Option<String>,
2000        /// Hash of a subset of the Materialize spec and other fields.
2001        /// This is used to determine when the spec has changed and we need to rollout.
2002        pub requested_rollout_hash: Option<String>,
2003        pub conditions: Vec<Condition>,
2004    }
2005
2006    impl MaterializeStatus {
2007        pub fn needs_update(&self, other: &Self) -> bool {
2008            let now = Timestamp::now();
2009            let mut a = self.clone();
2010            for condition in &mut a.conditions {
2011                condition.last_transition_time = Time(now);
2012            }
2013            let mut b = other.clone();
2014            for condition in &mut b.conditions {
2015                condition.last_transition_time = Time(now);
2016            }
2017            a != b
2018        }
2019    }
2020
2021    impl ManagedResource for Materialize {
2022        fn default_labels(&self) -> BTreeMap<String, String> {
2023            BTreeMap::from_iter([
2024                (
2025                    "materialize.cloud/organization-name".to_owned(),
2026                    self.name_unchecked(),
2027                ),
2028                (
2029                    "materialize.cloud/organization-namespace".to_owned(),
2030                    self.namespace(),
2031                ),
2032                (
2033                    "materialize.cloud/mz-resource-id".to_owned(),
2034                    self.resource_id().to_owned(),
2035                ),
2036            ])
2037        }
2038
2039        fn app_name(&self) -> Option<&str> {
2040            Some("environmentd")
2041        }
2042    }
2043
2044    impl From<v1alpha1::Materialize> for Materialize {
2045        fn from(value: v1alpha1::Materialize) -> Self {
2046            let is_promoting = value.is_promoting();
2047            let service_account_annotations = if let Some(environmentd_iam_role_arn) =
2048                value.spec.environmentd_iam_role_arn
2049            {
2050                let mut annotations = value.spec.service_account_annotations.unwrap_or_default();
2051                annotations
2052                    .entry("eks.amazonaws.com/role-arn".to_owned())
2053                    .or_insert(environmentd_iam_role_arn);
2054                Some(annotations)
2055            } else {
2056                value.spec.service_account_annotations
2057            };
2058            let mut mz = Materialize {
2059                metadata: value.metadata,
2060                spec: MaterializeSpec {
2061                    environmentd_image_ref: value.spec.environmentd_image_ref,
2062                    environmentd_extra_args: value.spec.environmentd_extra_args,
2063                    environmentd_extra_env: value.spec.environmentd_extra_env,
2064                    environmentd_connection_role_arn: value.spec.environmentd_connection_role_arn,
2065                    environmentd_resource_requirements: value
2066                        .spec
2067                        .environmentd_resource_requirements,
2068                    environmentd_scratch_volume_storage_requirement: value
2069                        .spec
2070                        .environmentd_scratch_volume_storage_requirement,
2071                    balancerd_resource_requirements: value.spec.balancerd_resource_requirements,
2072                    console_resource_requirements: value.spec.console_resource_requirements,
2073                    balancerd_replicas: value.spec.balancerd_replicas,
2074                    console_replicas: value.spec.console_replicas,
2075                    service_account_name: value.spec.service_account_name,
2076                    service_account_annotations,
2077                    service_account_labels: value.spec.service_account_labels,
2078                    pod_annotations: value.spec.pod_annotations,
2079                    pod_labels: value.spec.pod_labels,
2080                    force_promote: if value.spec.force_promote.is_empty()
2081                        || &value.spec.force_promote == "00000000-0000-0000-0000-000000000000"
2082                    {
2083                        None
2084                    } else {
2085                        Some(value.spec.force_promote.to_string())
2086                    },
2087                    force_rollout: value.spec.force_rollout,
2088                    rollout_strategy: value.spec.rollout_strategy,
2089                    rollout_request_timeout: value.spec.rollout_request_timeout,
2090                    backend_secret_name: value.spec.backend_secret_name,
2091                    authenticator_kind: value.spec.authenticator_kind,
2092                    enable_rbac: value.spec.enable_rbac,
2093                    environment_id: value.spec.environment_id,
2094                    system_parameter_configmap_name: value.spec.system_parameter_configmap_name,
2095                    balancerd_external_certificate_spec: value
2096                        .spec
2097                        .balancerd_external_certificate_spec,
2098                    console_external_certificate_spec: value.spec.console_external_certificate_spec,
2099                    internal_certificate_spec: value.spec.internal_certificate_spec,
2100                },
2101                status: None,
2102            };
2103            let calculated_rollout_hash = mz.generate_rollout_hash();
2104            let last_completed_rollout_hash = match value
2105                .status
2106                .as_ref()
2107                .and_then(|status| status.last_completed_rollout_hash.to_owned())
2108            {
2109                Some(last_completed_rollout_hash) => Some(last_completed_rollout_hash),
2110                None => {
2111                    let currently_rolling_out = value
2112                        .status
2113                        .as_ref()
2114                        .map(|status| {
2115                            status.last_completed_rollout_request != value.spec.request_rollout
2116                                // If this is the first apply,
2117                                // these could both be nil and we still need to do a rollout.
2118                                || status.last_completed_rollout_request.is_nil()
2119                        })
2120                        .unwrap_or(true);
2121                    if currently_rolling_out {
2122                        // If they store a change, we're going to start over on a new rollout.
2123                        None
2124                    } else {
2125                        Some(calculated_rollout_hash.clone())
2126                    }
2127                }
2128            };
2129            let requested_rollout_hash = if is_promoting {
2130                None
2131            } else {
2132                Some(calculated_rollout_hash)
2133            };
2134            mz.status = value.status.map(|status| MaterializeStatus {
2135                resource_id: status.resource_id,
2136                active_generation: status.active_generation,
2137                last_completed_rollout_environmentd_image_ref: status
2138                    .last_completed_rollout_environmentd_image_ref,
2139                last_completed_rollout_hash,
2140                requested_rollout_hash,
2141                conditions: status.conditions,
2142            });
2143            mz
2144        }
2145    }
2146
2147    /// Partial mirror of [`MaterializeSpec`] for field-wise conversion of
2148    /// objects that may be incomplete. See [`super::convert_v1alpha1_to_v1`].
2149    ///
2150    /// Each field is a [`PartialField`], distinguishing absent from explicit
2151    /// null from a value, and unknown fields pass through via `extra`, so
2152    /// serializing reproduces exactly what was deserialized. Adding a field
2153    /// to [`MaterializeSpec`] breaks the exhaustive destructures in the
2154    /// `From` impls below until its conversion is decided.
2155    #[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
2156    #[serde(rename_all = "camelCase")]
2157    pub struct PartialMaterializeSpec {
2158        #[serde(
2159            default,
2160            with = "double_option",
2161            skip_serializing_if = "Option::is_none"
2162        )]
2163        pub environmentd_image_ref: PartialField,
2164        #[serde(
2165            default,
2166            with = "double_option",
2167            skip_serializing_if = "Option::is_none"
2168        )]
2169        pub environmentd_extra_args: PartialField,
2170        #[serde(
2171            default,
2172            with = "double_option",
2173            skip_serializing_if = "Option::is_none"
2174        )]
2175        pub environmentd_extra_env: PartialField,
2176        #[serde(
2177            default,
2178            with = "double_option",
2179            skip_serializing_if = "Option::is_none"
2180        )]
2181        pub environmentd_connection_role_arn: PartialField,
2182        #[serde(
2183            default,
2184            with = "double_option",
2185            skip_serializing_if = "Option::is_none"
2186        )]
2187        pub environmentd_resource_requirements: PartialField,
2188        #[serde(
2189            default,
2190            with = "double_option",
2191            skip_serializing_if = "Option::is_none"
2192        )]
2193        pub environmentd_scratch_volume_storage_requirement: PartialField,
2194        #[serde(
2195            default,
2196            with = "double_option",
2197            skip_serializing_if = "Option::is_none"
2198        )]
2199        pub balancerd_resource_requirements: PartialField,
2200        #[serde(
2201            default,
2202            with = "double_option",
2203            skip_serializing_if = "Option::is_none"
2204        )]
2205        pub console_resource_requirements: PartialField,
2206        #[serde(
2207            default,
2208            with = "double_option",
2209            skip_serializing_if = "Option::is_none"
2210        )]
2211        pub balancerd_replicas: PartialField,
2212        #[serde(
2213            default,
2214            with = "double_option",
2215            skip_serializing_if = "Option::is_none"
2216        )]
2217        pub console_replicas: PartialField,
2218        #[serde(
2219            default,
2220            with = "double_option",
2221            skip_serializing_if = "Option::is_none"
2222        )]
2223        pub service_account_name: PartialField,
2224        #[serde(
2225            default,
2226            with = "double_option",
2227            skip_serializing_if = "Option::is_none"
2228        )]
2229        pub service_account_annotations: PartialField,
2230        #[serde(
2231            default,
2232            with = "double_option",
2233            skip_serializing_if = "Option::is_none"
2234        )]
2235        pub service_account_labels: PartialField,
2236        #[serde(
2237            default,
2238            with = "double_option",
2239            skip_serializing_if = "Option::is_none"
2240        )]
2241        pub pod_annotations: PartialField,
2242        #[serde(
2243            default,
2244            with = "double_option",
2245            skip_serializing_if = "Option::is_none"
2246        )]
2247        pub pod_labels: PartialField,
2248        #[serde(
2249            default,
2250            with = "double_option",
2251            skip_serializing_if = "Option::is_none"
2252        )]
2253        pub force_promote: PartialField,
2254        #[serde(
2255            default,
2256            with = "double_option",
2257            skip_serializing_if = "Option::is_none"
2258        )]
2259        pub force_rollout: PartialField,
2260        #[serde(
2261            default,
2262            with = "double_option",
2263            skip_serializing_if = "Option::is_none"
2264        )]
2265        pub rollout_strategy: PartialField,
2266        #[serde(
2267            default,
2268            with = "double_option",
2269            skip_serializing_if = "Option::is_none"
2270        )]
2271        pub rollout_request_timeout: PartialField,
2272        #[serde(
2273            default,
2274            with = "double_option",
2275            skip_serializing_if = "Option::is_none"
2276        )]
2277        pub backend_secret_name: PartialField,
2278        #[serde(
2279            default,
2280            with = "double_option",
2281            skip_serializing_if = "Option::is_none"
2282        )]
2283        pub authenticator_kind: PartialField,
2284        #[serde(
2285            default,
2286            with = "double_option",
2287            skip_serializing_if = "Option::is_none"
2288        )]
2289        pub enable_rbac: PartialField,
2290        #[serde(
2291            default,
2292            with = "double_option",
2293            skip_serializing_if = "Option::is_none"
2294        )]
2295        pub environment_id: PartialField,
2296        #[serde(
2297            default,
2298            with = "double_option",
2299            skip_serializing_if = "Option::is_none"
2300        )]
2301        pub system_parameter_configmap_name: PartialField,
2302        #[serde(
2303            default,
2304            with = "double_option",
2305            skip_serializing_if = "Option::is_none"
2306        )]
2307        pub balancerd_external_certificate_spec: PartialField,
2308        #[serde(
2309            default,
2310            with = "double_option",
2311            skip_serializing_if = "Option::is_none"
2312        )]
2313        pub console_external_certificate_spec: PartialField,
2314        #[serde(
2315            default,
2316            with = "double_option",
2317            skip_serializing_if = "Option::is_none"
2318        )]
2319        pub internal_certificate_spec: PartialField,
2320        #[serde(flatten)]
2321        pub extra: serde_json::Map<String, serde_json::Value>,
2322    }
2323
2324    /// Partial mirror of [`MaterializeStatus`], see [`PartialMaterializeSpec`].
2325    #[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
2326    #[serde(rename_all = "camelCase")]
2327    pub struct PartialMaterializeStatus {
2328        #[serde(
2329            default,
2330            with = "double_option",
2331            skip_serializing_if = "Option::is_none"
2332        )]
2333        pub resource_id: PartialField,
2334        #[serde(
2335            default,
2336            with = "double_option",
2337            skip_serializing_if = "Option::is_none"
2338        )]
2339        pub active_generation: PartialField,
2340        #[serde(
2341            default,
2342            with = "double_option",
2343            skip_serializing_if = "Option::is_none"
2344        )]
2345        pub last_completed_rollout_environmentd_image_ref: PartialField,
2346        #[serde(
2347            default,
2348            with = "double_option",
2349            skip_serializing_if = "Option::is_none"
2350        )]
2351        pub last_completed_rollout_hash: PartialField,
2352        #[serde(
2353            default,
2354            with = "double_option",
2355            skip_serializing_if = "Option::is_none"
2356        )]
2357        pub requested_rollout_hash: PartialField,
2358        #[serde(
2359            default,
2360            with = "double_option",
2361            skip_serializing_if = "Option::is_none"
2362        )]
2363        pub conditions: PartialField,
2364        #[serde(flatten)]
2365        pub extra: serde_json::Map<String, serde_json::Value>,
2366    }
2367
2368    /// Partial mirror of [`Materialize`], see [`PartialMaterializeSpec`].
2369    #[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
2370    #[serde(rename_all = "camelCase")]
2371    pub struct PartialMaterialize {
2372        #[serde(default, skip_serializing_if = "Option::is_none")]
2373        pub api_version: Option<String>,
2374        #[serde(default, skip_serializing_if = "Option::is_none")]
2375        pub kind: Option<serde_json::Value>,
2376        #[serde(default, skip_serializing_if = "Option::is_none")]
2377        pub metadata: Option<serde_json::Value>,
2378        #[serde(default, skip_serializing_if = "Option::is_none")]
2379        pub spec: Option<PartialMaterializeSpec>,
2380        #[serde(default, skip_serializing_if = "Option::is_none")]
2381        pub status: Option<PartialMaterializeStatus>,
2382        #[serde(flatten)]
2383        pub extra: serde_json::Map<String, serde_json::Value>,
2384    }
2385
2386    impl From<MaterializeSpec> for PartialMaterializeSpec {
2387        fn from(spec: MaterializeSpec) -> Self {
2388            let MaterializeSpec {
2389                environmentd_image_ref,
2390                environmentd_extra_args,
2391                environmentd_extra_env,
2392                environmentd_connection_role_arn,
2393                environmentd_resource_requirements,
2394                environmentd_scratch_volume_storage_requirement,
2395                balancerd_resource_requirements,
2396                console_resource_requirements,
2397                balancerd_replicas,
2398                console_replicas,
2399                service_account_name,
2400                service_account_annotations,
2401                service_account_labels,
2402                pod_annotations,
2403                pod_labels,
2404                force_promote,
2405                force_rollout,
2406                rollout_strategy,
2407                rollout_request_timeout,
2408                backend_secret_name,
2409                authenticator_kind,
2410                enable_rbac,
2411                environment_id,
2412                system_parameter_configmap_name,
2413                balancerd_external_certificate_spec,
2414                console_external_certificate_spec,
2415                internal_certificate_spec,
2416            } = spec;
2417            Self {
2418                environmentd_image_ref: present(environmentd_image_ref),
2419                environmentd_extra_args: present_opt(environmentd_extra_args),
2420                environmentd_extra_env: present_opt(environmentd_extra_env),
2421                environmentd_connection_role_arn: present_opt(environmentd_connection_role_arn),
2422                environmentd_resource_requirements: present_opt(environmentd_resource_requirements),
2423                environmentd_scratch_volume_storage_requirement: present_opt(
2424                    environmentd_scratch_volume_storage_requirement,
2425                ),
2426                balancerd_resource_requirements: present_opt(balancerd_resource_requirements),
2427                console_resource_requirements: present_opt(console_resource_requirements),
2428                balancerd_replicas: present_opt(balancerd_replicas),
2429                console_replicas: present_opt(console_replicas),
2430                service_account_name: present_opt(service_account_name),
2431                service_account_annotations: present_opt(service_account_annotations),
2432                service_account_labels: present_opt(service_account_labels),
2433                pod_annotations: present_opt(pod_annotations),
2434                pod_labels: present_opt(pod_labels),
2435                force_promote: present_opt(force_promote),
2436                force_rollout: present(force_rollout),
2437                rollout_strategy: present(rollout_strategy),
2438                rollout_request_timeout: present(rollout_request_timeout),
2439                backend_secret_name: present(backend_secret_name),
2440                authenticator_kind: present(authenticator_kind),
2441                enable_rbac: present(enable_rbac),
2442                environment_id: present(environment_id),
2443                system_parameter_configmap_name: present_opt(system_parameter_configmap_name),
2444                balancerd_external_certificate_spec: present_opt(
2445                    balancerd_external_certificate_spec,
2446                ),
2447                console_external_certificate_spec: present_opt(console_external_certificate_spec),
2448                internal_certificate_spec: present_opt(internal_certificate_spec),
2449                extra: serde_json::Map::new(),
2450            }
2451        }
2452    }
2453
2454    impl From<MaterializeStatus> for PartialMaterializeStatus {
2455        fn from(status: MaterializeStatus) -> Self {
2456            let MaterializeStatus {
2457                resource_id,
2458                active_generation,
2459                last_completed_rollout_environmentd_image_ref,
2460                last_completed_rollout_hash,
2461                requested_rollout_hash,
2462                conditions,
2463            } = status;
2464            Self {
2465                resource_id: present(resource_id),
2466                active_generation: present(active_generation),
2467                last_completed_rollout_environmentd_image_ref: present_opt(
2468                    last_completed_rollout_environmentd_image_ref,
2469                ),
2470                last_completed_rollout_hash: present_opt(last_completed_rollout_hash),
2471                requested_rollout_hash: present_opt(requested_rollout_hash),
2472                conditions: present(conditions),
2473                extra: serde_json::Map::new(),
2474            }
2475        }
2476    }
2477
2478    impl From<Materialize> for PartialMaterialize {
2479        fn from(mz: Materialize) -> Self {
2480            let Materialize {
2481                metadata,
2482                spec,
2483                status,
2484            } = mz;
2485            Self {
2486                api_version: Some("materialize.cloud/v1".to_owned()),
2487                kind: Some("Materialize".into()),
2488                metadata: Some(
2489                    serde_json::to_value(metadata).expect("ObjectMeta serializes to JSON"),
2490                ),
2491                spec: Some(spec.into()),
2492                status: status.map(Into::into),
2493                extra: serde_json::Map::new(),
2494            }
2495        }
2496    }
2497
2498    impl From<super::v1alpha1::PartialMaterializeSpec> for PartialMaterializeSpec {
2499        fn from(spec: super::v1alpha1::PartialMaterializeSpec) -> Self {
2500            let super::v1alpha1::PartialMaterializeSpec {
2501                environmentd_image_ref,
2502                environmentd_extra_args,
2503                environmentd_extra_env,
2504                environmentd_iam_role_arn,
2505                environmentd_connection_role_arn,
2506                environmentd_resource_requirements,
2507                environmentd_scratch_volume_storage_requirement,
2508                balancerd_resource_requirements,
2509                console_resource_requirements,
2510                balancerd_replicas,
2511                console_replicas,
2512                service_account_name,
2513                service_account_annotations,
2514                service_account_labels,
2515                pod_annotations,
2516                pod_labels,
2517                request_rollout: _,
2518                force_promote,
2519                force_rollout,
2520                in_place_rollout: _,
2521                rollout_strategy,
2522                rollout_request_timeout,
2523                backend_secret_name,
2524                authenticator_kind,
2525                enable_rbac,
2526                environment_id,
2527                system_parameter_configmap_name,
2528                balancerd_external_certificate_spec,
2529                console_external_certificate_spec,
2530                internal_certificate_spec,
2531                extra,
2532            } = spec;
2533            let service_account_annotations = merge_environmentd_iam_role_arn(
2534                service_account_annotations,
2535                environmentd_iam_role_arn,
2536            );
2537            // "" and the nil UUID mean "not force promoting" in v1alpha1,
2538            // which v1 spells as an absent field.
2539            let force_promote = match force_promote {
2540                Some(Some(value)) if value == "" || value == NIL_UUID_STR => None,
2541                other => other,
2542            };
2543            Self {
2544                environmentd_image_ref,
2545                environmentd_extra_args,
2546                environmentd_extra_env,
2547                environmentd_connection_role_arn,
2548                environmentd_resource_requirements,
2549                environmentd_scratch_volume_storage_requirement,
2550                balancerd_resource_requirements,
2551                console_resource_requirements,
2552                balancerd_replicas,
2553                console_replicas,
2554                service_account_name,
2555                service_account_annotations,
2556                service_account_labels,
2557                pod_annotations,
2558                pod_labels,
2559                force_promote,
2560                force_rollout,
2561                rollout_strategy,
2562                rollout_request_timeout,
2563                backend_secret_name,
2564                authenticator_kind,
2565                enable_rbac,
2566                environment_id,
2567                system_parameter_configmap_name,
2568                balancerd_external_certificate_spec,
2569                console_external_certificate_spec,
2570                internal_certificate_spec,
2571                extra,
2572            }
2573        }
2574    }
2575
2576    impl From<super::v1alpha1::PartialMaterializeStatus> for PartialMaterializeStatus {
2577        fn from(status: super::v1alpha1::PartialMaterializeStatus) -> Self {
2578            let super::v1alpha1::PartialMaterializeStatus {
2579                resource_id,
2580                active_generation,
2581                last_completed_rollout_request: _,
2582                last_completed_rollout_environmentd_image_ref,
2583                resources_hash: _,
2584                last_completed_rollout_hash,
2585                conditions,
2586                extra,
2587            } = status;
2588            Self {
2589                resource_id,
2590                active_generation,
2591                last_completed_rollout_environmentd_image_ref,
2592                last_completed_rollout_hash,
2593                // Derived from the complete object, spliced in by
2594                // `convert_v1alpha1_to_v1` when the input is complete, left
2595                // absent for partial objects so a field manager cannot gain
2596                // ownership of a field it never set.
2597                requested_rollout_hash: None,
2598                conditions,
2599                extra,
2600            }
2601        }
2602    }
2603
2604    impl From<super::v1alpha1::PartialMaterialize> for PartialMaterialize {
2605        fn from(mz: super::v1alpha1::PartialMaterialize) -> Self {
2606            let super::v1alpha1::PartialMaterialize {
2607                api_version: _,
2608                kind,
2609                metadata,
2610                spec,
2611                status,
2612                extra,
2613            } = mz;
2614            Self {
2615                api_version: Some("materialize.cloud/v1".to_owned()),
2616                kind,
2617                metadata,
2618                spec: spec.map(Into::into),
2619                status: status.map(Into::into),
2620                extra,
2621            }
2622        }
2623    }
2624}
2625
2626/// The nil UUID rendered the way it appears in JSON-encoded specs.
2627const NIL_UUID_STR: &str = "00000000-0000-0000-0000-000000000000";
2628
2629/// One field of a partial object: absent (`None`), explicit null
2630/// (`Some(None)`), or a value (`Some(Some(_))`).
2631///
2632/// Values are untyped [`serde_json::Value`]s because conversion must be a
2633/// total function over anything schema-shaped, including values that do not
2634/// validate. The type system is used for field *names*: the partial mirror
2635/// structs convert via exhaustive destructures, so adding a field to a spec
2636/// without deciding its conversion does not compile.
2637pub type PartialField = Option<Option<serde_json::Value>>;
2638
2639/// Serde adapter for [`PartialField`] distinguishing explicit null from an
2640/// absent field. Use with `#[serde(default, with = "double_option",
2641/// skip_serializing_if = "Option::is_none")]`.
2642mod double_option {
2643    use serde::{Deserialize, Deserializer, Serialize, Serializer};
2644
2645    pub fn deserialize<'de, T, D>(deserializer: D) -> Result<Option<Option<T>>, D::Error>
2646    where
2647        T: Deserialize<'de>,
2648        D: Deserializer<'de>,
2649    {
2650        Option::<T>::deserialize(deserializer).map(Some)
2651    }
2652
2653    pub fn serialize<T, S>(value: &Option<Option<T>>, serializer: S) -> Result<S::Ok, S::Error>
2654    where
2655        T: Serialize,
2656        S: Serializer,
2657    {
2658        match value {
2659            Some(inner) => inner.serialize(serializer),
2660            // Unreachable under skip_serializing_if = "Option::is_none".
2661            None => serializer.serialize_none(),
2662        }
2663    }
2664}
2665
2666/// A [`PartialField`] holding a value that is always present in the typed
2667/// struct.
2668fn present<T: Serialize>(value: T) -> PartialField {
2669    Some(Some(
2670        serde_json::to_value(value).expect("CRD field serializes to JSON"),
2671    ))
2672}
2673
2674/// A [`PartialField`] from a typed optional field, absent when `None`.
2675fn present_opt<T: Serialize>(value: Option<T>) -> PartialField {
2676    value.map(|value| Some(serde_json::to_value(value).expect("CRD field serializes to JSON")))
2677}
2678
2679/// Merges a v1alpha1 `environmentdIamRoleArn` value into the
2680/// `serviceAccountAnnotations` map as `eks.amazonaws.com/role-arn`, matching
2681/// the typed conversion. An existing annotation wins. Annotations that are
2682/// not an object pass through untouched.
2683fn merge_environmentd_iam_role_arn(
2684    annotations: PartialField,
2685    role_arn: PartialField,
2686) -> PartialField {
2687    let Some(Some(role_arn)) = role_arn else {
2688        return annotations;
2689    };
2690    let mut map = match annotations {
2691        Some(Some(serde_json::Value::Object(map))) => map,
2692        Some(Some(other)) => return Some(Some(other)),
2693        Some(None) | None => serde_json::Map::new(),
2694    };
2695    map.entry("eks.amazonaws.com/role-arn").or_insert(role_arn);
2696    Some(Some(serde_json::Value::Object(map)))
2697}
2698
2699/// Converts a JSON-encoded v1alpha1 Materialize object to v1, for use by the
2700/// CRD conversion webhook.
2701///
2702/// Output fields are exactly the input fields, mapped field-wise, plus
2703/// derived fields when the input is complete. Server-side apply round-trips
2704/// each field manager's owned subset of an object through the conversion
2705/// webhook when it reconciles managed fields recorded at another version.
2706/// Those subsets routinely lack required fields, so conversion must accept
2707/// partial objects, and it must not add fields the input did not carry,
2708/// since a field manager must not gain ownership of fields it never set.
2709///
2710/// Fields derived from the complete spec (`status.requestedRolloutHash`
2711/// here, `spec.requestRollout` in the other direction) cannot be computed
2712/// from a subset, so they are spliced in from the typed conversion only when
2713/// the input deserializes as a complete object. The request carries no
2714/// indicator of partialness, so a subset that happens to contain every
2715/// required field also gains the derived fields. That is the irreducible
2716/// ambiguity, and it is limited to exactly those fields.
2717pub fn convert_v1alpha1_to_v1(
2718    value: serde_json::Value,
2719) -> Result<serde_json::Value, anyhow::Error> {
2720    let complete = serde_json::from_value::<v1alpha1::Materialize>(value.clone()).ok();
2721    let partial: v1alpha1::PartialMaterialize = serde_json::from_value(value)?;
2722    let mut converted = v1::PartialMaterialize::from(partial);
2723    if let Some(complete) = complete {
2724        let typed = v1::Materialize::from(complete);
2725        if let (Some(status), Some(typed_status)) = (converted.status.as_mut(), typed.status) {
2726            status.requested_rollout_hash = present_opt(typed_status.requested_rollout_hash);
2727            status.last_completed_rollout_hash =
2728                present_opt(typed_status.last_completed_rollout_hash);
2729        }
2730    }
2731    Ok(serde_json::to_value(converted)?)
2732}
2733
2734/// Converts a JSON-encoded v1 Materialize object to v1alpha1, for use by the
2735/// CRD conversion webhook.
2736///
2737/// See [`convert_v1alpha1_to_v1`] for the conversion contract. In this
2738/// direction the derived fields are `spec.requestRollout` and
2739/// `status.lastCompletedRolloutRequest`, both deterministic UUIDs derived
2740/// from rollout hashes of the complete object, and `status.resourcesHash`,
2741/// which has no v1 counterpart but is required by the v1alpha1 schema
2742/// whenever a status is present, so complete conversions must carry the
2743/// typed conversion's placeholder to stay valid at the storage version.
2744pub fn convert_v1_to_v1alpha1(
2745    value: serde_json::Value,
2746) -> Result<serde_json::Value, anyhow::Error> {
2747    let complete = serde_json::from_value::<v1::Materialize>(value.clone()).ok();
2748    let partial: v1::PartialMaterialize = serde_json::from_value(value)?;
2749    let mut converted = v1alpha1::PartialMaterialize::from(partial);
2750    if let Some(complete) = complete {
2751        let typed = v1alpha1::Materialize::from(complete);
2752        if let Some(spec) = converted.spec.as_mut() {
2753            spec.request_rollout = present(typed.spec.request_rollout);
2754        }
2755        if let (Some(status), Some(typed_status)) = (converted.status.as_mut(), typed.status) {
2756            status.last_completed_rollout_request =
2757                present(typed_status.last_completed_rollout_request);
2758            status.resources_hash = present(typed_status.resources_hash);
2759        }
2760    }
2761    Ok(serde_json::to_value(converted)?)
2762}
2763
2764fn parse_image_ref(image_ref: &str) -> Option<Version> {
2765    image_ref
2766        .rsplit_once(':')
2767        .and_then(|(_repo, tag)| tag.strip_prefix('v'))
2768        .and_then(|tag| {
2769            // To work around Docker tag restrictions, build metadata in
2770            // a Docker tag is delimited by `--` rather than the SemVer
2771            // `+` delimiter. So we need to swap the delimiter back to
2772            // `+` before parsing it as SemVer.
2773            let tag = tag.replace("--", "+");
2774            Version::parse(&tag).ok()
2775        })
2776}
2777
2778#[cfg(test)]
2779mod tests {
2780    use std::time::Duration;
2781
2782    use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
2783    use k8s_openapi::jiff::Timestamp;
2784    use kube::core::ObjectMeta;
2785    use semver::Version;
2786
2787    use super::v1alpha1::{Materialize, MaterializeSpec, MaterializeStatus};
2788    use super::{DEFAULT_ROLLOUT_REQUEST_TIMEOUT, FORCE_ROLLOUT_ANNOTATION, RolloutRequestTimeout};
2789
2790    #[mz_ore::test]
2791    #[cfg_attr(miri, ignore)] // can't call foreign function `sha256_compress` on OS `linux`
2792    fn force_rollout_annotation_forces_new_generation() {
2793        // The force-rollout annotation must feed into both the v1 rollout
2794        // hash (so that a rollout is requested) and the v1alpha1 force
2795        // rollout value stamped onto the generated statefulset (so that the
2796        // requested rollout actually creates a new generation rather than
2797        // completing as a no-op).
2798        let mut mz = super::v1::Materialize {
2799            spec: super::v1::MaterializeSpec {
2800                environmentd_image_ref: "materialize/environmentd:v26.0.0".to_owned(),
2801                ..Default::default()
2802            },
2803            metadata: ObjectMeta::default(),
2804            status: None,
2805        };
2806        let hash_without_annotation = mz.generate_rollout_hash();
2807        let force_without_annotation = Materialize::from(mz.clone()).force_rollout_value();
2808
2809        mz.metadata.annotations = Some(std::collections::BTreeMap::from_iter([(
2810            FORCE_ROLLOUT_ANNOTATION.to_owned(),
2811            "a4b56cbb-a13e-4f95-8a9d-425c9ba28576".to_owned(),
2812        )]));
2813        assert_ne!(mz.generate_rollout_hash(), hash_without_annotation);
2814        assert_ne!(
2815            Materialize::from(mz.clone()).force_rollout_value(),
2816            force_without_annotation
2817        );
2818
2819        // Changing the annotation's value changes both again.
2820        let hash = mz.generate_rollout_hash();
2821        let force = Materialize::from(mz.clone()).force_rollout_value();
2822        mz.metadata.annotations.as_mut().unwrap().insert(
2823            FORCE_ROLLOUT_ANNOTATION.to_owned(),
2824            "3f61bf8d-0714-462c-8b3b-3d9a68d0bcba".to_owned(),
2825        );
2826        assert_ne!(mz.generate_rollout_hash(), hash);
2827        assert_ne!(Materialize::from(mz.clone()).force_rollout_value(), force);
2828    }
2829
2830    #[mz_ore::test]
2831    fn meets_minimum_version() {
2832        let mut mz = Materialize {
2833            spec: MaterializeSpec {
2834                environmentd_image_ref:
2835                    "materialize/environmentd:devel-47116c24b8d0df33d3f60a9ee476aa8d7bce5953"
2836                        .to_owned(),
2837                ..Default::default()
2838            },
2839            metadata: ObjectMeta {
2840                ..Default::default()
2841            },
2842            status: None,
2843        };
2844
2845        // true cases
2846        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2847        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0".to_owned();
2848        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2849        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.35.0".to_owned();
2850        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2851        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.3".to_owned();
2852        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2853        mz.spec.environmentd_image_ref = "materialize/environmentd@41af286dc0b172ed2f1ca934fd2278de4a1192302ffa07087cea2682e7d372e3".to_owned();
2854        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2855        mz.spec.environmentd_image_ref = "my.private.registry:5000:v0.34.3".to_owned();
2856        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2857        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.asdf.0".to_owned();
2858        assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2859        mz.spec.environmentd_image_ref =
2860            "materialize/environmentd:v0.146.0-dev.0--pr.g5a05a9e4ba873be8adaa528644aaae6e4c7cd29b"
2861                .to_owned();
2862        assert!(mz.meets_minimum_version(&Version::parse("0.146.0-dev.0").unwrap()));
2863
2864        // false cases
2865        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0-dev".to_owned();
2866        assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2867        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.33.0".to_owned();
2868        assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2869        mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0".to_owned();
2870        assert!(!mz.meets_minimum_version(&Version::parse("1.0.0").unwrap()));
2871        mz.spec.environmentd_image_ref = "my.private.registry:5000:v0.33.3".to_owned();
2872        assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
2873    }
2874
2875    #[mz_ore::test]
2876    fn within_upgrade_window() {
2877        let mut mz = Materialize {
2878            spec: MaterializeSpec {
2879                environmentd_image_ref: "materialize/environmentd:v26.0.0".to_owned(),
2880                ..Default::default()
2881            },
2882            metadata: ObjectMeta {
2883                ..Default::default()
2884            },
2885            status: Some(MaterializeStatus {
2886                last_completed_rollout_environmentd_image_ref: Some(
2887                    "materialize/environmentd:v26.0.0".to_owned(),
2888                ),
2889                ..Default::default()
2890            }),
2891        };
2892
2893        // Pass: upgrading from 26.0.0 to 27.7.3 (within 1 major version)
2894        mz.spec.environmentd_image_ref = "materialize/environmentd:v27.7.3".to_owned();
2895        assert!(mz.within_upgrade_window());
2896
2897        // Pass: upgrading from 26.0.0 to 27.7.8-dev.0 (within 1 major version, pre-release)
2898        mz.spec.environmentd_image_ref = "materialize/environmentd:v27.7.8-dev.0".to_owned();
2899        assert!(mz.within_upgrade_window());
2900
2901        // Fail: upgrading from 26.0.0 to 28.0.1 (more than 1 major version)
2902        mz.spec.environmentd_image_ref = "materialize/environmentd:v28.0.1".to_owned();
2903        assert!(!mz.within_upgrade_window());
2904
2905        // Pass: upgrading from 26.0.0 to 28.0.1.not_a_valid_version (invalid version, defaults to true)
2906        mz.spec.environmentd_image_ref =
2907            "materialize/environmentd:v28.0.1.not_a_valid_version".to_owned();
2908        assert!(mz.within_upgrade_window());
2909
2910        // Pass: upgrading from 0.164.0 to 26.1.0 (self managed 25.2 to 26.0)
2911        mz.status
2912            .as_mut()
2913            .unwrap()
2914            .last_completed_rollout_environmentd_image_ref =
2915            Some("materialize/environmentd:v0.147.20".to_owned());
2916        mz.spec.environmentd_image_ref = "materialize/environmentd:v26.1.0".to_owned();
2917        assert!(mz.within_upgrade_window());
2918
2919        // Pass: upgrading from 26.11.0-dev.0+b to 26.11.0-dev.0+a (same major.minor.patch.prerelease, different build metadata)
2920        mz.status
2921            .as_mut()
2922            .unwrap()
2923            .last_completed_rollout_environmentd_image_ref =
2924            Some("materialize/environmentd:v26.11.0-dev.0+b".to_owned());
2925        mz.spec.environmentd_image_ref = "materialize/environmentd:v26.11.0-dev.0+a".to_owned();
2926        assert!(mz.within_upgrade_window());
2927    }
2928
2929    #[mz_ore::test]
2930    fn is_valid_upgrade_version() {
2931        let success_tests = [
2932            (Version::new(0, 83, 0), Version::new(0, 83, 0)),
2933            (Version::new(0, 83, 0), Version::new(0, 84, 0)),
2934            (Version::new(0, 9, 0), Version::new(0, 10, 0)),
2935            (Version::new(0, 99, 0), Version::new(0, 100, 0)),
2936            (Version::new(0, 83, 0), Version::new(0, 83, 1)),
2937            (Version::new(0, 83, 0), Version::new(0, 83, 2)),
2938            (Version::new(0, 83, 2), Version::new(0, 83, 10)),
2939            // 0.147.20 to 26.0.0 represents the Self Managed 25.2 to 26.0 upgrade
2940            (Version::new(0, 147, 20), Version::new(26, 0, 0)),
2941            (Version::new(0, 164, 0), Version::new(26, 0, 0)),
2942            (Version::new(26, 0, 0), Version::new(26, 1, 0)),
2943            (Version::new(26, 5, 3), Version::new(26, 10, 0)),
2944            (Version::new(0, 130, 0), Version::new(0, 147, 0)),
2945        ];
2946        for (active_version, next_version) in success_tests {
2947            assert!(
2948                Materialize::is_valid_upgrade_version(&active_version, &next_version),
2949                "v{active_version} can upgrade to v{next_version}"
2950            );
2951        }
2952
2953        let failure_tests = [
2954            (Version::new(0, 83, 0), Version::new(0, 82, 0)),
2955            (Version::new(0, 83, 3), Version::new(0, 83, 2)),
2956            (Version::new(0, 83, 3), Version::new(1, 83, 3)),
2957            (Version::new(0, 83, 0), Version::new(0, 85, 0)),
2958            (Version::new(26, 0, 0), Version::new(28, 0, 0)),
2959            (Version::new(0, 130, 0), Version::new(26, 1, 0)),
2960            // Disallow anything before 0.147.20 to upgrade
2961            (Version::new(0, 147, 1), Version::new(26, 0, 0)),
2962            // Disallow anything between 0.148.0 and 0.164.0 to upgrade
2963            (Version::new(0, 148, 0), Version::new(26, 0, 0)),
2964        ];
2965        for (active_version, next_version) in failure_tests {
2966            assert!(
2967                !Materialize::is_valid_upgrade_version(&active_version, &next_version),
2968                "v{active_version} can't upgrade to v{next_version}"
2969            );
2970        }
2971    }
2972
2973    #[mz_ore::test]
2974    fn rollout_request_timeout() {
2975        let mz_with = |timeout: &str| Materialize {
2976            spec: MaterializeSpec {
2977                rollout_request_timeout: RolloutRequestTimeout(timeout.to_owned()),
2978                ..Default::default()
2979            },
2980            metadata: ObjectMeta::default(),
2981            status: None,
2982        };
2983
2984        // The default const is a valid duration and resolves to 24h.
2985        let default = humantime::parse_duration(DEFAULT_ROLLOUT_REQUEST_TIMEOUT).unwrap();
2986        assert_eq!(default, Duration::from_secs(24 * 60 * 60));
2987
2988        // The field's Default (used by `MaterializeSpec::default()` and serde's
2989        // `#[serde(default)]`) is the 24h default, with no empty intermediate.
2990        assert_eq!(
2991            RolloutRequestTimeout::default().0,
2992            DEFAULT_ROLLOUT_REQUEST_TIMEOUT
2993        );
2994        assert_eq!(
2995            Materialize {
2996                spec: MaterializeSpec::default(),
2997                metadata: ObjectMeta::default(),
2998                status: None,
2999            }
3000            .rollout_request_timeout(),
3001            default
3002        );
3003
3004        // Parseable values are honored.
3005        assert_eq!(
3006            mz_with("1h").rollout_request_timeout(),
3007            Duration::from_secs(60 * 60)
3008        );
3009        assert_eq!(
3010            mz_with("90m").rollout_request_timeout(),
3011            Duration::from_secs(90 * 60)
3012        );
3013        assert_eq!(
3014            mz_with("1h 30m").rollout_request_timeout(),
3015            Duration::from_secs(90 * 60)
3016        );
3017        // Unparseable values fall back to the default.
3018        assert_eq!(mz_with("not a duration").rollout_request_timeout(), default);
3019    }
3020
3021    #[mz_ore::test]
3022    fn rollout_request_timeout_schema_default() {
3023        // The default must be surfaced in the generated CRD's OpenAPI schema
3024        // (not just in the Rust helper), so the Kubernetes API server defaults
3025        // omitted fields and `kubectl explain` shows it.
3026        let crd = serde_json::to_value(<Materialize as kube::CustomResourceExt>::crd())
3027            .expect("CRD serializes");
3028        let default = &crd["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"]
3029            ["properties"]["rolloutRequestTimeout"]["default"];
3030        assert_eq!(
3031            default,
3032            &serde_json::json!(DEFAULT_ROLLOUT_REQUEST_TIMEOUT),
3033            "rolloutRequestTimeout schema default missing/wrong in generated CRD",
3034        );
3035    }
3036
3037    #[mz_ore::test]
3038    fn rollout_in_progress_since() {
3039        let now = Timestamp::now();
3040        let condition = |type_: &str, status: &str| Condition {
3041            type_: type_.to_owned(),
3042            status: status.to_owned(),
3043            last_transition_time: Time(now),
3044            message: String::new(),
3045            observed_generation: None,
3046            reason: "Test".to_owned(),
3047        };
3048        let mz_with = |conditions: Vec<Condition>| Materialize {
3049            spec: MaterializeSpec::default(),
3050            metadata: ObjectMeta::default(),
3051            status: Some(MaterializeStatus {
3052                conditions,
3053                ..Default::default()
3054            }),
3055        };
3056
3057        // No status at all.
3058        let mz = Materialize {
3059            spec: MaterializeSpec::default(),
3060            metadata: ObjectMeta::default(),
3061            status: None,
3062        };
3063        assert_eq!(mz.rollout_in_progress_since(), None);
3064
3065        // A timeout-eligible rollout in progress is signalled by an `Unknown`
3066        // `UpToDate` condition (the Applying and ReadyToPromote phases).
3067        assert_eq!(
3068            mz_with(vec![condition("UpToDate", "Unknown")]).rollout_in_progress_since(),
3069            Some(now)
3070        );
3071
3072        // The `Promoting` phase is also `Unknown`, but must NOT be reported:
3073        // once promoting, the rollout can no longer be cancelled by the
3074        // timeout.
3075        assert_eq!(
3076            mz_with(vec![Condition {
3077                reason: "Promoting".to_owned(),
3078                ..condition("UpToDate", "Unknown")
3079            }])
3080            .rollout_in_progress_since(),
3081            None
3082        );
3083
3084        // A settled rollout (True/False) is not in progress.
3085        assert_eq!(
3086            mz_with(vec![condition("UpToDate", "True")]).rollout_in_progress_since(),
3087            None
3088        );
3089        assert_eq!(
3090            mz_with(vec![condition("UpToDate", "False")]).rollout_in_progress_since(),
3091            None
3092        );
3093    }
3094
3095    #[mz_ore::test]
3096    fn up_to_date_transition_time() {
3097        // Two distinct, fixed instants so we can tell "carried the old
3098        // timestamp" apart from "reset to now".
3099        let stored = Timestamp::from_second(1_000).unwrap();
3100        let now = Timestamp::from_second(2_000).unwrap();
3101
3102        let condition = |status: &str| Condition {
3103            type_: "UpToDate".to_owned(),
3104            status: status.to_owned(),
3105            last_transition_time: Time(stored),
3106            message: String::new(),
3107            observed_generation: None,
3108            reason: "Test".to_owned(),
3109        };
3110        let mz_with = |conditions: Vec<Condition>| Materialize {
3111            spec: MaterializeSpec::default(),
3112            metadata: ObjectMeta::default(),
3113            status: Some(MaterializeStatus {
3114                conditions,
3115                ..Default::default()
3116            }),
3117        };
3118
3119        // No prior condition: use `now`.
3120        let mz = Materialize {
3121            spec: MaterializeSpec::default(),
3122            metadata: ObjectMeta::default(),
3123            status: None,
3124        };
3125        assert_eq!(mz.up_to_date_transition_time("Unknown", now), now);
3126
3127        // Same status as the prior condition: carry its timestamp forward, so
3128        // consecutive same-status phases (Applying -> ReadyToPromote) share one
3129        // timer.
3130        assert_eq!(
3131            mz_with(vec![condition("Unknown")]).up_to_date_transition_time("Unknown", now),
3132            stored
3133        );
3134
3135        // Status changed: reset to `now`.
3136        assert_eq!(
3137            mz_with(vec![condition("Unknown")]).up_to_date_transition_time("True", now),
3138            now
3139        );
3140    }
3141
3142    #[mz_ore::test]
3143    fn active_environmentd_image_ref() {
3144        const OLD: &str = "materialize/environmentd:v26.0.0";
3145        const NEW: &str = "materialize/environmentd:v27.0.0";
3146
3147        let mz_with = |spec_image: &str, status: Option<MaterializeStatus>| Materialize {
3148            spec: MaterializeSpec {
3149                environmentd_image_ref: spec_image.to_owned(),
3150                ..Default::default()
3151            },
3152            metadata: ObjectMeta::default(),
3153            status,
3154        };
3155
3156        // No status yet (pre-initial-reconcile): fall back to spec.
3157        let mz = mz_with(NEW, None);
3158        assert_eq!(mz.active_environmentd_image_ref(), NEW);
3159
3160        // Status present but last_completed_rollout_environmentd_image_ref
3161        // unset (e.g. resource upgraded from older orchestratord that didn't
3162        // populate the field): fall back to spec.
3163        let mz = mz_with(
3164            NEW,
3165            Some(MaterializeStatus {
3166                last_completed_rollout_environmentd_image_ref: None,
3167                ..Default::default()
3168            }),
3169        );
3170        assert_eq!(mz.active_environmentd_image_ref(), NEW);
3171
3172        // Steady state: spec image == last completed image. Either source is
3173        // fine; the method must return that image.
3174        let mz = mz_with(
3175            NEW,
3176            Some(MaterializeStatus {
3177                last_completed_rollout_environmentd_image_ref: Some(NEW.to_owned()),
3178                ..Default::default()
3179            }),
3180        );
3181        assert_eq!(mz.active_environmentd_image_ref(), NEW);
3182
3183        // DEP-42 / mid-rollout: spec image == NEW but last_completed_* still
3184        // holds OLD — either because the user canceled the rollout by
3185        // reverting only requestRollout, or because the new generation has
3186        // not yet been promoted. The active environmentd is still OLD, so
3187        // downstream resources must track OLD. Without this method,
3188        // balancerd would inherit the spec's NEW image while environmentd
3189        // still runs OLD, leaving balancerd pods skewed from the running
3190        // env.
3191        let mz = mz_with(
3192            NEW,
3193            Some(MaterializeStatus {
3194                last_completed_rollout_environmentd_image_ref: Some(OLD.to_owned()),
3195                ..Default::default()
3196            }),
3197        );
3198        assert_eq!(mz.active_environmentd_image_ref(), OLD);
3199    }
3200
3201    // Server-side apply round-trips each field manager's owned subset of an
3202    // object through the conversion webhook. Such subsets lack required
3203    // fields, so conversion must map them field-wise instead of erroring.
3204    #[mz_ore::test]
3205    fn convert_partial_v1alpha1_to_v1() {
3206        let subset = serde_json::json!({
3207            "apiVersion": "materialize.cloud/v1alpha1",
3208            "kind": "Materialize",
3209            "metadata": {"name": "mz", "namespace": "materialize"},
3210            "spec": {
3211                "environmentdIamRoleArn": "arn:aws:iam::123456789012:role/mz",
3212                "requestRollout": "1e6ef7bc-bff2-4bd4-90dc-eda5697bd0e6",
3213                "inPlaceRollout": false,
3214                "forcePromote": "",
3215                "serviceAccountLabels": {"team": "data"},
3216            },
3217            "status": {
3218                "activeGeneration": 3,
3219                "lastCompletedRolloutRequest": "1e6ef7bc-bff2-4bd4-90dc-eda5697bd0e6",
3220                "resourcesHash": "abc123",
3221            },
3222        });
3223        let converted = super::convert_v1alpha1_to_v1(subset).unwrap();
3224        assert_eq!(converted["apiVersion"], "materialize.cloud/v1");
3225        assert_eq!(converted["kind"], "Materialize");
3226        assert_eq!(converted["metadata"]["name"], "mz");
3227        let spec = converted["spec"].as_object().unwrap();
3228        assert!(!spec.contains_key("requestRollout"));
3229        assert!(!spec.contains_key("inPlaceRollout"));
3230        assert!(!spec.contains_key("forcePromote"));
3231        assert!(!spec.contains_key("environmentdIamRoleArn"));
3232        assert_eq!(
3233            spec["serviceAccountAnnotations"]["eks.amazonaws.com/role-arn"],
3234            "arn:aws:iam::123456789012:role/mz"
3235        );
3236        assert_eq!(spec["serviceAccountLabels"]["team"], "data");
3237        let status = converted["status"].as_object().unwrap();
3238        assert_eq!(status["activeGeneration"], 3);
3239        assert!(!status.contains_key("lastCompletedRolloutRequest"));
3240        assert!(!status.contains_key("resourcesHash"));
3241        assert!(!status.contains_key("requestedRolloutHash"));
3242
3243        // A meaningful forcePromote value survives the conversion.
3244        let subset = serde_json::json!({
3245            "apiVersion": "materialize.cloud/v1alpha1",
3246            "kind": "Materialize",
3247            "metadata": {"name": "mz"},
3248            "spec": {"forcePromote": "1e6ef7bc-bff2-4bd4-90dc-eda5697bd0e6"},
3249        });
3250        let converted = super::convert_v1alpha1_to_v1(subset).unwrap();
3251        assert_eq!(
3252            converted["spec"]["forcePromote"],
3253            "1e6ef7bc-bff2-4bd4-90dc-eda5697bd0e6"
3254        );
3255    }
3256
3257    #[mz_ore::test]
3258    fn convert_partial_v1_to_v1alpha1() {
3259        let subset = serde_json::json!({
3260            "apiVersion": "materialize.cloud/v1",
3261            "kind": "Materialize",
3262            "metadata": {"name": "mz"},
3263            "spec": {"environmentdImageRef": "materialize/environmentd:v26.0.0"},
3264            "status": {"requestedRolloutHash": "abc123", "activeGeneration": 1},
3265        });
3266        let converted = super::convert_v1_to_v1alpha1(subset).unwrap();
3267        assert_eq!(converted["apiVersion"], "materialize.cloud/v1alpha1");
3268        assert_eq!(
3269            converted["spec"]["environmentdImageRef"],
3270            "materialize/environmentd:v26.0.0"
3271        );
3272        let status = converted["status"].as_object().unwrap();
3273        assert_eq!(status["activeGeneration"], 1);
3274        assert!(!status.contains_key("requestedRolloutHash"));
3275        assert!(!status.contains_key("resourcesHash"));
3276        // Derived fields must not be invented for partial objects, a field
3277        // manager must not gain ownership of fields it never set.
3278        assert!(
3279            !converted["spec"]
3280                .as_object()
3281                .unwrap()
3282                .contains_key("requestRollout")
3283        );
3284    }
3285
3286    #[mz_ore::test]
3287    #[cfg_attr(miri, ignore)] // can't call foreign function `sha256_compress` on OS `linux`
3288    fn convert_full_v1alpha1_to_v1_derives_fields() {
3289        let mz = Materialize {
3290            metadata: ObjectMeta {
3291                name: Some("mz".to_owned()),
3292                namespace: Some("materialize".to_owned()),
3293                ..Default::default()
3294            },
3295            spec: MaterializeSpec {
3296                environmentd_image_ref: "materialize/environmentd:v26.0.0".to_owned(),
3297                backend_secret_name: "mz-backend".to_owned(),
3298                ..Default::default()
3299            },
3300            status: Some(MaterializeStatus::default()),
3301        };
3302        let value = serde_json::to_value(&mz).unwrap();
3303        let converted = super::convert_v1alpha1_to_v1(value).unwrap();
3304        assert_eq!(converted["apiVersion"], "materialize.cloud/v1");
3305        // Complete objects take the typed conversion, which computes fields
3306        // derived from the whole spec.
3307        assert!(converted["status"]["requestedRolloutHash"].is_string());
3308    }
3309
3310    #[mz_ore::test]
3311    fn convert_rejects_non_objects() {
3312        assert!(super::convert_v1alpha1_to_v1(serde_json::json!("not an object")).is_err());
3313        assert!(super::convert_v1_to_v1alpha1(serde_json::json!(42)).is_err());
3314        assert!(
3315            super::convert_v1alpha1_to_v1(serde_json::json!({"spec": "not an object"})).is_err()
3316        );
3317    }
3318
3319    #[mz_ore::test]
3320    fn convert_preserves_null_vs_absent() {
3321        let subset = serde_json::json!({
3322            "apiVersion": "materialize.cloud/v1alpha1",
3323            "kind": "Materialize",
3324            "metadata": {"name": "mz"},
3325            "spec": {
3326                "environmentdExtraArgs": null,
3327                "backendSecretName": "mz-backend",
3328            },
3329        });
3330        let converted = super::convert_v1alpha1_to_v1(subset).unwrap();
3331        let spec = converted["spec"].as_object().unwrap();
3332        assert!(spec.contains_key("environmentdExtraArgs"));
3333        assert!(spec["environmentdExtraArgs"].is_null());
3334        assert!(!spec.contains_key("consoleReplicas"));
3335    }
3336
3337    // A subset containing every required field deserializes as a complete
3338    // object and takes the derived-field splice, but the conversion must
3339    // still not add defaults or nulls for fields the subset did not carry.
3340    // Field managers must not gain ownership of fields they never set.
3341    #[mz_ore::test]
3342    #[cfg_attr(miri, ignore)] // can't call foreign function `sha256_compress` on OS `linux`
3343    fn convert_faithful_output_for_required_field_subsets() {
3344        let subset = serde_json::json!({
3345            "apiVersion": "materialize.cloud/v1alpha1",
3346            "kind": "Materialize",
3347            "metadata": {"name": "mz"},
3348            "spec": {
3349                "environmentdImageRef": "materialize/environmentd:v26.0.0",
3350                "backendSecretName": "mz-backend",
3351            },
3352        });
3353        let converted = super::convert_v1alpha1_to_v1(subset).unwrap();
3354        let spec = converted["spec"].as_object().unwrap();
3355        let mut keys: Vec<_> = spec.keys().cloned().collect();
3356        keys.sort();
3357        assert_eq!(keys, ["backendSecretName", "environmentdImageRef"]);
3358    }
3359
3360    #[mz_ore::test]
3361    fn convert_passes_unknown_fields_through() {
3362        let subset = serde_json::json!({
3363            "apiVersion": "materialize.cloud/v1alpha1",
3364            "kind": "Materialize",
3365            "metadata": {"name": "mz"},
3366            "spec": {"someFutureField": {"a": 1}},
3367            "someTopLevelField": true,
3368        });
3369        let converted = super::convert_v1alpha1_to_v1(subset).unwrap();
3370        assert_eq!(converted["spec"]["someFutureField"]["a"], 1);
3371        assert_eq!(converted["someTopLevelField"], true);
3372    }
3373
3374    #[mz_ore::test]
3375    #[cfg_attr(miri, ignore)] // can't call foreign function `sha256_compress` on OS `linux`
3376    fn convert_full_v1_to_v1alpha1_derives_request_rollout() {
3377        let mz = super::v1::Materialize {
3378            metadata: ObjectMeta {
3379                name: Some("mz".to_owned()),
3380                namespace: Some("materialize".to_owned()),
3381                ..Default::default()
3382            },
3383            spec: super::v1::MaterializeSpec {
3384                environmentd_image_ref: "materialize/environmentd:v26.0.0".to_owned(),
3385                backend_secret_name: "mz-backend".to_owned(),
3386                ..Default::default()
3387            },
3388            status: Some(super::v1::MaterializeStatus::default()),
3389        };
3390        let expected = Materialize::from(mz.clone()).spec.request_rollout;
3391        let converted = super::convert_v1_to_v1alpha1(serde_json::to_value(&mz).unwrap()).unwrap();
3392        assert_eq!(converted["apiVersion"], "materialize.cloud/v1alpha1");
3393        assert_eq!(
3394            converted["spec"]["requestRollout"],
3395            expected.hyphenated().to_string()
3396        );
3397        // The status fields required by the v1alpha1 schema but absent from
3398        // v1 must be spliced in, else the converted object fails validation
3399        // at the storage version.
3400        let status = converted["status"].as_object().unwrap();
3401        assert!(status["resourcesHash"].is_string());
3402        assert!(status["lastCompletedRolloutRequest"].is_string());
3403    }
3404
3405    // The `From<Materialize> for PartialMaterialize` impls are the
3406    // compile-time guard tying the partial mirrors to the typed structs.
3407    // Their output must be faithful: unset optional fields stay absent
3408    // rather than becoming nulls.
3409    #[mz_ore::test]
3410    fn partial_mirror_from_typed_omits_unset_fields() {
3411        let mz = Materialize {
3412            metadata: ObjectMeta::default(),
3413            spec: MaterializeSpec {
3414                environmentd_image_ref: "materialize/environmentd:v26.0.0".to_owned(),
3415                backend_secret_name: "mz-backend".to_owned(),
3416                ..Default::default()
3417            },
3418            status: None,
3419        };
3420        let value = serde_json::to_value(super::v1alpha1::PartialMaterialize::from(mz)).unwrap();
3421        let spec = value["spec"].as_object().unwrap();
3422        assert!(!spec.contains_key("balancerdReplicas"));
3423        for (key, field_value) in spec {
3424            assert!(!field_value.is_null(), "unexpected null for {key}");
3425        }
3426        assert!(!value.as_object().unwrap().contains_key("status"));
3427    }
3428}