chore: Run cargo +nightly fmt

Author: sbernauer

Cargo.lock

@@ -13,17 +13,18 @@ repository = "https://github.com/stackabletech/hive-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"
 pin-project = "1.1"
-rstest = "0.23"
+rstest = "0.24"
 semver = "1.0"
 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

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

Cargo.toml

@@ -23,7 +23,9 @@ use stackable_operator::{
     kube::{runtime::reflector::ObjectRef, CustomResource, ResourceExt},
     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,
@@ -603,7 +605,7 @@ impl HiveCluster {
     pub fn rolegroup(
         &self,
         rolegroup_ref: &RoleGroupRef<HiveCluster>,
-    ) -> Result<RoleGroup<MetaStoreConfigFragment>, Error> {
+    ) -> Result<RoleGroup<MetaStoreConfigFragment, GenericProductSpecificCommonConfig>, Error> {
         let role_variant =
             HiveRole::from_str(&rolegroup_ref.role).with_context(|_| UnknownHiveRoleSnafu {
                 role: rolegroup_ref.role.to_owned(),

crate-hashes.json

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

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

@@ -6,6 +6,7 @@ use std::{
     sync::Arc,
 };
 
+use const_format::concatcp;
 use fnv::FnvHasher;
 use indoc::formatdoc;
 use product_config::{
@@ -22,7 +23,6 @@ use stackable_hive_crd::{
     STACKABLE_CONFIG_MOUNT_DIR_NAME, STACKABLE_LOG_CONFIG_MOUNT_DIR,
     STACKABLE_LOG_CONFIG_MOUNT_DIR_NAME, STACKABLE_LOG_DIR, STACKABLE_LOG_DIR_NAME,
 };
-
 use stackable_operator::{
     builder::{
         self,
@@ -87,19 +87,23 @@ use stackable_operator::{
 use strum::EnumDiscriminants;
 use tracing::warn;
 
-use crate::kerberos::{add_kerberos_pod_config, kerberos_config_properties};
 use crate::{
     command::build_container_command_args,
-    discovery, kerberos,
-    kerberos::kerberos_container_start_commands,
+    discovery,
+    kerberos::{
+        self, add_kerberos_pod_config, kerberos_config_properties,
+        kerberos_container_start_commands,
+    },
     operations::{graceful_shutdown::add_graceful_shutdown_config, pdb::add_pdbs},
     product_logging::{extend_role_group_config_map, resolve_vector_aggregator_address},
     OPERATOR_NAME,
 };
 
+pub const HIVE_CONTROLLER_NAME: &str = "hivecluster";
+pub const HIVE_FULL_CONTROLLER_NAME: &str = concatcp!(HIVE_CONTROLLER_NAME, '.', OPERATOR_NAME);
+
 /// Used as runAsUser in the pod security context. This is specified in the kafka image file
 pub const HIVE_UID: i64 = 1000;
-pub const HIVE_CONTROLLER_NAME: &str = "hivecluster";
 const DOCKER_IMAGE_BASE_NAME: &str = "hive";
 
 pub const MAX_HIVE_LOG_FILES_SIZE: MemoryQuantity = MemoryQuantity {

rust/crd/src/lib.rs

@@ -1,12 +1,11 @@
 mod command;
 mod controller;
 mod discovery;
-
 mod kerberos;
 mod operations;
 mod product_logging;
 
-use crate::controller::HIVE_CONTROLLER_NAME;
+use std::sync::Arc;
 
 use clap::{crate_description, crate_version, Parser};
 use futures::stream::StreamExt;
@@ -17,12 +16,18 @@ use stackable_operator::{
         apps::v1::StatefulSet,
         core::v1::{ConfigMap, Service},
     },
-    kube::core::DeserializeGuard,
-    kube::runtime::{watcher, Controller},
+    kube::{
+        core::DeserializeGuard,
+        runtime::{
+            events::{Recorder, Reporter},
+            watcher, Controller,
+        },
+    },
     logging::controller::report_controller_reconciled,
     CustomResourceExt,
 };
-use std::sync::Arc;
+
+use crate::controller::HIVE_FULL_CONTROLLER_NAME;
 
 mod built_info {
     include!(concat!(env!("OUT_DIR"), "/built.rs"));
@@ -72,6 +77,13 @@ async fn main() -> anyhow::Result<()> {
                 &cluster_info_opts,
             )
             .await?;
+            let event_recorder = Arc::new(Recorder::new(
+                client.as_kube_client(),
+                Reporter {
+                    controller: HIVE_FULL_CONTROLLER_NAME.to_string(),
+                    instance: None,
+                },
+            ));
 
             Controller::new(
                 watch_namespace.get_api::<DeserializeGuard<HiveCluster>>(&client),
@@ -98,14 +110,23 @@ async fn main() -> anyhow::Result<()> {
                     product_config,
                 }),
             )
-            .map(|res| {
-                report_controller_reconciled(
-                    &client,
-                    &format!("{HIVE_CONTROLLER_NAME}.{OPERATOR_NAME}"),
-                    &res,
-                );
-            })
-            .collect::<()>()
+            // 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,
+                            HIVE_FULL_CONTROLLER_NAME,
+                            &result,
+                        )
+                        .await;
+                    }
+                },
+            )
             .await;
         }
     }