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

mz_orchestratord/
k8s.rs
