feat: Add simple CR applier

This implements are very simple CR applier. This commit is a first step
towards maintaining CRs in our operator. It is not refined and is missing
various features to be usable in the real world.

Author: Techassi

Cargo.lock

@@ -23,8 +23,13 @@ const_format = "0.2.34"
 futures = { version = "0.3", features = ["compat"] }
 h2 = "0.4"
 hex = "0.4"
-kube-runtime = { version = "1.0", features = ["unstable-runtime-stream-control"] }
-ldap3 = { version = "0.11", default-features = false, features = ["gssapi", "tls"] }
+kube-runtime = { version = "2.0.1", features = [
+  "unstable-runtime-stream-control",
+] }
+ldap3 = { version = "0.11", default-features = false, features = [
+  "gssapi",
+  "tls",
+] }
 libc = "0.2"
 native-tls = "0.2"
 openssl = "0.10"
@@ -54,5 +59,7 @@ uuid = { version = "1.10.0", features = ["v4"] }
 yasna = "0.5"
 
 [patch."https://github.com/stackabletech/operator-rs.git"]
+stackable-operator = { path = "../operator-rs/crates/stackable-operator" }
+# stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "feat/crd-maintenance" }
 # stackable-operator = { path = "../operator-rs/crates/stackable-operator" }
 # stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }

Cargo.nix

@@ -28,6 +28,6 @@ k8s_yaml(helm(
    namespace="stackable-operators",
    set=[
       'image.repository=' + registry + '/' + operator_name,
-      'telemetry.consoleLog.level=trace,h2=off',
+      'telemetry.consoleLog.level=info',
    ],
 ))

Cargo.toml

@@ -122,6 +122,8 @@ rules:
       - secretclasses
       - truststores
     verbs:
+      - create
+      - patch
       - get
       - watch
       - list

Tiltfile

