add console crd and operator

Author: doy-materialize

misc/helm-charts/operator/templates/clusterrole.yaml

@@ -90,6 +90,8 @@ rules:
   - materializes/status
   - balancers
   - balancers/status
+  - consoles
+  - consoles/status
   - vpcendpoints
   verbs:
   - create

src/cloud-resources/src/crd.rs

@@ -32,6 +32,7 @@ use crate::crd::generated::cert_manager::certificates::{
 use mz_ore::retry::Retry;
 
 pub mod balancer;
+pub mod console;
 pub mod generated;
 pub mod materialize;
 #[cfg(feature = "vpc-endpoints")]

src/cloud-resources/src/crd/console.rs

@@ -0,0 +1,145 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+use std::collections::BTreeMap;
+
+use k8s_openapi::{
+    api::core::v1::ResourceRequirements, apimachinery::pkg::apis::meta::v1::Condition,
+};
+use kube::CustomResource;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::crd::{ManagedResource, MaterializeCertSpec, new_resource_id};
+use mz_server_core::listeners::AuthenticatorKind;
+
+pub mod v1alpha1 {
+    use super::*;
+
+    #[derive(Clone, Debug, Default, PartialEq, Deserialize, Serialize, JsonSchema)]
+    #[serde(rename_all = "camelCase")]
+    pub struct BalancerdRef {
+        // The service name for the balancerd service to connect to
+        pub service_name: String,
+        // The namespace the balancerd service runs in
+        pub namespace: String,
+        // The configuration for generating an x509 certificate using cert-manager for balancerd
+        // to present to incoming connections.
+        // The dns_names and issuer_ref fields are required.
+        pub external_certificate_spec: Option<MaterializeCertSpec>,
+        /// How to authenticate with Materialize.
+        #[serde(default)]
+        pub authenticator_kind: AuthenticatorKind,
+    }
+
+    #[derive(
+        CustomResource, Clone, Debug, Default, PartialEq, Deserialize, Serialize, JsonSchema,
+    )]
+    #[serde(rename_all = "camelCase")]
+    #[kube(
+        namespaced,
+        group = "materialize.cloud",
+        version = "v1alpha1",
+        kind = "Console",
+        singular = "console",
+        plural = "consoles",
+        status = "ConsoleStatus",
+        printcolumn = r#"{"name": "ImageRef", "type": "string", "description": "Reference to the Docker image.", "jsonPath": ".spec.consoleImageRef", "priority": 1}"#,
+        printcolumn = r#"{"name": "Ready", "type": "string", "description": "Whether the deployment is ready", "jsonPath": ".status.conditions[?(@.type==\"Ready\")].status", "priority": 1}"#
+    )]
+    pub struct ConsoleSpec {
+        /// The console image to run.
+        pub console_image_ref: String,
+        // Resource requirements for the console pod
+        pub resource_requirements: Option<ResourceRequirements>,
+        // Number of console pods to create
+        pub replicas: Option<i32>,
+        // The configuration for generating an x509 certificate using cert-manager for console
+        // to present to incoming connections.
+        // The dns_names and issuer_ref fields are required.
+        pub external_certificate_spec: Option<MaterializeCertSpec>,
+        // Annotations to apply to the pods
+        pub pod_annotations: Option<BTreeMap<String, String>>,
+        // Labels to apply to the pods
+        pub pod_labels: Option<BTreeMap<String, String>>,
+
+        // Connection information for the balancerd service to use
+        pub balancerd: BalancerdRef,
+
+        // This can be set to override the randomly chosen resource id
+        pub resource_id: Option<String>,
+    }
+
+    impl Console {
+        pub fn name_prefixed(&self, suffix: &str) -> String {
+            format!("mz{}-{}", self.resource_id(), suffix)
+        }
+
+        pub fn resource_id(&self) -> &str {
+            &self.status.as_ref().unwrap().resource_id
+        }
+
+        pub fn deployment_name(&self) -> String {
+            self.name_prefixed("console")
+        }
+
+        pub fn replicas(&self) -> i32 {
+            self.spec.replicas.unwrap_or(2)
+        }
+
+        pub fn app_name(&self) -> String {
+            "console".to_owned()
+        }
+
+        pub fn service_name(&self) -> String {
+            self.name_prefixed("console")
+        }
+
+        pub fn configmap_name(&self) -> String {
+            self.name_prefixed("console")
+        }
+
+        pub fn external_certificate_name(&self) -> String {
+            self.name_prefixed("console-external")
+        }
+
+        pub fn external_certificate_secret_name(&self) -> String {
+            self.name_prefixed("console-external-tls")
+        }
+
+        pub fn status(&self) -> ConsoleStatus {
+            self.status.clone().unwrap_or_else(|| ConsoleStatus {
+                resource_id: self
+                    .spec
+                    .resource_id
+                    .clone()
+                    .unwrap_or_else(new_resource_id),
+                conditions: vec![],
+            })
+        }
+    }
+
+    #[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema, PartialEq)]
+    #[serde(rename_all = "camelCase")]
+    pub struct ConsoleStatus {
+        /// Resource identifier used as a name prefix to avoid pod name collisions.
+        pub resource_id: String,
+
+        pub conditions: Vec<Condition>,
+    }
+
+    impl ManagedResource for Console {
+        fn default_labels(&self) -> BTreeMap<String, String> {
+            BTreeMap::from_iter([(
+                "materialize.cloud/mz-resource-id".to_owned(),
+                self.resource_id().to_owned(),
+            )])
+        }
+    }
+}

