1use 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 #[default]
41 WaitUntilReady,
42
43 ManuallyPromote,
71
72 ImmediatelyPromoteCausingDowntime,
83}
84
85pub const DEFAULT_ROLLOUT_REQUEST_TIMEOUT: &str = "24h";
90
91#[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 pub environmentd_image_ref: String,
141 pub environmentd_extra_args: Option<Vec<String>>,
143 pub environmentd_extra_env: Option<Vec<EnvVar>>,
145 #[kube(deprecated)]
154 pub environmentd_iam_role_arn: Option<String>,
155 pub environmentd_connection_role_arn: Option<String>,
158 pub environmentd_resource_requirements: Option<ResourceRequirements>,
160 pub environmentd_scratch_volume_storage_requirement: Option<Quantity>,
162 pub balancerd_resource_requirements: Option<ResourceRequirements>,
164 pub console_resource_requirements: Option<ResourceRequirements>,
166 pub balancerd_replicas: Option<i32>,
168 pub console_replicas: Option<i32>,
170
171 pub service_account_name: Option<String>,
174 pub service_account_annotations: Option<BTreeMap<String, String>>,
181 pub service_account_labels: Option<BTreeMap<String, String>>,
183 pub pod_annotations: Option<BTreeMap<String, String>>,
185 pub pod_labels: Option<BTreeMap<String, String>>,
187
188 #[serde(default)]
200 pub request_rollout: Uuid,
201 #[serde(default)]
206 pub force_promote: String,
207 #[serde(default)]
214 pub force_rollout: Uuid,
215 #[kube(deprecated)]
219 #[serde(default)]
220 pub in_place_rollout: bool,
221 #[serde(default)]
223 pub rollout_strategy: MaterializeRolloutStrategy,
224 #[serde(default)]
245 pub rollout_request_timeout: RolloutRequestTimeout,
246 pub backend_secret_name: String,
251 #[serde(default)]
253 pub authenticator_kind: AuthenticatorKind,
254 #[serde(default)]
256 pub enable_rbac: bool,
257
258 #[serde(default)]
265 pub environment_id: Uuid,
266
267 pub system_parameter_configmap_name: Option<String>,
282
283 pub balancerd_external_certificate_spec: Option<MaterializeCertSpec>,
287 pub console_external_certificate_spec: Option<MaterializeCertSpec>,
292 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 .and_then(|requirements| requirements.get("memory").cloned())
450 .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 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 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 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 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 pub fn meets_minimum_version(&self, minimum: &Version) -> bool {
635 let version = parse_image_ref(&self.spec.environmentd_image_ref);
636 match version {
637 Some(version) => version.cmp_precedence(minimum).is_ge(),
639 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 pub fn is_valid_upgrade_version(active_version: &Version, next_version: &Version) -> bool {
658 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 return (active_version.minor == 147 && active_version.patch >= 20)
672 || active_version.minor >= 164;
673 } else {
674 return false;
675 }
676 }
677 if next_version.minor == 147 && active_version.minor == 130 {
679 return true;
680 }
681 return next_version.minor <= active_version.minor + 1;
683 } else if active_version.major >= 26 {
684 return next_version.major <= active_version.major + 1;
686 }
687
688 true
689 }
690
691 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 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 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 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 pub resource_id: String,
753 pub active_generation: u64,
755 pub last_completed_rollout_request: Uuid,
757 pub last_completed_rollout_environmentd_image_ref: Option<String>,
761 pub resources_hash: String,
766 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 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 last_completed_rollout_request: status
867 .last_completed_rollout_hash
868 .as_ref()
869 .map(|hash| Uuid::new_v5(&Uuid::NAMESPACE_OID, hash.as_bytes()))
870 .unwrap_or(Uuid::nil()),
871 last_completed_rollout_hash: status.last_completed_rollout_hash,
872 resources_hash: "".to_owned(),
873 }),
874 }
875 }
876 }
877}
878
879pub mod v1 {
880 use super::*;
881
882 #[derive(
883 CustomResource,
884 Clone,
885 Debug,
886 Default,
887 PartialEq,
888 Deserialize,
889 Serialize,
890 JsonSchema
891 )]
892 #[serde(rename_all = "camelCase")]
893 #[kube(
894 namespaced,
895 group = "materialize.cloud",
896 version = "v1",
897 kind = "Materialize",
898 singular = "materialize",
899 plural = "materializes",
900 shortname = "mzs",
901 status = "MaterializeStatus",
902 printcolumn = r#"{"name": "ImageRefRunning", "type": "string", "description": "Reference to the Docker image that is currently in use.", "jsonPath": ".status.lastCompletedRolloutEnvironmentdImageRef", "priority": 1}"#,
903 printcolumn = r#"{"name": "ImageRefToDeploy", "type": "string", "description": "Reference to the Docker image which will be deployed on the next rollout.", "jsonPath": ".spec.environmentdImageRef", "priority": 1}"#,
904 printcolumn = r#"{"name": "UpToDate", "type": "string", "description": "Whether the spec has been applied", "jsonPath": ".status.conditions[?(@.type==\"UpToDate\")].status", "priority": 1}"#
905 )]
906 pub struct MaterializeSpec {
907 pub environmentd_image_ref: String,
909 pub environmentd_extra_args: Option<Vec<String>>,
911 pub environmentd_extra_env: Option<Vec<EnvVar>>,
913 pub environmentd_connection_role_arn: Option<String>,
916 pub environmentd_resource_requirements: Option<ResourceRequirements>,
918 pub environmentd_scratch_volume_storage_requirement: Option<Quantity>,
920 pub balancerd_resource_requirements: Option<ResourceRequirements>,
924 pub console_resource_requirements: Option<ResourceRequirements>,
928 pub balancerd_replicas: Option<i32>,
932 pub console_replicas: Option<i32>,
936
937 pub service_account_name: Option<String>,
940 pub service_account_annotations: Option<BTreeMap<String, String>>,
947 pub service_account_labels: Option<BTreeMap<String, String>>,
949 pub pod_annotations: Option<BTreeMap<String, String>>,
951 pub pod_labels: Option<BTreeMap<String, String>>,
953
954 pub force_promote: Option<String>,
961 #[serde(default)]
965 pub force_rollout: Uuid,
966 #[serde(default)]
968 pub rollout_strategy: MaterializeRolloutStrategy,
969 #[serde(default)]
990 pub rollout_request_timeout: RolloutRequestTimeout,
991 pub backend_secret_name: String,
995 #[serde(default)]
997 pub authenticator_kind: AuthenticatorKind,
998 #[serde(default)]
1000 pub enable_rbac: bool,
1001
1002 #[serde(default)]
1009 pub environment_id: Uuid,
1010
1011 pub system_parameter_configmap_name: Option<String>,
1026
1027 pub balancerd_external_certificate_spec: Option<MaterializeCertSpec>,
1033 pub console_external_certificate_spec: Option<MaterializeCertSpec>,
1040 pub internal_certificate_spec: Option<MaterializeCertSpec>,
1045 }
1046
1047 impl Materialize {
1048 pub fn generate_rollout_hash(&self) -> String {
1049 let mut hasher = Sha256::new();
1050 let spec = MaterializeSpec {
1053 environmentd_image_ref: self.spec.environmentd_image_ref.clone(),
1054 environmentd_extra_args: self.spec.environmentd_extra_args.clone(),
1055 environmentd_extra_env: self.spec.environmentd_extra_env.clone(),
1056 environmentd_connection_role_arn: self
1057 .spec
1058 .environmentd_connection_role_arn
1059 .clone(),
1060 environmentd_resource_requirements: self
1061 .spec
1062 .environmentd_resource_requirements
1063 .clone(),
1064 environmentd_scratch_volume_storage_requirement: self
1065 .spec
1066 .environmentd_scratch_volume_storage_requirement
1067 .clone(),
1068 balancerd_resource_requirements: None,
1069 console_resource_requirements: None,
1070 balancerd_replicas: None,
1071 console_replicas: None,
1072 service_account_name: self.spec.service_account_name.clone(),
1073 service_account_annotations: self.spec.service_account_annotations.clone(),
1074 service_account_labels: self.spec.service_account_labels.clone(),
1075 pod_annotations: self.spec.pod_annotations.clone(),
1076 pod_labels: self.spec.pod_labels.clone(),
1077 force_promote: None,
1078 force_rollout: self.spec.force_rollout,
1079 rollout_strategy: self.spec.rollout_strategy.clone(),
1080 rollout_request_timeout: self.spec.rollout_request_timeout.clone(),
1081 backend_secret_name: self.spec.backend_secret_name.clone(),
1082 authenticator_kind: self.spec.authenticator_kind,
1083 enable_rbac: self.spec.enable_rbac,
1084 environment_id: self.spec.environment_id,
1085 system_parameter_configmap_name: self.spec.system_parameter_configmap_name.clone(),
1086 balancerd_external_certificate_spec: None,
1087 console_external_certificate_spec: None,
1088 internal_certificate_spec: self.spec.internal_certificate_spec.clone(),
1089 };
1090 hasher.update(&serde_json::to_vec(&spec).unwrap());
1091 if let Some(annotation) = self
1092 .metadata
1093 .annotations
1094 .as_ref()
1095 .and_then(|annotations| annotations.get(FORCE_ROLLOUT_ANNOTATION))
1096 {
1097 hasher.update(annotation);
1098 }
1099 format!("{:x}", hasher.finalize())
1100 }
1101
1102 pub fn backend_secret_name(&self) -> String {
1103 self.spec.backend_secret_name.clone()
1104 }
1105
1106 pub fn namespace(&self) -> String {
1107 self.meta().namespace.clone().unwrap()
1108 }
1109
1110 pub fn create_service_account(&self) -> bool {
1111 self.spec.service_account_name.is_none()
1112 }
1113
1114 pub fn service_account_name(&self) -> String {
1115 self.spec
1116 .service_account_name
1117 .clone()
1118 .unwrap_or_else(|| self.name_unchecked())
1119 }
1120
1121 pub fn role_name(&self) -> String {
1122 self.name_unchecked()
1123 }
1124
1125 pub fn role_binding_name(&self) -> String {
1126 self.name_unchecked()
1127 }
1128
1129 pub fn environmentd_statefulset_name(&self, generation: u64) -> String {
1130 self.name_prefixed(&format!("environmentd-{generation}"))
1131 }
1132
1133 pub fn environmentd_app_name(&self) -> String {
1134 "environmentd".to_owned()
1135 }
1136
1137 pub fn environmentd_service_name(&self) -> String {
1138 self.name_prefixed("environmentd")
1139 }
1140
1141 pub fn environmentd_service_internal_fqdn(&self) -> String {
1142 format!(
1143 "{}.{}.svc.cluster.local",
1144 self.environmentd_service_name(),
1145 self.meta().namespace.as_ref().unwrap()
1146 )
1147 }
1148
1149 pub fn environmentd_generation_service_name(&self, generation: u64) -> String {
1150 self.name_prefixed(&format!("environmentd-{generation}"))
1151 }
1152
1153 pub fn balancerd_app_name(&self) -> String {
1154 "balancerd".to_owned()
1155 }
1156
1157 pub fn environmentd_certificate_name(&self) -> String {
1158 self.name_prefixed("environmentd-external")
1159 }
1160
1161 pub fn environmentd_certificate_secret_name(&self) -> String {
1162 self.name_prefixed("environmentd-tls")
1163 }
1164
1165 pub fn balancerd_deployment_name(&self) -> String {
1166 self.name_prefixed("balancerd")
1167 }
1168
1169 pub fn balancerd_service_name(&self) -> String {
1170 self.name_prefixed("balancerd")
1171 }
1172
1173 pub fn console_app_name(&self) -> String {
1174 "console".to_owned()
1175 }
1176
1177 pub fn balancerd_external_certificate_name(&self) -> String {
1178 self.name_prefixed("balancerd-external")
1179 }
1180
1181 pub fn balancerd_external_certificate_secret_name(&self) -> String {
1182 self.name_prefixed("balancerd-external-tls")
1183 }
1184
1185 pub fn balancerd_replicas(&self) -> i32 {
1186 self.spec.balancerd_replicas.unwrap_or(2)
1187 }
1188
1189 pub fn console_replicas(&self) -> i32 {
1190 self.spec.console_replicas.unwrap_or(2)
1191 }
1192
1193 pub fn console_configmap_name(&self) -> String {
1194 self.name_prefixed("console")
1195 }
1196
1197 pub fn console_deployment_name(&self) -> String {
1198 self.name_prefixed("console")
1199 }
1200
1201 pub fn console_service_name(&self) -> String {
1202 self.name_prefixed("console")
1203 }
1204
1205 pub fn console_external_certificate_name(&self) -> String {
1206 self.name_prefixed("console-external")
1207 }
1208
1209 pub fn console_external_certificate_secret_name(&self) -> String {
1210 self.name_prefixed("console-external-tls")
1211 }
1212
1213 pub fn persist_pubsub_service_name(&self, generation: u64) -> String {
1214 self.name_prefixed(&format!("persist-pubsub-{generation}"))
1215 }
1216
1217 pub fn listeners_configmap_name(&self, generation: u64) -> String {
1218 self.name_prefixed(&format!("listeners-{generation}"))
1219 }
1220
1221 pub fn name_prefixed(&self, suffix: &str) -> String {
1222 format!("mz{}-{}", self.resource_id(), suffix)
1223 }
1224
1225 pub fn resource_id(&self) -> &str {
1226 &self.status.as_ref().unwrap().resource_id
1227 }
1228
1229 pub fn system_parameter_configmap_name(&self) -> Option<String> {
1230 self.spec.system_parameter_configmap_name.clone()
1231 }
1232
1233 pub fn environmentd_scratch_volume_storage_requirement(&self) -> Quantity {
1234 self.spec
1235 .environmentd_scratch_volume_storage_requirement
1236 .clone()
1237 .unwrap_or_else(|| {
1238 self.spec
1239 .environmentd_resource_requirements
1240 .as_ref()
1241 .and_then(|requirements| {
1242 requirements
1243 .requests
1244 .as_ref()
1245 .or(requirements.limits.as_ref())
1246 })
1247 .and_then(|requirements| requirements.get("memory").cloned())
1252 .unwrap_or_else(|| Quantity("4096Mi".to_string()))
1254 })
1255 }
1256
1257 pub fn environment_id(&self, cloud_provider: &str, region: &str) -> String {
1258 format!(
1259 "{}-{}-{}-0",
1260 cloud_provider, region, self.spec.environment_id,
1261 )
1262 }
1263
1264 pub fn rollout_requested(&self) -> bool {
1265 self.status
1266 .as_ref()
1267 .map(|status| status.last_completed_rollout_hash != status.requested_rollout_hash)
1268 .unwrap_or(false)
1269 }
1270
1271 pub fn set_force_promote(&mut self) {
1272 self.spec.force_promote = Some(self.generate_rollout_hash());
1273 }
1274
1275 pub fn should_force_promote(&self) -> bool {
1276 self.spec.force_promote.as_ref()
1277 == self
1278 .status
1279 .as_ref()
1280 .and_then(|status| status.requested_rollout_hash.as_ref())
1281 || self.spec.rollout_strategy
1282 == MaterializeRolloutStrategy::ImmediatelyPromoteCausingDowntime
1283 }
1284
1285 pub fn conditions_need_update(&self) -> bool {
1286 let Some(status) = self.status.as_ref() else {
1287 return true;
1288 };
1289 if status.conditions.is_empty() {
1290 return true;
1291 }
1292 for condition in &status.conditions {
1293 if condition.observed_generation != self.meta().generation {
1294 return true;
1295 }
1296 }
1297 false
1298 }
1299
1300 pub fn is_ready_to_promote(&self, rollout_hash: &str) -> bool {
1301 let Some(status) = self.status.as_ref() else {
1302 return false;
1303 };
1304 if status.conditions.is_empty() {
1305 return false;
1306 }
1307 status
1308 .conditions
1309 .iter()
1310 .any(|condition| condition.reason == "ReadyToPromote")
1311 && status.requested_rollout_hash.as_deref() == Some(rollout_hash)
1312 }
1313
1314 pub fn is_promoting(&self) -> bool {
1315 let Some(status) = self.status.as_ref() else {
1316 return false;
1317 };
1318 if status.conditions.is_empty() {
1319 return false;
1320 }
1321 status
1322 .conditions
1323 .iter()
1324 .any(|condition| condition.reason == "Promoting")
1325 }
1326
1327 pub fn update_in_progress(&self) -> bool {
1328 let Some(status) = self.status.as_ref() else {
1329 return false;
1330 };
1331 if status.conditions.is_empty() {
1332 return false;
1333 }
1334 for condition in &status.conditions {
1335 if condition.type_ == "UpToDate" && condition.status == "Unknown" {
1336 return true;
1337 }
1338 }
1339 false
1340 }
1341
1342 pub fn meets_minimum_version(&self, minimum: &Version) -> bool {
1346 let version = parse_image_ref(&self.spec.environmentd_image_ref);
1347 match version {
1348 Some(version) => version.cmp_precedence(minimum).is_ge(),
1350 None => {
1356 tracing::warn!(
1357 image_ref = %self.spec.environmentd_image_ref,
1358 "failed to parse image ref",
1359 );
1360 true
1361 }
1362 }
1363 }
1364
1365 pub fn is_valid_upgrade_version(active_version: &Version, next_version: &Version) -> bool {
1369 if next_version.cmp_precedence(active_version) == std::cmp::Ordering::Less {
1374 return false;
1375 }
1376
1377 if active_version.major == 0 {
1378 if next_version.major != active_version.major {
1379 if next_version.major == 26 {
1380 return (active_version.minor == 147 && active_version.patch >= 20)
1384 || active_version.minor >= 164;
1385 } else {
1386 return false;
1387 }
1388 }
1389 if next_version.minor == 147 && active_version.minor == 130 {
1391 return true;
1392 }
1393 return next_version.minor <= active_version.minor + 1;
1395 } else if active_version.major >= 26 {
1396 return next_version.major <= active_version.major + 1;
1398 }
1399
1400 true
1401 }
1402
1403 pub fn within_upgrade_window(&self) -> bool {
1406 let active_environmentd_version = self
1407 .status
1408 .as_ref()
1409 .and_then(|status| {
1410 status
1411 .last_completed_rollout_environmentd_image_ref
1412 .as_ref()
1413 })
1414 .and_then(|image_ref| parse_image_ref(image_ref));
1415
1416 if let (Some(next_environmentd_version), Some(active_environmentd_version)) = (
1417 parse_image_ref(&self.spec.environmentd_image_ref),
1418 active_environmentd_version,
1419 ) {
1420 Self::is_valid_upgrade_version(
1421 &active_environmentd_version,
1422 &next_environmentd_version,
1423 )
1424 } else {
1425 true
1428 }
1429 }
1430
1431 pub fn status(&self) -> MaterializeStatus {
1432 self.status.clone().unwrap_or_else(|| {
1433 let mut status = MaterializeStatus::default();
1434
1435 status.resource_id = new_resource_id();
1436
1437 if let Some(last_active_generation) = self
1442 .annotations()
1443 .get(LAST_KNOWN_ACTIVE_GENERATION_ANNOTATION)
1444 {
1445 status.active_generation = last_active_generation
1446 .parse()
1447 .expect("valid int generation");
1448 }
1449
1450 status.last_completed_rollout_environmentd_image_ref =
1453 Some(self.spec.environmentd_image_ref.clone());
1454
1455 status
1456 })
1457 }
1458 }
1459
1460 #[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema, PartialEq)]
1461 #[serde(rename_all = "camelCase")]
1462 pub struct MaterializeStatus {
1463 pub resource_id: String,
1465 pub active_generation: u64,
1467 pub last_completed_rollout_environmentd_image_ref: Option<String>,
1471 pub last_completed_rollout_hash: Option<String>,
1473 pub requested_rollout_hash: Option<String>,
1476 pub conditions: Vec<Condition>,
1477 }
1478
1479 impl MaterializeStatus {
1480 pub fn needs_update(&self, other: &Self) -> bool {
1481 let now = Timestamp::now();
1482 let mut a = self.clone();
1483 for condition in &mut a.conditions {
1484 condition.last_transition_time = Time(now);
1485 }
1486 let mut b = other.clone();
1487 for condition in &mut b.conditions {
1488 condition.last_transition_time = Time(now);
1489 }
1490 a != b
1491 }
1492 }
1493
1494 impl ManagedResource for Materialize {
1495 fn default_labels(&self) -> BTreeMap<String, String> {
1496 BTreeMap::from_iter([
1497 (
1498 "materialize.cloud/organization-name".to_owned(),
1499 self.name_unchecked(),
1500 ),
1501 (
1502 "materialize.cloud/organization-namespace".to_owned(),
1503 self.namespace(),
1504 ),
1505 (
1506 "materialize.cloud/mz-resource-id".to_owned(),
1507 self.resource_id().to_owned(),
1508 ),
1509 ])
1510 }
1511
1512 fn app_name(&self) -> Option<&str> {
1513 Some("environmentd")
1514 }
1515 }
1516
1517 impl From<v1alpha1::Materialize> for Materialize {
1518 fn from(value: v1alpha1::Materialize) -> Self {
1519 let is_promoting = value.is_promoting();
1520 let service_account_annotations = if let Some(environmentd_iam_role_arn) =
1521 value.spec.environmentd_iam_role_arn
1522 {
1523 let mut annotations = value.spec.service_account_annotations.unwrap_or_default();
1524 annotations
1525 .entry("eks.amazonaws.com/role-arn".to_owned())
1526 .or_insert(environmentd_iam_role_arn);
1527 Some(annotations)
1528 } else {
1529 value.spec.service_account_annotations
1530 };
1531 let mut mz = Materialize {
1532 metadata: value.metadata,
1533 spec: MaterializeSpec {
1534 environmentd_image_ref: value.spec.environmentd_image_ref,
1535 environmentd_extra_args: value.spec.environmentd_extra_args,
1536 environmentd_extra_env: value.spec.environmentd_extra_env,
1537 environmentd_connection_role_arn: value.spec.environmentd_connection_role_arn,
1538 environmentd_resource_requirements: value
1539 .spec
1540 .environmentd_resource_requirements,
1541 environmentd_scratch_volume_storage_requirement: value
1542 .spec
1543 .environmentd_scratch_volume_storage_requirement,
1544 balancerd_resource_requirements: value.spec.balancerd_resource_requirements,
1545 console_resource_requirements: value.spec.console_resource_requirements,
1546 balancerd_replicas: value.spec.balancerd_replicas,
1547 console_replicas: value.spec.console_replicas,
1548 service_account_name: value.spec.service_account_name,
1549 service_account_annotations,
1550 service_account_labels: value.spec.service_account_labels,
1551 pod_annotations: value.spec.pod_annotations,
1552 pod_labels: value.spec.pod_labels,
1553 force_promote: if value.spec.force_promote.is_empty()
1554 || &value.spec.force_promote == "00000000-0000-0000-0000-000000000000"
1555 {
1556 None
1557 } else {
1558 Some(value.spec.force_promote.to_string())
1559 },
1560 force_rollout: value.spec.force_rollout,
1561 rollout_strategy: value.spec.rollout_strategy,
1562 rollout_request_timeout: value.spec.rollout_request_timeout,
1563 backend_secret_name: value.spec.backend_secret_name,
1564 authenticator_kind: value.spec.authenticator_kind,
1565 enable_rbac: value.spec.enable_rbac,
1566 environment_id: value.spec.environment_id,
1567 system_parameter_configmap_name: value.spec.system_parameter_configmap_name,
1568 balancerd_external_certificate_spec: value
1569 .spec
1570 .balancerd_external_certificate_spec,
1571 console_external_certificate_spec: value.spec.console_external_certificate_spec,
1572 internal_certificate_spec: value.spec.internal_certificate_spec,
1573 },
1574 status: None,
1575 };
1576 let calculated_rollout_hash = mz.generate_rollout_hash();
1577 let last_completed_rollout_hash = match value
1578 .status
1579 .as_ref()
1580 .and_then(|status| status.last_completed_rollout_hash.to_owned())
1581 {
1582 Some(last_completed_rollout_hash) => Some(last_completed_rollout_hash),
1583 None => {
1584 let currently_rolling_out = value
1585 .status
1586 .as_ref()
1587 .map(|status| {
1588 status.last_completed_rollout_request != value.spec.request_rollout
1589 || status.last_completed_rollout_request.is_nil()
1592 })
1593 .unwrap_or(true);
1594 if currently_rolling_out {
1595 None
1597 } else {
1598 Some(calculated_rollout_hash.clone())
1599 }
1600 }
1601 };
1602 let requested_rollout_hash = if is_promoting {
1603 None
1604 } else {
1605 Some(calculated_rollout_hash)
1606 };
1607 mz.status = value.status.map(|status| MaterializeStatus {
1608 resource_id: status.resource_id,
1609 active_generation: status.active_generation,
1610 last_completed_rollout_environmentd_image_ref: status
1611 .last_completed_rollout_environmentd_image_ref,
1612 last_completed_rollout_hash,
1613 requested_rollout_hash,
1614 conditions: status.conditions,
1615 });
1616 mz
1617 }
1618 }
1619}
1620
1621fn parse_image_ref(image_ref: &str) -> Option<Version> {
1622 image_ref
1623 .rsplit_once(':')
1624 .and_then(|(_repo, tag)| tag.strip_prefix('v'))
1625 .and_then(|tag| {
1626 let tag = tag.replace("--", "+");
1631 Version::parse(&tag).ok()
1632 })
1633}
1634
1635#[cfg(test)]
1636mod tests {
1637 use std::time::Duration;
1638
1639 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
1640 use k8s_openapi::jiff::Timestamp;
1641 use kube::core::ObjectMeta;
1642 use semver::Version;
1643
1644 use super::v1alpha1::{Materialize, MaterializeSpec, MaterializeStatus};
1645 use super::{DEFAULT_ROLLOUT_REQUEST_TIMEOUT, RolloutRequestTimeout};
1646
1647 #[mz_ore::test]
1648 fn meets_minimum_version() {
1649 let mut mz = Materialize {
1650 spec: MaterializeSpec {
1651 environmentd_image_ref:
1652 "materialize/environmentd:devel-47116c24b8d0df33d3f60a9ee476aa8d7bce5953"
1653 .to_owned(),
1654 ..Default::default()
1655 },
1656 metadata: ObjectMeta {
1657 ..Default::default()
1658 },
1659 status: None,
1660 };
1661
1662 assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1664 mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0".to_owned();
1665 assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1666 mz.spec.environmentd_image_ref = "materialize/environmentd:v0.35.0".to_owned();
1667 assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1668 mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.3".to_owned();
1669 assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1670 mz.spec.environmentd_image_ref = "materialize/environmentd@41af286dc0b172ed2f1ca934fd2278de4a1192302ffa07087cea2682e7d372e3".to_owned();
1671 assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1672 mz.spec.environmentd_image_ref = "my.private.registry:5000:v0.34.3".to_owned();
1673 assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1674 mz.spec.environmentd_image_ref = "materialize/environmentd:v0.asdf.0".to_owned();
1675 assert!(mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1676 mz.spec.environmentd_image_ref =
1677 "materialize/environmentd:v0.146.0-dev.0--pr.g5a05a9e4ba873be8adaa528644aaae6e4c7cd29b"
1678 .to_owned();
1679 assert!(mz.meets_minimum_version(&Version::parse("0.146.0-dev.0").unwrap()));
1680
1681 mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0-dev".to_owned();
1683 assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1684 mz.spec.environmentd_image_ref = "materialize/environmentd:v0.33.0".to_owned();
1685 assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1686 mz.spec.environmentd_image_ref = "materialize/environmentd:v0.34.0".to_owned();
1687 assert!(!mz.meets_minimum_version(&Version::parse("1.0.0").unwrap()));
1688 mz.spec.environmentd_image_ref = "my.private.registry:5000:v0.33.3".to_owned();
1689 assert!(!mz.meets_minimum_version(&Version::parse("0.34.0").unwrap()));
1690 }
1691
1692 #[mz_ore::test]
1693 fn within_upgrade_window() {
1694 let mut mz = Materialize {
1695 spec: MaterializeSpec {
1696 environmentd_image_ref: "materialize/environmentd:v26.0.0".to_owned(),
1697 ..Default::default()
1698 },
1699 metadata: ObjectMeta {
1700 ..Default::default()
1701 },
1702 status: Some(MaterializeStatus {
1703 last_completed_rollout_environmentd_image_ref: Some(
1704 "materialize/environmentd:v26.0.0".to_owned(),
1705 ),
1706 ..Default::default()
1707 }),
1708 };
1709
1710 mz.spec.environmentd_image_ref = "materialize/environmentd:v27.7.3".to_owned();
1712 assert!(mz.within_upgrade_window());
1713
1714 mz.spec.environmentd_image_ref = "materialize/environmentd:v27.7.8-dev.0".to_owned();
1716 assert!(mz.within_upgrade_window());
1717
1718 mz.spec.environmentd_image_ref = "materialize/environmentd:v28.0.1".to_owned();
1720 assert!(!mz.within_upgrade_window());
1721
1722 mz.spec.environmentd_image_ref =
1724 "materialize/environmentd:v28.0.1.not_a_valid_version".to_owned();
1725 assert!(mz.within_upgrade_window());
1726
1727 mz.status
1729 .as_mut()
1730 .unwrap()
1731 .last_completed_rollout_environmentd_image_ref =
1732 Some("materialize/environmentd:v0.147.20".to_owned());
1733 mz.spec.environmentd_image_ref = "materialize/environmentd:v26.1.0".to_owned();
1734 assert!(mz.within_upgrade_window());
1735
1736 mz.status
1738 .as_mut()
1739 .unwrap()
1740 .last_completed_rollout_environmentd_image_ref =
1741 Some("materialize/environmentd:v26.11.0-dev.0+b".to_owned());
1742 mz.spec.environmentd_image_ref = "materialize/environmentd:v26.11.0-dev.0+a".to_owned();
1743 assert!(mz.within_upgrade_window());
1744 }
1745
1746 #[mz_ore::test]
1747 fn is_valid_upgrade_version() {
1748 let success_tests = [
1749 (Version::new(0, 83, 0), Version::new(0, 83, 0)),
1750 (Version::new(0, 83, 0), Version::new(0, 84, 0)),
1751 (Version::new(0, 9, 0), Version::new(0, 10, 0)),
1752 (Version::new(0, 99, 0), Version::new(0, 100, 0)),
1753 (Version::new(0, 83, 0), Version::new(0, 83, 1)),
1754 (Version::new(0, 83, 0), Version::new(0, 83, 2)),
1755 (Version::new(0, 83, 2), Version::new(0, 83, 10)),
1756 (Version::new(0, 147, 20), Version::new(26, 0, 0)),
1758 (Version::new(0, 164, 0), Version::new(26, 0, 0)),
1759 (Version::new(26, 0, 0), Version::new(26, 1, 0)),
1760 (Version::new(26, 5, 3), Version::new(26, 10, 0)),
1761 (Version::new(0, 130, 0), Version::new(0, 147, 0)),
1762 ];
1763 for (active_version, next_version) in success_tests {
1764 assert!(
1765 Materialize::is_valid_upgrade_version(&active_version, &next_version),
1766 "v{active_version} can upgrade to v{next_version}"
1767 );
1768 }
1769
1770 let failure_tests = [
1771 (Version::new(0, 83, 0), Version::new(0, 82, 0)),
1772 (Version::new(0, 83, 3), Version::new(0, 83, 2)),
1773 (Version::new(0, 83, 3), Version::new(1, 83, 3)),
1774 (Version::new(0, 83, 0), Version::new(0, 85, 0)),
1775 (Version::new(26, 0, 0), Version::new(28, 0, 0)),
1776 (Version::new(0, 130, 0), Version::new(26, 1, 0)),
1777 (Version::new(0, 147, 1), Version::new(26, 0, 0)),
1779 (Version::new(0, 148, 0), Version::new(26, 0, 0)),
1781 ];
1782 for (active_version, next_version) in failure_tests {
1783 assert!(
1784 !Materialize::is_valid_upgrade_version(&active_version, &next_version),
1785 "v{active_version} can't upgrade to v{next_version}"
1786 );
1787 }
1788 }
1789
1790 #[mz_ore::test]
1791 fn rollout_request_timeout() {
1792 let mz_with = |timeout: &str| Materialize {
1793 spec: MaterializeSpec {
1794 rollout_request_timeout: RolloutRequestTimeout(timeout.to_owned()),
1795 ..Default::default()
1796 },
1797 metadata: ObjectMeta::default(),
1798 status: None,
1799 };
1800
1801 let default = humantime::parse_duration(DEFAULT_ROLLOUT_REQUEST_TIMEOUT).unwrap();
1803 assert_eq!(default, Duration::from_secs(24 * 60 * 60));
1804
1805 assert_eq!(
1808 RolloutRequestTimeout::default().0,
1809 DEFAULT_ROLLOUT_REQUEST_TIMEOUT
1810 );
1811 assert_eq!(
1812 Materialize {
1813 spec: MaterializeSpec::default(),
1814 metadata: ObjectMeta::default(),
1815 status: None,
1816 }
1817 .rollout_request_timeout(),
1818 default
1819 );
1820
1821 assert_eq!(
1823 mz_with("1h").rollout_request_timeout(),
1824 Duration::from_secs(60 * 60)
1825 );
1826 assert_eq!(
1827 mz_with("90m").rollout_request_timeout(),
1828 Duration::from_secs(90 * 60)
1829 );
1830 assert_eq!(
1831 mz_with("1h 30m").rollout_request_timeout(),
1832 Duration::from_secs(90 * 60)
1833 );
1834 assert_eq!(mz_with("not a duration").rollout_request_timeout(), default);
1836 }
1837
1838 #[mz_ore::test]
1839 fn rollout_request_timeout_schema_default() {
1840 let crd = serde_json::to_value(<Materialize as kube::CustomResourceExt>::crd())
1844 .expect("CRD serializes");
1845 let default = &crd["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"]
1846 ["properties"]["rolloutRequestTimeout"]["default"];
1847 assert_eq!(
1848 default,
1849 &serde_json::json!(DEFAULT_ROLLOUT_REQUEST_TIMEOUT),
1850 "rolloutRequestTimeout schema default missing/wrong in generated CRD",
1851 );
1852 }
1853
1854 #[mz_ore::test]
1855 fn rollout_in_progress_since() {
1856 let now = Timestamp::now();
1857 let condition = |type_: &str, status: &str| Condition {
1858 type_: type_.to_owned(),
1859 status: status.to_owned(),
1860 last_transition_time: Time(now),
1861 message: String::new(),
1862 observed_generation: None,
1863 reason: "Test".to_owned(),
1864 };
1865 let mz_with = |conditions: Vec<Condition>| Materialize {
1866 spec: MaterializeSpec::default(),
1867 metadata: ObjectMeta::default(),
1868 status: Some(MaterializeStatus {
1869 conditions,
1870 ..Default::default()
1871 }),
1872 };
1873
1874 let mz = Materialize {
1876 spec: MaterializeSpec::default(),
1877 metadata: ObjectMeta::default(),
1878 status: None,
1879 };
1880 assert_eq!(mz.rollout_in_progress_since(), None);
1881
1882 assert_eq!(
1885 mz_with(vec![condition("UpToDate", "Unknown")]).rollout_in_progress_since(),
1886 Some(now)
1887 );
1888
1889 assert_eq!(
1893 mz_with(vec![Condition {
1894 reason: "Promoting".to_owned(),
1895 ..condition("UpToDate", "Unknown")
1896 }])
1897 .rollout_in_progress_since(),
1898 None
1899 );
1900
1901 assert_eq!(
1903 mz_with(vec![condition("UpToDate", "True")]).rollout_in_progress_since(),
1904 None
1905 );
1906 assert_eq!(
1907 mz_with(vec![condition("UpToDate", "False")]).rollout_in_progress_since(),
1908 None
1909 );
1910 }
1911
1912 #[mz_ore::test]
1913 fn up_to_date_transition_time() {
1914 let stored = Timestamp::from_second(1_000).unwrap();
1917 let now = Timestamp::from_second(2_000).unwrap();
1918
1919 let condition = |status: &str| Condition {
1920 type_: "UpToDate".to_owned(),
1921 status: status.to_owned(),
1922 last_transition_time: Time(stored),
1923 message: String::new(),
1924 observed_generation: None,
1925 reason: "Test".to_owned(),
1926 };
1927 let mz_with = |conditions: Vec<Condition>| Materialize {
1928 spec: MaterializeSpec::default(),
1929 metadata: ObjectMeta::default(),
1930 status: Some(MaterializeStatus {
1931 conditions,
1932 ..Default::default()
1933 }),
1934 };
1935
1936 let mz = Materialize {
1938 spec: MaterializeSpec::default(),
1939 metadata: ObjectMeta::default(),
1940 status: None,
1941 };
1942 assert_eq!(mz.up_to_date_transition_time("Unknown", now), now);
1943
1944 assert_eq!(
1948 mz_with(vec![condition("Unknown")]).up_to_date_transition_time("Unknown", now),
1949 stored
1950 );
1951
1952 assert_eq!(
1954 mz_with(vec![condition("Unknown")]).up_to_date_transition_time("True", now),
1955 now
1956 );
1957 }
1958
1959 #[mz_ore::test]
1960 fn active_environmentd_image_ref() {
1961 const OLD: &str = "materialize/environmentd:v26.0.0";
1962 const NEW: &str = "materialize/environmentd:v27.0.0";
1963
1964 let mz_with = |spec_image: &str, status: Option<MaterializeStatus>| Materialize {
1965 spec: MaterializeSpec {
1966 environmentd_image_ref: spec_image.to_owned(),
1967 ..Default::default()
1968 },
1969 metadata: ObjectMeta::default(),
1970 status,
1971 };
1972
1973 let mz = mz_with(NEW, None);
1975 assert_eq!(mz.active_environmentd_image_ref(), NEW);
1976
1977 let mz = mz_with(
1981 NEW,
1982 Some(MaterializeStatus {
1983 last_completed_rollout_environmentd_image_ref: None,
1984 ..Default::default()
1985 }),
1986 );
1987 assert_eq!(mz.active_environmentd_image_ref(), NEW);
1988
1989 let mz = mz_with(
1992 NEW,
1993 Some(MaterializeStatus {
1994 last_completed_rollout_environmentd_image_ref: Some(NEW.to_owned()),
1995 ..Default::default()
1996 }),
1997 );
1998 assert_eq!(mz.active_environmentd_image_ref(), NEW);
1999
2000 let mz = mz_with(
2009 NEW,
2010 Some(MaterializeStatus {
2011 last_completed_rollout_environmentd_image_ref: Some(OLD.to_owned()),
2012 ..Default::default()
2013 }),
2014 );
2015 assert_eq!(mz.active_environmentd_image_ref(), OLD);
2016 }
2017}