@@ -1,7 +1,7 @@
 use stackable_operator::{
     k8s_openapi::api::core::v1::{ConfigMap, Secret},
     kube::api::PartialObjectMeta,
-    schemars::{self, schema::Schema},
+    schemars::{Schema, SchemaGenerator},
 };
 
 use crate::crd::secret_class::v1alpha1::{
@@ -113,7 +113,7 @@ impl CertificateKeyGeneration {
     //             - '3072'
     //             - '4096'
     //           type: string
-    pub fn tls_key_length_schema(_: &mut schemars::gen::SchemaGenerator) -> Schema {
+    pub fn tls_key_length_schema(_: &mut SchemaGenerator) -> Schema {
         serde_json::from_value(serde_json::json!({
             "type": "integer",
             "enum": [

deploy/helm/secret-operator/templates/roles.yaml

@@ -1,7 +1,7 @@
 use stackable_operator::{
     k8s_openapi::api::core::v1::{ConfigMap, Secret},
     kube::api::PartialObjectMeta,
-    schemars::{self, schema::Schema},
+    schemars::{Schema, SchemaGenerator},
     shared::time::Duration,
 };
 
@@ -131,7 +131,7 @@ impl CertificateKeyGeneration {
     //             - '3072'
     //             - '4096'
     //           type: string
-    pub fn tls_key_length_schema(_: &mut schemars::gen::SchemaGenerator) -> Schema {
+    pub fn tls_key_length_schema(_: &mut SchemaGenerator) -> Schema {
         serde_json::from_value(serde_json::json!({
             "type": "integer",
             "enum": [

rust/operator-binary/src/crd/secret_class/v1alpha1_impl.rs

@@ -16,17 +16,25 @@ use grpc::csi::v1::{
 };
 use stackable_operator::{
     YamlSchema,
-    cli::{CommonOptions, ProductOperatorRun},
+    cli::{CommonOptions, RunArguments},
+    client::Client,
+    crd::maintainer::{
+        CustomResourceDefinitionMaintainer, CustomResourceDefinitionMaintainerOptions,
+    },
     shared::yaml::SerializeOptions,
     telemetry::Tracing,
+    webhook::WebhookServer,
+};
+use tokio::{
+    signal::unix::{SignalKind, signal},
+    sync::oneshot,
 };
-use tokio::signal::unix::{SignalKind, signal};
 use tokio_stream::wrappers::UnixListenerStream;
 use tonic::transport::Server;
 use utils::{TonicUnixStream, uds_bind_private};
 use webhooks::conversion::conversion_webhook;
 
-use crate::crd::{SecretClass, TrustStore};
+use crate::crd::{SecretClass, SecretClassVersion, TrustStore, TrustStoreVersion, v1alpha2};
 
 mod backend;
 mod crd;
@@ -62,7 +70,7 @@ struct SecretOperatorRun {
     privileged: bool,
 
     #[clap(flatten)]
-    common: ProductOperatorRun,
+    common: RunArguments,
 }
 
 mod built_info {
@@ -83,7 +91,7 @@ async fn main() -> anyhow::Result<()> {
             csi_endpoint,
             privileged,
             common:
-                ProductOperatorRun {
+                RunArguments {
                     common:
                         CommonOptions {
                             telemetry,
@@ -92,7 +100,7 @@ async fn main() -> anyhow::Result<()> {
                     product_config: _,
                     watch_namespace,
                     operator_environment,
-                    disable_crd_maintenance,
+                    maintenance,
                 },
         }) => {
             // NOTE (@NickLarsenNZ): Before stackable-telemetry was used:
@@ -122,6 +130,27 @@ async fn main() -> anyhow::Result<()> {
             {
                 let _ = std::fs::remove_file(&csi_endpoint);
             }
+
+            let (conversion_webhook, certificate_rx) = conversion_webhook(&operator_environment)
+                .await
+                .context("failed to create conversion webhook")?;
+
+            let (maintainer, initial_reconcile_rx) = CustomResourceDefinitionMaintainer::new(
+                client.as_kube_client(),
+                certificate_rx,
+                [
+                    SecretClass::merged_crd(SecretClassVersion::V1Alpha2).unwrap(),
+                    TrustStore::merged_crd(TrustStoreVersion::V1Alpha1).unwrap(),
+                ],
+                CustomResourceDefinitionMaintainerOptions {
+                    operator_service_name: operator_environment.operator_service_name,
+                    operator_namespace: operator_environment.operator_namespace,
+                    field_manager: OPERATOR_NAME.to_owned(),
+                    webhook_https_port: WebhookServer::DEFAULT_HTTPS_PORT,
+                    disabled: maintenance.disable_crd_maintenance,
+                },
+            );
+
             let mut sigterm = signal(SignalKind::terminate())?;
             let csi_server = Server::builder()
                 .add_service(
@@ -151,19 +180,42 @@ async fn main() -> anyhow::Result<()> {
             let truststore_controller =
                 truststore_controller::start(&client, &watch_namespace).map(anyhow::Ok);
 
-            let conversion_webhook = conversion_webhook(
-                client.as_kube_client(),
-                operator_environment,
-                disable_crd_maintenance,
-            )
-            .await
-            .context("failed to create conversion webhook")?;
             let conversion_webhook = conversion_webhook
                 .run()
                 .map_err(|err| anyhow!(err).context("failed to run conversion webhook"));
 
-            try_join!(csi_server, truststore_controller, conversion_webhook,)?;
+            let maintainer = maintainer
+                .run()
+                .map_err(|err| anyhow!(err).context("failed to run CRD maintainer"));
+
+            let cr_applier = apply_crs(initial_reconcile_rx, client.clone())
+                .map_err(|err| anyhow!(err).context("failed to apply default custom resources"));
+
+            try_join!(
+                csi_server,
+                truststore_controller,
+                conversion_webhook,
+                maintainer,
+                cr_applier,
+            )?;
         }
     }
     Ok(())
 }
+
+async fn apply_crs(
+    initial_reconcile_rx: oneshot::Receiver<()>,
+    client: Client,
+) -> anyhow::Result<()> {
+    initial_reconcile_rx.await?;
+
+    tracing::info!("applying default custom resources");
+
+    let deserializer = serde_yaml::Deserializer::from_slice(include_bytes!("secretclass.yaml"));
+    let tls_secret_class: v1alpha2::SecretClass =
+        serde_yaml::with::singleton_map_recursive::deserialize(deserializer)?;
+
+    client.create_if_missing(&tls_secret_class).await?;
+
+    Ok(())
+}

rust/operator-binary/src/crd/secret_class/v1alpha2_impl.rs

@@ -0,0 +1,13 @@
+---
+apiVersion: secrets.stackable.tech/v1alpha2
+kind: SecretClass
+metadata:
+  name: tls
+spec:
+  backend:
+    autoTls:
+      ca:
+        secret:
+          name: secret-provisioner-tls-ca
+          namespace: stackable-operator
+        autoGenerate: true

rust/operator-binary/src/main.rs

@@ -1,22 +1,17 @@
 use stackable_operator::{
     cli::OperatorEnvironmentOptions,
-    kube::Client,
     webhook::{
-        constants::DEFAULT_SOCKET_ADDRESS,
         servers::{ConversionWebhookOptions, ConversionWebhookServer},
+        x509_cert::Certificate,
     },
 };
+use tokio::sync::mpsc;
 
-use crate::{
-    OPERATOR_NAME,
-    crd::{SecretClass, SecretClassVersion, TrustStore, TrustStoreVersion},
-};
+use crate::crd::{SecretClass, SecretClassVersion, TrustStore, TrustStoreVersion};
 
 pub async fn conversion_webhook(
-    client: Client,
-    operator_environment: OperatorEnvironmentOptions,
-    disable_crd_management: bool,
-) -> anyhow::Result<ConversionWebhookServer> {
+    operator_environment: &OperatorEnvironmentOptions,
+) -> anyhow::Result<(ConversionWebhookServer, mpsc::Receiver<Certificate>)> {
     let crds_and_handlers = [
         (
             SecretClass::merged_crd(SecretClassVersion::V1Alpha2)?,
@@ -29,12 +24,10 @@ pub async fn conversion_webhook(
     ];
 
     let options = ConversionWebhookOptions {
-        socket_addr: DEFAULT_SOCKET_ADDRESS,
-        field_manager: OPERATOR_NAME.to_owned(),
-        namespace: operator_environment.operator_namespace,
-        service_name: operator_environment.operator_service_name,
-        maintain_crds: !disable_crd_management,
+        socket_addr: ConversionWebhookServer::DEFAULT_SOCKET_ADDRESS,
+        namespace: operator_environment.operator_namespace.clone(),
+        service_name: operator_environment.operator_service_name.clone(),
     };
 
-    Ok(ConversionWebhookServer::new(crds_and_handlers, options, client).await?)
+    Ok(ConversionWebhookServer::new(crds_and_handlers, options).await?)
 }