src/orchestratord/src/bin/orchestratord.rs

@@ -17,7 +17,8 @@ use http::HeaderValue;
 use k8s_openapi::{
     api::{
         apps::v1::Deployment,
-        core::v1::{Affinity, ResourceRequirements, Service, Toleration},
+        core::v1::{Affinity, ConfigMap, ResourceRequirements, Service, Toleration},
+        networking::v1::NetworkPolicy,
     },
     apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceColumnDefinition,
 };
@@ -335,14 +336,14 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
                     clusterd_node_selector: args.clusterd_node_selector,
                     clusterd_affinity: args.clusterd_affinity,
                     clusterd_tolerations: args.clusterd_tolerations,
-                    console_node_selector: args.console_node_selector,
-                    console_affinity: args.console_affinity,
-                    console_tolerations: args.console_tolerations,
-                    console_default_resources: args.console_default_resources,
+                    console_node_selector: args.console_node_selector.clone(),
+                    console_affinity: args.console_affinity.clone(),
+                    console_tolerations: args.console_tolerations.clone(),
+                    console_default_resources: args.console_default_resources.clone(),
                     image_pull_policy: args.image_pull_policy,
                     network_policies_internal_enabled: args.network_policies_internal_enabled,
                     network_policies_ingress_enabled: args.network_policies_ingress_enabled,
-                    network_policies_ingress_cidrs: args.network_policies_ingress_cidrs,
+                    network_policies_ingress_cidrs: args.network_policies_ingress_cidrs.clone(),
                     network_policies_egress_enabled: args.network_policies_egress_enabled,
                     network_policies_egress_cidrs: args.network_policies_egress_cidrs,
                     environmentd_cluster_replica_sizes: args.environmentd_cluster_replica_sizes,
@@ -399,12 +400,12 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
                     enable_security_context: args.enable_security_context,
                     enable_prometheus_scrape_annotations: args.enable_prometheus_scrape_annotations,
                     image_pull_policy: args.image_pull_policy,
-                    scheduler_name: args.scheduler_name,
+                    scheduler_name: args.scheduler_name.clone(),
                     balancerd_node_selector: args.balancerd_node_selector,
                     balancerd_affinity: args.balancerd_affinity,
                     balancerd_tolerations: args.balancerd_tolerations,
                     balancerd_default_resources: args.balancerd_default_resources,
-                    default_certificate_specs: args.default_certificate_specs,
+                    default_certificate_specs: args.default_certificate_specs.clone(),
                     environmentd_sql_port: args.environmentd_sql_port,
                     environmentd_http_port: args.environmentd_http_port,
                     balancerd_sql_port: args.balancerd_sql_port,
@@ -440,6 +441,67 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
         .run(),
     );
 
+    mz_ore::task::spawn(
+        || "console controller",
+        k8s_controller::Controller::namespaced_all(
+            client.clone(),
+            controller::console::Context::new(
+                controller::console::Config {
+                    enable_security_context: args.enable_security_context,
+                    enable_prometheus_scrape_annotations: args.enable_prometheus_scrape_annotations,
+                    image_pull_policy: args.image_pull_policy,
+                    scheduler_name: args.scheduler_name,
+                    console_node_selector: args.console_node_selector,
+                    console_affinity: args.console_affinity,
+                    console_tolerations: args.console_tolerations,
+                    console_default_resources: args.console_default_resources,
+                    network_policies_ingress_enabled: args.network_policies_ingress_enabled,
+                    network_policies_ingress_cidrs: args.network_policies_ingress_cidrs,
+                    default_certificate_specs: args.default_certificate_specs,
+                    console_http_port: args.console_http_port,
+                    balancerd_http_port: args.balancerd_http_port,
+                },
+                client.clone(),
+            )
+            .await,
+            watcher::Config::default().timeout(29),
+        )
+        .with_controller(|controller| {
+            controller
+                .owns(
+                    Api::<Deployment>::all(client.clone()),
+                    watcher::Config::default()
+                        .labels("materialize.cloud/mz-resource-id")
+                        .timeout(29),
+                )
+                .owns(
+                    Api::<Service>::all(client.clone()),
+                    watcher::Config::default()
+                        .labels("materialize.cloud/mz-resource-id")
+                        .timeout(29),
+                )
+                .owns(
+                    Api::<Certificate>::all(client.clone()),
+                    watcher::Config::default()
+                        .labels("materialize.cloud/mz-resource-id")
+                        .timeout(29),
+                )
+                .owns(
+                    Api::<NetworkPolicy>::all(client.clone()),
+                    watcher::Config::default()
+                        .labels("materialize.cloud/mz-resource-id")
+                        .timeout(29),
+                )
+                .owns(
+                    Api::<ConfigMap>::all(client.clone()),
+                    watcher::Config::default()
+                        .labels("materialize.cloud/mz-resource-id")
+                        .timeout(29),
+                )
+        })
+        .run(),
+    );
+
     info!("All tasks started successfully.");
 
     future::pending().await

src/orchestratord/src/controller.rs

@@ -8,4 +8,5 @@
 // by the Apache License, Version 2.0.
 
 pub mod balancer;
+pub mod console;
 pub mod materialize;