mz_orchestratord/
k8s.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    ByteString,
15    api::{
16        apps::v1::Deployment,
17        policy::v1::{PodDisruptionBudget, PodDisruptionBudgetSpec},
18    },
19    apiextensions_apiserver::pkg::apis::apiextensions::v1::{
20        CustomResourceColumnDefinition, CustomResourceConversion, ServiceReference,
21        WebhookClientConfig, WebhookConversion,
22    },
23    apimachinery::pkg::util::intstr::IntOrString,
24};
25use kube::{
26    Api, Client, CustomResourceExt, Resource, ResourceExt,
27    api::{DeleteParams, ObjectMeta, Patch, PatchParams, PostParams},
28    core::Status,
29    core::response::StatusSummary,
30};
31use serde::{Serialize, de::DeserializeOwned};
32use tracing::info;
33
34use mz_cloud_resources::crd::{self, VersionedCrd, register_versioned_crds};
35
36pub(crate) const FIELD_MANAGER: &str = "orchestratord.materialize.cloud";
37
38pub async fn get_resource<K>(api: &Api<K>, name: &str) -> Result<Option<K>, anyhow::Error>
39where
40    K: Resource + Clone + Send + DeserializeOwned + Serialize + std::fmt::Debug + 'static,
41    <K as Resource>::DynamicType: Default,
42{
43    match api.get(name).await {
44        Ok(k) => Ok(Some(k)),
45        Err(kube::Error::Api(e)) if e.code == 404 => Ok(None),
46        Err(e) => Err(e.into()),
47    }
48}
49
50pub async fn apply_resource<K>(api: &Api<K>, resource: &K) -> Result<K, anyhow::Error>
51where
52    K: Resource + Clone + Send + DeserializeOwned + Serialize + std::fmt::Debug + 'static,
53    <K as Resource>::DynamicType: Default,
54{
55    Ok(api
56        .patch(
57            &resource.name_unchecked(),
58            &PatchParams::apply(FIELD_MANAGER).force(),
59            &Patch::Apply(resource),
60        )
61        .await?)
62}
63
64pub async fn replace_resource<K>(api: &Api<K>, resource: &K) -> Result<K, anyhow::Error>
65where
66    K: Resource + Clone + Send + DeserializeOwned + Serialize + std::fmt::Debug + 'static,
67    <K as Resource>::DynamicType: Default,
68{
69    if resource.meta().resource_version.is_none() {
70        return Err(kube::Error::Api(
71            Box::new(Status {
72                status: Some(StatusSummary::Failure),
73                message: "Must use apply_resource instead of replace_resource to apply fully created resources.".to_string(),
74                reason: "BadRequest".to_string(),
75                code: 400,
76                metadata: None,
77                details: None,
78            }),
79        )
80        .into());
81    }
82    Ok(api
83        .replace(&resource.name_unchecked(), &PostParams::default(), resource)
84        .await?)
85}
86
87pub async fn delete_resource<K>(api: &Api<K>, name: &str) -> Result<(), anyhow::Error>
88where
89    K: Resource + Clone + Send + DeserializeOwned + Serialize + std::fmt::Debug + 'static,
90    <K as Resource>::DynamicType: Default,
91{
92    match kube::runtime::wait::delete::delete_and_finalize(
93        api.clone(),
94        name,
95        &DeleteParams::foreground(),
96    )
97    .await
98    {
99        Ok(_) => Ok(()),
100        Err(kube::runtime::wait::delete::Error::Delete(kube::Error::Api(e))) if e.code == 404 => {
101            // the resource already doesn't exist
102            Ok(())
103        }
104        Err(e) => Err(e.into()),
105    }
106}
107
108/// A PodDisruptionBudget letting at most one of `deployment`'s pods be evicted
109/// at a time. With a single replica it never blocks a drain.
110pub fn pod_disruption_budget(metadata: ObjectMeta, deployment: &Deployment) -> PodDisruptionBudget {
111    PodDisruptionBudget {
112        metadata,
113        spec: Some(PodDisruptionBudgetSpec {
114            max_unavailable: Some(IntOrString::Int(1)),
115            selector: deployment.spec.as_ref().map(|spec| spec.selector.clone()),
116            // The default, `IfHealthyBudget`, refuses to evict unready pods
117            // once none are healthy, so a crashlooping Deployment would block
118            // node drains.
119            unhealthy_pod_eviction_policy: Some("AlwaysAllow".into()),
120            ..Default::default()
121        }),
122        status: None,
123    }
124}
125
126/// Configuration for the conversion webhook that serves the v1 version of the
127/// Materialize CRD. When present, the v1 version is registered alongside
128/// v1alpha1 with webhook conversion between them; when absent, only v1alpha1
129/// is registered.
130#[derive(Debug, Clone)]
131pub struct ConversionWebhookConfig {
132    pub service_name: String,
133    pub service_namespace: String,
134    pub service_port: u16,
135    pub ca_cert_path: String,
136}
137
138pub async fn register_crds(
139    client: Client,
140    additional_crd_columns: Vec<CustomResourceColumnDefinition>,
141    conversion_webhook: Option<ConversionWebhookConfig>,
142) -> Result<(), anyhow::Error> {
143    let (mut mz_crds, mz_conversion) = match conversion_webhook {
144        Some(config) => {
145            let ca_bytes = tokio::fs::read(config.ca_cert_path).await?;
146            let conversion = CustomResourceConversion {
147                strategy: "Webhook".to_owned(),
148                webhook: Some(WebhookConversion {
149                    client_config: Some(WebhookClientConfig {
150                        ca_bundle: Some(ByteString(ca_bytes)),
151                        service: Some(ServiceReference {
152                            name: config.service_name,
153                            namespace: config.service_namespace,
154                            path: Some("/convert".to_owned()),
155                            port: Some(config.service_port.into()),
156                        }),
157                        url: None,
158                    }),
159                    conversion_review_versions: vec!["v1".to_owned()],
160                }),
161            };
162            (
163                vec![
164                    crd::materialize::v1::Materialize::crd(),
165                    crd::materialize::v1alpha1::Materialize::crd(),
166                ],
167                Some(conversion),
168            )
169        }
170        None => (vec![crd::materialize::v1alpha1::Materialize::crd()], None),
171    };
172    let default_columns = mz_crds[0].spec.versions[0]
173        .additional_printer_columns
174        .take()
175        .expect("should contain ImageRef and UpToDate columns");
176    mz_crds[0].spec.versions[0].additional_printer_columns = Some(
177        additional_crd_columns
178            .into_iter()
179            .chain(default_columns)
180            .collect(),
181    );
182    tokio::time::timeout(
183        Duration::from_secs(120),
184        register_versioned_crds(
185            client.clone(),
186            vec![
187                VersionedCrd {
188                    crds: mz_crds,
189                    stored_version: String::from("v1alpha1"),
190                    conversion: mz_conversion,
191                },
192                VersionedCrd {
193                    crds: vec![crd::balancer::v1alpha1::Balancer::crd()],
194                    stored_version: String::from("v1alpha1"),
195                    conversion: None,
196                },
197                VersionedCrd {
198                    crds: vec![crd::console::v1alpha1::Console::crd()],
199                    stored_version: String::from("v1alpha1"),
200                    conversion: None,
201                },
202                VersionedCrd {
203                    crds: vec![crd::vpc_endpoint::v1::VpcEndpoint::crd()],
204                    stored_version: String::from("v1"),
205                    conversion: None,
206                },
207            ],
208            FIELD_MANAGER,
209        ),
210    )
211    .await??;
212
213    info!("Done rewriting CRDs");
214
215    Ok(())
216}
217
218/// Get the recommended Kubernetes labels (app.kubernetes.io/*)
219/// WARNING: this is duplicated in src/orchestrator/src/lib.rs and src/cloud-resources/src/crd.rs
220pub fn recommended_k8s_labels(app_name: String) -> BTreeMap<String, String> {
221    let mut labels = BTreeMap::new();
222    labels.insert(
223        "app.kubernetes.io/managed-by".into(),
224        "materialize-operator".into(),
225    );
226    labels.insert("app.kubernetes.io/part-of".into(), "materialize".into());
227    labels.insert("app.kubernetes.io/name".into(), app_name.to_owned());
228    // legacy label
229    labels.insert("app".into(), app_name.to_owned());
230    labels
231}
mz_orchestratord/k8s.rs

mz_orchestratord/
k8s.rs
