chore: Aggregate emitted Kubernetes events

Author: sbernauer

crates/stackable-operator/CHANGELOG.md

@@ -4,6 +4,17 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- BREAKING: Aggregate emitted Kubernetes events on the CustomResources thanks to the new [kube feature](https://github.com/kube-rs/controller-rs/pull/116).
+  Instead of spamming the same Warning over and over it now uses `EventSeries` to have a nice - single - Age `3s (x11 over 53s)`.
+
+  First off this makes `report_controller_error` async.
+  Also you now need to pass a `Recorder` instead of a `Client` instead.
+  You need to store the `Recorder` across all `reconcile` invocations, so that it can do the aggregation correctly.
+
+  Also make sure that the operator has the permission to `patch` events (previously only `create` was needed)! ([#867]).
+
 ### Changed
 
 - BREAKING: Bump all Rust dependencies (except opentelemetry-related) and enable Kubernetes 1.32 (via `kube` 0.98.0 and `k8s-openapi` 0.23.0) ([#867]).

crates/stackable-operator/src/logging/controller.rs

@@ -8,13 +8,14 @@ use kube::{
     core::DynamicObject,
     runtime::{
         controller::{self, Action},
+        events::Recorder,
         reflector::ObjectRef,
     },
     Resource,
 };
 use tracing;
 
-use crate::{client::Client, logging::k8s_events::publish_controller_error_as_k8s_event};
+use crate::logging::k8s_events::publish_controller_error_as_k8s_event;
 
 /// [`Error`] extensions that help report reconciliation errors
 ///
@@ -43,8 +44,8 @@ pub trait ReconcilerError: Error {
 /// * Kubernetes [`Event`]s, if there is an error that is relevant to the end user
 ///
 /// [`Event`]: `k8s_openapi::api::events::v1::Event`
-pub fn report_controller_reconciled<K, ReconcileErr, QueueErr>(
-    client: &Client,
+pub async fn report_controller_reconciled<K, ReconcileErr, QueueErr>(
+    recorder: &Recorder,
     controller_name: &str,
     result: &Result<(ObjectRef<K>, Action), controller::Error<ReconcileErr, QueueErr>>,
 ) where
@@ -60,13 +61,13 @@ pub fn report_controller_reconciled<K, ReconcileErr, QueueErr>(
                 "Reconciled object"
             );
         }
-        Err(err) => report_controller_error(client, controller_name, err),
+        Err(err) => report_controller_error(recorder, controller_name, err).await,
     }
 }
 
 /// Reports an error to the operator administrator and, if relevant, the end user
-fn report_controller_error<ReconcileErr, QueueErr>(
-    client: &Client,
+async fn report_controller_error<ReconcileErr, QueueErr>(
+    recorder: &Recorder,
     controller_name: &str,
     error: &controller::Error<ReconcileErr, QueueErr>,
 ) where
@@ -78,5 +79,5 @@ fn report_controller_error<ReconcileErr, QueueErr>(
         error = error as &dyn std::error::Error,
         "Failed to reconcile object",
     );
-    publish_controller_error_as_k8s_event(client, controller_name, error);
+    publish_controller_error_as_k8s_event(recorder, error).await;
 }

crates/stackable-operator/src/logging/k8s_events.rs

@@ -2,12 +2,10 @@
 
 use std::error::Error;
 
-use crate::client::Client;
 use kube::runtime::{
     controller,
-    events::{Event, EventType, Recorder, Reporter},
+    events::{Event, EventType, Recorder},
 };
-use tracing::Instrument;
 
 use super::controller::ReconcilerError;
 
@@ -41,10 +39,9 @@ fn error_to_event<E: ReconcilerError>(err: &E) -> Event {
 /// Reports an error coming from a controller to Kubernetes
 ///
 /// This is intended to be executed on the log entries returned by [`kube::runtime::Controller::run`]
-#[tracing::instrument(skip(client))]
-pub fn publish_controller_error_as_k8s_event<ReconcileErr, QueueErr>(
-    client: &Client,
-    controller: &str,
+#[tracing::instrument(skip(recorder))]
+pub async fn publish_controller_error_as_k8s_event<ReconcileErr, QueueErr>(
+    recorder: &Recorder,
     controller_error: &controller::Error<ReconcileErr, QueueErr>,
 ) where
     ReconcileErr: ReconcilerError,
@@ -55,27 +52,15 @@ pub fn publish_controller_error_as_k8s_event<ReconcileErr, QueueErr>(
         return;
     };
 
-    let recorder = Recorder::new(
-        client.as_kube_client(),
-        Reporter {
-            controller: controller.to_string(),
-            instance: None,
-        },
-    );
-    let event = error_to_event(error);
-    let obj_ref = obj.clone().into();
-    // Run in the background
-    tokio::spawn(
-        async move {
-            if let Err(err) = recorder.publish(&event, &obj_ref).await {
-                tracing::error!(
-                    error = &err as &dyn std::error::Error,
-                    "Failed to report error as K8s event"
-                );
-            }
-        }
-        .in_current_span(),
-    );
+    if let Err(err) = recorder
+        .publish(&error_to_event(error), &obj.clone().into())
+        .await
+    {
+        tracing::error!(
+            error = &err as &dyn std::error::Error,
+            "Failed to report error as K8s event"
+        );
+    }
 }
 
 mod message {