Skip to main content

mz_cloud_resources/crd/
materialize.rs

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