chore: Update to operator-rs 0.84.0

Author: sbernauer

Cargo.lock

@@ -17,7 +17,7 @@ futures = { version = "0.3", features = ["compat"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 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" }
 strum = { version = "0.26", features = ["derive"] }
 tokio = { version = "1.40", features = ["full"] }
 tracing = "0.1"

Cargo.nix

@@ -32,6 +32,7 @@ rules:
       - events
     verbs:
       - create
+      - patch
   - apiGroups:
       - ""
     resources:

Cargo.toml

@@ -8,14 +8,20 @@ use stackable_operator::{
     k8s_openapi::api::core::v1::{Node, Pod},
     kube::{
         core::{error_boundary, DeserializeGuard, ObjectMeta},
-        runtime::{controller, reflector::ObjectRef, watcher, Controller},
+        runtime::{
+            controller,
+            events::{Recorder, Reporter},
+            reflector::ObjectRef,
+            watcher, Controller,
+        },
         Resource,
     },
     logging::controller::{report_controller_reconciled, ReconcilerError},
     namespace::WatchNamespace,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
+const FULL_CONTROLLER_NAME: &str = "pod.enrichment.commons.stackable.tech";
 const FIELD_MANAGER_SCOPE: &str = "enrichment.stackable.tech/pod";
 const ANNOTATION_NODE_ADDRESS: &str = "enrichment.stackable.tech/node-address";
 
@@ -58,6 +64,13 @@ impl ReconcilerError for Error {
 }
 
 pub async fn start(client: &stackable_operator::client::Client, watch_namespace: &WatchNamespace) {
+    let event_recorder = Arc::new(Recorder::new(
+        client.as_kube_client(),
+        Reporter {
+            controller: FULL_CONTROLLER_NAME.to_string(),
+            instance: None,
+        },
+    ));
     let controller = Controller::new(
         watch_namespace.get_api::<DeserializeGuard<Pod>>(client),
         watcher::Config::default().labels("enrichment.stackable.tech/enabled=true"),
@@ -87,9 +100,19 @@ pub async fn start(client: &stackable_operator::client::Client, watch_namespace:
                 client: client.clone(),
             }),
         )
-        .for_each(|res| async move {
-            report_controller_reconciled(client, "pod.enrichment.commons.stackable.tech", &res)
-        })
+        // 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 = event_recorder.clone();
+                async move {
+                    report_controller_reconciled(&event_recorder, FULL_CONTROLLER_NAME, &result)
+                        .await;
+                }
+            },
+        )
         .await;
 }
 

crate-hashes.json

@@ -12,13 +12,20 @@ use stackable_operator::{
         self,
         api::{EvictParams, PartialObjectMeta},
         core::DynamicObject,
-        runtime::{controller::Action, reflector::ObjectRef, watcher, Controller},
+        runtime::{
+            controller::Action,
+            events::{Recorder, Reporter},
+            reflector::ObjectRef,
+            watcher, Controller,
+        },
     },
     logging::controller::{report_controller_reconciled, ReconcilerError},
     namespace::WatchNamespace,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
+const FULL_CONTROLLER_NAME: &str = "pod.restarter.commons.stackable.tech";
+
 struct Ctx {
     client: Client,
 }
@@ -66,6 +73,13 @@ pub async fn start(client: &Client, watch_namespace: &WatchNamespace) {
         watch_namespace.get_api::<PartialObjectMeta<Pod>>(client),
         watcher::Config::default(),
     );
+    let event_recorder = Arc::new(Recorder::new(
+        client.as_kube_client(),
+        Reporter {
+            controller: FULL_CONTROLLER_NAME.to_string(),
+            instance: None,
+        },
+    ));
     controller
         .run(
             reconcile,
@@ -74,9 +88,19 @@ pub async fn start(client: &Client, watch_namespace: &WatchNamespace) {
                 client: client.clone(),
             }),
         )
-        .for_each(|res| async move {
-            report_controller_reconciled(client, "pod.restarter.commons.stackable.tech", &res)
-        })
+        // 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 = event_recorder.clone();
+                async move {
+                    report_controller_reconciled(&event_recorder, FULL_CONTROLLER_NAME, &result)
+                        .await;
+                }
+            },
+        )
         .await;
 }
 

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

@@ -17,6 +17,7 @@ use stackable_operator::kube::core::{error_boundary, DeserializeGuard, DynamicOb
 use stackable_operator::kube::runtime::controller::{
     trigger_self, trigger_with, Action, ReconcileRequest,
 };
+use stackable_operator::kube::runtime::events::{Recorder, Reporter};
 use stackable_operator::kube::runtime::reflector::{ObjectRef, Store};
 use stackable_operator::kube::runtime::{
     applier, metadata_watcher, reflector, watcher, Config, WatchStreamExt,
@@ -26,6 +27,8 @@ use stackable_operator::logging::controller::{report_controller_reconciled, Reco
 use stackable_operator::namespace::WatchNamespace;
 use strum::{EnumDiscriminants, IntoStaticStr};
 
+const FULL_CONTROLLER_NAME: &str = "statefulset.restarter.commons.stackable.tech";
+
 struct Ctx {
     kube: kube::Client,
     cms: Store<PartialObjectMeta<ConfigMap>>,
@@ -79,6 +82,13 @@ pub async fn start(client: &Client, watch_namespace: &WatchNamespace) {
     let secret_store = reflector::store::Writer::<PartialObjectMeta<Secret>>::new(());
     let cms_inited = Arc::new(AtomicBool::from(false));
     let secrets_inited = Arc::new(AtomicBool::from(false));
+    let event_recorder = Arc::new(Recorder::new(
+        client.as_kube_client(),
+        Reporter {
+            controller: FULL_CONTROLLER_NAME.to_string(),
+            instance: None,
+        },
+    ));
 
     applier(
         |sts, ctx| Box::pin(reconcile(sts, ctx)),
@@ -123,9 +133,18 @@ pub async fn start(client: &Client, watch_namespace: &WatchNamespace) {
         ),
         Config::default(),
     )
-    .for_each(|res| async move {
-        report_controller_reconciled(client, "statefulset.restarter.commons.stackable.tech", &res)
-    })
+    // 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 = event_recorder.clone();
+            async move {
+                report_controller_reconciled(&event_recorder, FULL_CONTROLLER_NAME, &result).await;
+            }
+        },
+    )
     .await;
 }