fix: Prevent Pod 0 restart by utilizing mutating webhook

Cargo.toml

@@ -10,19 +10,21 @@ edition = "2021"
 repository = "https://github.com/stackabletech/commons-operator"
 
 [workspace.dependencies]
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", features = ["telemetry"], tag = "stackable-operator-0.100.1" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", features = ["telemetry", "webhook"], tag = "stackable-operator-0.100.1" }
 
 anyhow = "1.0"
 built = { version = "0.8", features = ["chrono", "git2"] }
 clap = "4.5"
 futures = { version = "0.3", features = ["compat"] }
 http = "1.3"
+json-patch = "4.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 snafu = "0.8"
 strum = { version = "0.27", features = ["derive"] }
 tokio = { version = "1.40", features = ["full"] }
 tracing = "0.1"
 
-# [patch."https://github.com/stackabletech/operator-rs.git"]
-# stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }
+[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/mutating-webhook" }

_TEST.yaml

@@ -0,0 +1,63 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: nginx
+  labels:
+    app: nginx
+spec:
+  ports:
+  - port: 80
+    name: web
+  clusterIP: None
+  selector:
+    app: nginx
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: web-config
+data:
+  foo: bar
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: web-config-2
+data:
+  foo: bar
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: web
+  labels:
+    restarter.stackable.tech/enabled: "true"
+spec:
+  selector:
+    matchLabels:
+      app: nginx
+  serviceName: nginx
+  replicas: 3
+  template:
+    metadata:
+      labels:
+        app: nginx
+    spec:
+      terminationGracePeriodSeconds: 10
+      containers:
+        - name: nginx
+          image: registry.k8s.io/nginx-slim:0.24
+          ports:
+            - containerPort: 80
+              name: web
+          volumeMounts:
+            - name: config
+              mountPath: "/config"
+          envFrom:
+            - configMapRef:
+                name: web-config-2
+      volumes:
+        - name: config
+          configMap:
+            name: web-config

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

@@ -46,3 +46,10 @@ rules:
       - pods/eviction
     verbs:
       - create
+  # Required to maintain MutatingWebhookConfigurations. The operator needs to do this, as it needs
+  # to enter e.g. it's generated certificate in the webhooks.
+  - apiGroups: [admissionregistration.k8s.io]
+    resources: [mutatingwebhookconfigurations]
+    verbs:
+      - create
+      - patch

rust/operator-binary/Cargo.toml

@@ -15,6 +15,7 @@ anyhow.workspace = true
 clap.workspace = true
 http.workspace = true
 futures.workspace = true
+json-patch.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 snafu.workspace = true

rust/operator-binary/src/main.rs

@@ -2,10 +2,10 @@
 // This will need changes in our and upstream error types.
 #![allow(clippy::large_enum_variant)]
 
-mod restart_controller;
-
+use anyhow::anyhow;
 use clap::Parser;
-use futures::FutureExt;
+use futures::{FutureExt, TryFutureExt};
+use restart_controller::statefulset::create_context;
 use stackable_operator::{
     YamlSchema as _,
     cli::{Command, RunArguments},
@@ -17,16 +17,37 @@ use stackable_operator::{
     shared::yaml::SerializeOptions,
     telemetry::Tracing,
 };
+use webhook::create_webhook_server;
+
+mod restart_controller;
+mod utils;
+mod webhook;
 
 mod built_info {
     include!(concat!(env!("OUT_DIR"), "/built.rs"));
 }
 
+pub const OPERATOR_NAME: &str = "commons.stackable.tech";
+pub const FIELD_MANAGER: &str = "commons-operator";
+
 #[derive(Parser)]
 #[clap(about, author)]
 struct Opts {
     #[clap(subcommand)]
-    cmd: Command,
+    cmd: Command<CommonsOperatorRunArguments>,
+}
+
+#[derive(Debug, PartialEq, Eq, Parser)]
+pub struct CommonsOperatorRunArguments {
+    #[command(flatten)]
+    pub common: RunArguments,
+
+    /// Don't start the controller mutating webhook and maintain the MutatingWebhookConfiguration.
+    ///
+    /// The mutating webhook is used to prevent an unneeded restart of the first Pod of freshly
+    /// created StatefulSets. It can be turned off in case you can accept an unneeded Pod restart.
+    #[arg(long, env)]
+    pub disable_restarter_mutating_webhook: bool,
 }
 
 #[tokio::main]
@@ -41,12 +62,16 @@ async fn main() -> anyhow::Result<()> {
             S3Bucket::merged_crd(S3BucketVersion::V1Alpha1)?
                 .print_yaml_schema(built_info::PKG_VERSION, SerializeOptions::default())?;
         }
-        Command::Run(RunArguments {
-            product_config: _,
-            watch_namespace,
-            operator_environment: _,
-            maintenance,
-            common,
+        Command::Run(CommonsOperatorRunArguments {
+            common:
+                RunArguments {
+                    product_config: _,
+                    watch_namespace,
+                    operator_environment,
+                    maintenance,
+                    common,
+                },
+            disable_restarter_mutating_webhook,
         }) => {
             // NOTE (@NickLarsenNZ): Before stackable-telemetry was used:
             // - The console log level was set by `COMMONS_OPERATOR_LOG`, and is now `CONSOLE_LOG` (when using Tracing::pre_configured).
@@ -76,12 +101,34 @@ async fn main() -> anyhow::Result<()> {
             )
             .await?;
 
-            let sts_restart_controller =
-                restart_controller::statefulset::start(&client, &watch_namespace).map(anyhow::Ok);
+            let (ctx, cm_store_tx, secret_store_tx) = create_context(client.clone());
+            let sts_restart_controller = restart_controller::statefulset::start(
+                ctx.clone(),
+                cm_store_tx,
+                secret_store_tx,
+                &watch_namespace,
+            )
+            .map(anyhow::Ok);
             let pod_restart_controller =
                 restart_controller::pod::start(&client, &watch_namespace).map(anyhow::Ok);
 
-            futures::try_join!(sts_restart_controller, pod_restart_controller, eos_checker)?;
+            let webhook_server = create_webhook_server(
+                ctx,
+                &operator_environment,
+                disable_restarter_mutating_webhook,
+                client.as_kube_client(),
+            )
+            .await?;
+            let webhook_server = webhook_server
+                .run()
+                .map_err(|err| anyhow!(err).context("failed to run webhook"));
+
+            futures::try_join!(
+                sts_restart_controller,
+                pod_restart_controller,
+                webhook_server,
+                eos_checker,
+            )?;
         }
     }
 

rust/operator-binary/src/restart_controller/pod.rs

@@ -42,7 +42,7 @@ enum Error {
     #[snafu(display(
         "failed to parse expiry timestamp annotation ({annotation:?}: {value:?}) as RFC 3999"
     ))]
-    UnparseableExpiryTimestamp {
+    UnparsableExpiryTimestamp {
         source: chrono::ParseError,
         annotation: String,
         value: String,
@@ -60,7 +60,7 @@ impl ReconcilerError for Error {
         match self {
             Error::PodHasNoName => None,
             Error::PodHasNoNamespace => None,
-            Error::UnparseableExpiryTimestamp {
+            Error::UnparsableExpiryTimestamp {
                 source: _,
                 annotation: _,
                 value: _,
@@ -73,6 +73,8 @@ impl ReconcilerError for Error {
 pub async fn start(client: &Client, watch_namespace: &WatchNamespace) {
     let controller = Controller::new(
         watch_namespace.get_api::<PartialObjectMeta<Pod>>(client),
+        // TODO: Can we only watch a subset of Pods with a specify label, e.g.
+        // vendor=Stackable to reduce the memory footprint?
         watcher::Config::default(),
     );
     let event_recorder = Arc::new(Recorder::new(
@@ -124,7 +126,7 @@ async fn reconcile(pod: Arc<PartialObjectMeta<Pod>>, ctx: Arc<Ctx>) -> Result<Ac
         .flatten()
         .filter(|(k, _)| k.starts_with("restarter.stackable.tech/expires-at."))
         .map(|(k, v)| {
-            DateTime::parse_from_rfc3339(v).context(UnparseableExpiryTimestampSnafu {
+            DateTime::parse_from_rfc3339(v).context(UnparsableExpiryTimestampSnafu {
                 annotation: k,
                 value: v,
             })