mz_cloud_resources/
crd.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
10//! Kubernetes custom resources
11
12use std::collections::BTreeMap;
13use std::time::Duration;
14
15use futures::future::join_all;
16use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::{
17    CustomResourceConversion, CustomResourceDefinition,
18};
19use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference;
20use kube::{
21    Api, Client, Resource, ResourceExt,
22    api::{ObjectMeta, Patch, PatchParams},
23    core::crd::merge_crds,
24    runtime::{conditions, wait::await_condition},
25};
26use rand::{Rng, distr::Uniform};
27use schemars::JsonSchema;
28use serde::{Deserialize, Serialize};
29use tracing::{info, warn};
30
31use crate::crd::generated::cert_manager::certificates::{
32    CertificateIssuerRef, CertificatePrivateKeyAlgorithm, CertificateSecretTemplate,
33};
34use mz_ore::retry::Retry;
35
36pub mod balancer;
37pub mod console;
38pub mod generated;
39pub mod materialize;
40#[cfg(feature = "vpc-endpoints")]
41pub mod vpc_endpoint;
42
43// This is intentionally a subset of the fields of a Certificate.
44// We do not want customers to configure options that may conflict with
45// things we override or expand in our code.
46#[derive(Clone, Debug, Default, PartialEq, Deserialize, Serialize, JsonSchema)]
47#[serde(rename_all = "camelCase")]
48pub struct MaterializeCertSpec {
49    /// Additional DNS names the certificate will be valid for.
50    pub dns_names: Option<Vec<String>>,
51    /// Duration the certificate will be requested for.
52    /// Value must be in units accepted by Go
53    /// [`time.ParseDuration`](https://golang.org/pkg/time/#ParseDuration).
54    pub duration: Option<String>,
55    /// Duration before expiration the certificate will be renewed.
56    /// Value must be in units accepted by Go
57    /// [`time.ParseDuration`](https://golang.org/pkg/time/#ParseDuration).
58    pub renew_before: Option<String>,
59    /// Reference to an `Issuer` or `ClusterIssuer` that will generate the certificate.
60    pub issuer_ref: Option<CertificateIssuerRef>,
61    /// Additional annotations and labels to include in the Certificate object.
62    pub secret_template: Option<CertificateSecretTemplate>,
63    /// Optional algorithm to use for the private key. If not specified, a recommended default will be chosen.
64    #[serde(skip_serializing_if = "Option::is_none")]
65    pub private_key_algorithm: Option<CertificatePrivateKeyAlgorithm>,
66    /// Optional size for the private key.
67    #[serde(skip_serializing_if = "Option::is_none")]
68    pub private_key_size: Option<i64>,
69}
70
71pub trait ManagedResource: Resource<DynamicType = ()> + Sized {
72    fn default_labels(&self) -> BTreeMap<String, String> {
73        BTreeMap::new()
74    }
75
76    fn app_name(&self) -> Option<&str> {
77        None
78    }
79
80    fn managed_resource_meta(&self, name: String) -> ObjectMeta {
81        let mut labels = self.default_labels();
82        labels.extend(recommended_k8s_labels(self.app_name()));
83        ObjectMeta {
84            namespace: Some(self.meta().namespace.clone().unwrap()),
85            name: Some(name),
86            labels: Some(labels),
87            owner_references: Some(vec![owner_reference(self)]),
88            ..Default::default()
89        }
90    }
91}
92
93/// Get the recommended Kubernetes labels (app.kubernetes.io/*)
94/// WARNING: this is duplicated in src/orchestrator/src/lib.rs and src/orchestratord/src/k8s.rs
95pub fn recommended_k8s_labels(app_name: Option<&str>) -> BTreeMap<String, String> {
96    let mut labels = BTreeMap::new();
97    labels.insert(
98        "app.kubernetes.io/managed-by".to_owned(),
99        "materialize-operator".to_owned(),
100    );
101    labels.insert(
102        "app.kubernetes.io/part-of".to_owned(),
103        "materialize".to_owned(),
104    );
105    if let Some(app) = app_name {
106        labels.insert("app.kubernetes.io/name".to_owned(), app.to_owned());
107        // legacy label
108        labels.insert("app".to_owned(), app.to_owned());
109    }
110    labels
111}
112
113fn owner_reference<T: Resource<DynamicType = ()>>(t: &T) -> OwnerReference {
114    OwnerReference {
115        api_version: T::api_version(&()).to_string(),
116        kind: T::kind(&()).to_string(),
117        name: t.name_unchecked(),
118        uid: t.uid().unwrap(),
119        block_owner_deletion: Some(true),
120        ..Default::default()
121    }
122}
123
124#[derive(Debug, Clone)]
125pub struct VersionedCrd {
126    pub crds: Vec<CustomResourceDefinition>,
127    pub stored_version: String,
128    /// Conversion configuration to apply after merging CRDs.
129    /// `merge_crds` drops the conversion field, so we must set it after merging.
130    pub conversion: Option<CustomResourceConversion>,
131}
132
133pub async fn register_versioned_crds(
134    kube_client: Client,
135    versioned_crds: Vec<VersionedCrd>,
136    field_manager: &str,
137) -> Result<(), anyhow::Error> {
138    let crd_futures = versioned_crds
139        .into_iter()
140        .map(|versioned_crd| register_w_retry(kube_client.clone(), versioned_crd, field_manager));
141    for res in join_all(crd_futures).await {
142        if res.is_err() {
143            return res;
144        }
145    }
146    Ok(())
147}
148
149async fn register_w_retry(
150    kube_client: Client,
151    versioned_crds: VersionedCrd,
152    field_manager: &str,
153) -> Result<(), anyhow::Error> {
154    Retry::default()
155        .max_duration(Duration::from_secs(30))
156        .clamp_backoff(Duration::from_secs(5))
157        .retry_async(|_| async {
158            let res = register_custom_resource(
159                kube_client.clone(),
160                versioned_crds.clone(),
161                field_manager,
162            )
163            .await;
164            if let Err(err) = &res {
165                warn!(err = %err);
166            }
167            res
168        })
169        .await?;
170    Ok(())
171}
172
173/// Registers a custom resource with Kubernetes,
174/// the specification of which is automatically derived from the structs.
175async fn register_custom_resource(
176    kube_client: Client,
177    versioned_crds: VersionedCrd,
178    field_manager: &str,
179) -> Result<(), anyhow::Error> {
180    let crds = versioned_crds.crds;
181    let crd_name = format!("{}.{}", &crds[0].spec.names.plural, &crds[0].spec.group);
182    info!("Registering {} crd", &crd_name);
183    let crd_api = Api::<CustomResourceDefinition>::all(kube_client);
184    let mut crd = merge_crds(crds, &versioned_crds.stored_version).unwrap();
185    if let Some(conversion) = versioned_crds.conversion {
186        crd.spec.conversion = Some(conversion);
187    }
188    let crd_json = serde_json::to_string(&serde_json::json!(&crd))?;
189    info!(crd_json = %crd_json);
190    crd_api
191        .patch(
192            &crd_name,
193            &PatchParams::apply(field_manager).force(),
194            &Patch::Apply(crd),
195        )
196        .await?;
197    await_condition(crd_api, &crd_name, conditions::is_crd_established()).await?;
198    info!("Done registering {} crd", &crd_name);
199    Ok(())
200}
201
202pub fn new_resource_id() -> String {
203    // DNS-1035 names are supposed to be case insensitive,
204    // so we define our own character set, rather than use the
205    // built-in Alphanumeric distribution from rand, which
206    // includes both upper and lowercase letters.
207    const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789";
208    rand::rng()
209        .sample_iter(Uniform::new(0, CHARSET.len()).expect("valid range"))
210        .take(10)
211        .map(|i| char::from(CHARSET[i]))
212        .collect()
213}
mz_cloud_resources/crd.rs

mz_cloud_resources/
crd.rs
