chore: Update to operator-rs 0.84.0

Author: sbernauer

Cargo.lock

@@ -13,6 +13,7 @@ repository = "https://github.com/stackabletech/zookeeper-operator"
 anyhow = "1.0"
 built = { version = "0.7", features = ["chrono", "git2"] }
 clap = "4.5"
+const_format = "0.2"
 fnv = "1.0"
 futures = { version = "0.3", features = ["compat"] }
 indoc = "2.0"
@@ -22,7 +23,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
 snafu = "0.8"
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.83.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.84.0" }
 product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.7.0" }
 strum = { version = "0.26", features = ["derive"] }
 tokio = { version = "1.40", features = ["full"] }

Cargo.nix

@@ -106,6 +106,7 @@ rules:
       - events
     verbs:
       - create
+      - patch
   - apiGroups:
       - {{ include "operator.name" . }}.stackable.tech
     resources:

Cargo.toml

@@ -25,7 +25,9 @@ use stackable_operator::{
     memory::{BinaryMultiple, MemoryQuantity},
     product_config_utils::{self, Configuration},
     product_logging::{self, spec::Logging},
-    role_utils::{GenericRoleConfig, Role, RoleGroup, RoleGroupRef},
+    role_utils::{
+        GenericProductSpecificCommonConfig, GenericRoleConfig, Role, RoleGroup, RoleGroupRef,
+    },
     schemars::{self, JsonSchema},
     status::condition::{ClusterCondition, HasStatusCondition},
     time::Duration,
@@ -517,7 +519,7 @@ impl ZookeeperCluster {
     pub fn rolegroup(
         &self,
         rolegroup_ref: &RoleGroupRef<ZookeeperCluster>,
-    ) -> Result<RoleGroup<ZookeeperConfigFragment>, Error> {
+    ) -> Result<RoleGroup<ZookeeperConfigFragment, GenericProductSpecificCommonConfig>, Error> {
         let role_variant = ZookeeperRole::from_str(&rolegroup_ref.role).with_context(|_| {
             UnknownZookeeperRoleSnafu {
                 role: rolegroup_ref.role.to_owned(),

crate-hashes.json

@@ -14,6 +14,7 @@ stackable-zookeeper-crd = { path = "../crd" }
 
 anyhow.workspace = true
 clap.workspace = true
+const_format.workspace = true
 fnv.workspace = true
 futures.workspace = true
 indoc.workspace = true

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

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use clap::{crate_description, crate_version, Parser};
-use futures::StreamExt;
+use futures::{pin_mut, StreamExt};
 use stackable_operator::{
     cli::{Command, ProductOperatorRun},
     k8s_openapi::api::{
@@ -10,16 +10,19 @@ use stackable_operator::{
     },
     kube::{
         core::DeserializeGuard,
-        runtime::{reflector::ObjectRef, watcher, Controller},
+        runtime::{
+            events::{Recorder, Reporter},
+            reflector::ObjectRef,
+            watcher, Controller,
+        },
         Resource,
     },
     logging::controller::report_controller_reconciled,
     CustomResourceExt,
 };
 use stackable_zookeeper_crd::{ZookeeperCluster, ZookeeperZnode, APP_NAME, OPERATOR_NAME};
 
-use crate::zk_controller::ZK_CONTROLLER_NAME;
-use crate::znode_controller::ZNODE_CONTROLLER_NAME;
+use crate::{zk_controller::ZK_FULL_CONTROLLER_NAME, znode_controller::ZNODE_FULL_CONTROLLER_NAME};
 
 mod command;
 mod discovery;
@@ -79,13 +82,20 @@ async fn main() -> anyhow::Result<()> {
             )
             .await?;
 
-            let zk_controller_builder = Controller::new(
+            let zk_controller = Controller::new(
                 watch_namespace.get_api::<DeserializeGuard<ZookeeperCluster>>(&client),
                 watcher::Config::default(),
             );
 
-            let zk_store = zk_controller_builder.store();
-            let zk_controller = zk_controller_builder
+            let zk_event_recorder = Arc::new(Recorder::new(
+                client.as_kube_client(),
+                Reporter {
+                    controller: ZK_FULL_CONTROLLER_NAME.to_string(),
+                    instance: None,
+                },
+            ));
+            let zk_store = zk_controller.store();
+            let zk_controller = zk_controller
                 .owns(
                     watch_namespace.get_api::<DeserializeGuard<Service>>(&client),
                     watcher::Config::default(),
@@ -125,19 +135,38 @@ async fn main() -> anyhow::Result<()> {
                         product_config,
                     }),
                 )
-                .map(|res| {
-                    report_controller_reconciled(
-                        &client,
-                        &format!("{ZK_CONTROLLER_NAME}.{OPERATOR_NAME}"),
-                        &res,
-                    );
-                });
-            let znode_controller_builder = Controller::new(
+                // We can let the reporting happen in the background
+                .for_each_concurrent(
+                    16, // concurrency limit
+                    |result| {
+                        // The event_recorder needs to be shared across all invocations, so that
+                        // events are correctly aggregated
+                        let event_recorder = zk_event_recorder.clone();
+                        async move {
+                            report_controller_reconciled(
+                                &event_recorder,
+                                ZK_FULL_CONTROLLER_NAME,
+                                &result,
+                            )
+                            .await;
+                        }
+                    },
+                );
+
+            let znode_controller = Controller::new(
                 watch_namespace.get_api::<DeserializeGuard<ZookeeperZnode>>(&client),
                 watcher::Config::default(),
             );
-            let znode_store = znode_controller_builder.store();
-            let znode_controller = znode_controller_builder
+            let znode_event_recorder = Arc::new(Recorder::new(
+                client.as_kube_client(),
+                Reporter {
+                    controller: ZNODE_FULL_CONTROLLER_NAME.to_string(),
+                    instance: None,
+                },
+            ));
+
+            let znode_store = znode_controller.store();
+            let znode_controller = znode_controller
                 .owns(
                     watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
                     watcher::Config::default(),
@@ -168,17 +197,27 @@ async fn main() -> anyhow::Result<()> {
                         client: client.clone(),
                     }),
                 )
-                .map(|res| {
-                    report_controller_reconciled(
-                        &client,
-                        &format!("{ZNODE_CONTROLLER_NAME}.{OPERATOR_NAME}"),
-                        &res,
-                    );
-                });
+                // We can let the reporting happen in the background
+                .for_each_concurrent(
+                    16, // concurrency limit
+                    move |result| {
+                        // The event_recorder needs to be shared across all invocations, so that
+                        // events are correctly aggregated
+                        let event_recorder = znode_event_recorder.clone();
+                        async move {
+                            report_controller_reconciled(
+                                &event_recorder,
+                                ZNODE_FULL_CONTROLLER_NAME,
+                                &result,
+                            )
+                            .await;
+                        }
+                    },
+                );
 
-            futures::stream::select(zk_controller, znode_controller)
-                .collect::<()>()
-                .await;
+            pin_mut!(zk_controller, znode_controller);
+            // kube-runtime's Controller will tokio::spawn each reconciliation, so this only concerns the internal watch machinery
+            futures::future::select(zk_controller, znode_controller).await;
         }
     }
 

rust/crd/src/lib.rs

@@ -7,6 +7,7 @@ use std::{
     sync::Arc,
 };
 
+use const_format::concatcp;
 use fnv::FnvHasher;
 use indoc::formatdoc;
 use product_config::{
@@ -82,6 +83,8 @@ use crate::{
 };
 
 pub const ZK_CONTROLLER_NAME: &str = "zookeepercluster";
+pub const ZK_FULL_CONTROLLER_NAME: &str = concatcp!(ZK_CONTROLLER_NAME, '.', OPERATOR_NAME);
+
 pub const ZK_UID: i64 = 1000;
 pub struct Ctx {
     pub client: stackable_operator::client::Client,

rust/operator-binary/Cargo.toml

@@ -3,6 +3,7 @@
 //! See [`ZookeeperZnode`] for more details.
 use std::{borrow::Cow, convert::Infallible, sync::Arc};
 
+use const_format::concatcp;
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
@@ -32,6 +33,7 @@ use crate::{
 };
 
 pub const ZNODE_CONTROLLER_NAME: &str = "znode";
+pub const ZNODE_FULL_CONTROLLER_NAME: &str = concatcp!(ZNODE_CONTROLLER_NAME, '.', OPERATOR_NAME);
 
 pub struct Ctx {
     pub client: stackable_operator::client::Client,