chore: Update to operator-rs 0.84.1

Author: sbernauer

Cargo.lock

@@ -13,21 +13,22 @@ repository = "https://github.com/stackabletech/hdfs-operator"
 anyhow = "1.0"
 built = { version = "0.7", features = ["chrono", "git2"] }
 clap = "4.5"
+const_format = "0.2"
 futures = { version = "0.3", features = ["compat"] }
 indoc = "2.0"
-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.1" }
 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"] }
 tracing = "0.1"
 tracing-futures = { version = "0.2", features = ["futures-03"] }
 
-#[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 = "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 = "main" }

Cargo.nix

@@ -91,6 +91,7 @@ rules:
       - events
     verbs:
       - create
+      - patch
   - apiGroups:
       - listeners.stackable.tech
     resources:

Cargo.toml

@@ -37,7 +37,9 @@ use stackable_operator::{
         self,
         spec::{ContainerLogConfig, Logging},
     },
-    role_utils::{GenericRoleConfig, Role, RoleGroup, RoleGroupRef},
+    role_utils::{
+        GenericProductSpecificCommonConfig, GenericRoleConfig, Role, RoleGroup, RoleGroupRef,
+    },
     schemars::{self, JsonSchema},
     status::condition::{ClusterCondition, HasStatusCondition},
     time::Duration,
@@ -539,23 +541,23 @@ impl HdfsCluster {
     pub fn namenode_rolegroup(
         &self,
         role_group: &str,
-    ) -> Option<&RoleGroup<NameNodeConfigFragment>> {
+    ) -> Option<&RoleGroup<NameNodeConfigFragment, GenericProductSpecificCommonConfig>> {
         self.spec.name_nodes.as_ref()?.role_groups.get(role_group)
     }
 
     /// Get a reference to the datanode [`RoleGroup`] struct if it exists.
     pub fn datanode_rolegroup(
         &self,
         role_group: &str,
-    ) -> Option<&RoleGroup<DataNodeConfigFragment>> {
+    ) -> Option<&RoleGroup<DataNodeConfigFragment, GenericProductSpecificCommonConfig>> {
         self.spec.data_nodes.as_ref()?.role_groups.get(role_group)
     }
 
     /// Get a reference to the journalnode [`RoleGroup`] struct if it exists.
     pub fn journalnode_rolegroup(
         &self,
         role_group: &str,
-    ) -> Option<&RoleGroup<JournalNodeConfigFragment>> {
+    ) -> Option<&RoleGroup<JournalNodeConfigFragment, GenericProductSpecificCommonConfig>> {
         self.spec
             .journal_nodes
             .as_ref()?

crate-hashes.json

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

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

@@ -1,14 +1,13 @@
 use snafu::{ResultExt, Snafu};
-use stackable_hdfs_crd::{constants::CONTROLLER_NAME, HdfsCluster, HdfsRole};
+use stackable_hdfs_crd::{HdfsCluster, HdfsRole};
 use stackable_operator::{
-    client::Client,
-    kube::runtime::{
-        events::{Event, EventType, Recorder, Reporter},
-        reflector::ObjectRef,
-    },
+    k8s_openapi::api::core::v1::ObjectReference,
+    kube::runtime::events::{Event, EventType},
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
+use crate::hdfs_controller::Ctx;
+
 #[derive(Snafu, Debug, EnumDiscriminants)]
 #[strum_discriminants(derive(IntoStaticStr))]
 pub enum Error {
@@ -18,30 +17,25 @@ pub enum Error {
     },
 }
 
-/// Publish a Kubernetes event for the `hdfs` cluster resource.
-pub async fn publish_event(
-    hdfs: &HdfsCluster,
-    client: &Client,
-    action: &str,
-    reason: &str,
-    message: &str,
+/// Publish a Kubernetes warning event for the `hdfs` cluster resource.
+pub async fn publish_warning_event(
+    ctx: &Ctx,
+    hdfs_object_ref: &ObjectReference,
+    action: String,
+    reason: String,
+    message: String,
 ) -> Result<(), Error> {
-    let reporter = Reporter {
-        controller: CONTROLLER_NAME.into(),
-        instance: None,
-    };
-
-    let object_ref = ObjectRef::from_obj(hdfs);
-
-    let recorder = Recorder::new(client.as_kube_client(), reporter, object_ref.into());
-    recorder
-        .publish(Event {
-            action: action.into(),
-            reason: reason.into(),
-            note: Some(message.into()),
-            type_: EventType::Warning,
-            secondary: None,
-        })
+    ctx.event_recorder
+        .publish(
+            &Event {
+                action,
+                reason,
+                note: Some(message),
+                type_: EventType::Warning,
+                secondary: None,
+            },
+            hdfs_object_ref,
+        )
         .await
         .context(PublishEventSnafu)
 }

rust/crd/src/lib.rs

@@ -3,13 +3,13 @@ use std::{
     sync::Arc,
 };
 
+use const_format::concatcp;
 use product_config::{
     types::PropertyNameKind,
     writer::{to_hadoop_xml, to_java_properties_string, PropertiesWriterError},
     ProductConfigManager,
 };
 use snafu::{OptionExt, ResultExt, Snafu};
-use stackable_operator::k8s_openapi::api::core::v1::ServiceAccount;
 use stackable_operator::{
     builder::{
         configmap::ConfigMapBuilder,
@@ -48,6 +48,9 @@ use stackable_operator::{
     time::Duration,
     utils::cluster_info::KubernetesClusterInfo,
 };
+use stackable_operator::{
+    k8s_openapi::api::core::v1::ServiceAccount, kube::runtime::events::Recorder,
+};
 use strum::{EnumDiscriminants, IntoEnumIterator, IntoStaticStr};
 
 use stackable_hdfs_crd::{
@@ -60,7 +63,7 @@ use crate::{
     config::{CoreSiteConfigBuilder, HdfsSiteConfigBuilder},
     container::{self, ContainerConfig, TLS_STORE_DIR, TLS_STORE_PASSWORD},
     discovery::{self, build_discovery_configmap},
-    event::{build_invalid_replica_message, publish_event},
+    event::{build_invalid_replica_message, publish_warning_event},
     operations::{
         graceful_shutdown::{self, add_graceful_shutdown_config},
         pdb::add_pdbs,
@@ -71,7 +74,9 @@ use crate::{
 };
 
 pub const RESOURCE_MANAGER_HDFS_CONTROLLER: &str = "hdfs-operator-hdfs-controller";
-const HDFS_CONTROLLER: &str = "hdfs-controller";
+const HDFS_CONTROLLER_NAME: &str = "hdfs-controller";
+pub const HDFS_FULL_CONTROLLER_NAME: &str = concatcp!(HDFS_CONTROLLER_NAME, '.', OPERATOR_NAME);
+
 const DOCKER_IMAGE_BASE_NAME: &str = "hadoop";
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
@@ -258,6 +263,7 @@ pub type HdfsOperatorResult<T> = Result<T, Error>;
 pub struct Ctx {
     pub client: Client,
     pub product_config: ProductConfigManager,
+    pub event_recorder: Arc<Recorder>,
 }
 
 pub async fn reconcile_hdfs(
@@ -296,6 +302,7 @@ pub async fn reconcile_hdfs(
     )
     .context(InvalidProductConfigSnafu)?;
 
+    let hdfs_obj_ref = hdfs.object_ref(&());
     // A list of all name and journal nodes across all role groups is needed for all ConfigMaps and initialization checks.
     let namenode_podrefs = hdfs
         .pod_refs(&HdfsRole::NameNode)
@@ -308,7 +315,7 @@ pub async fn reconcile_hdfs(
         APP_NAME,
         OPERATOR_NAME,
         RESOURCE_MANAGER_HDFS_CONTROLLER,
-        &hdfs.object_ref(&()),
+        &hdfs_obj_ref,
         ClusterResourceApplyStrategy::from(&hdfs.spec.cluster_operation),
     )
     .context(CreateClusterResourcesSnafu)?;
@@ -367,13 +374,13 @@ pub async fn reconcile_hdfs(
             continue;
         };
 
-        if let Some(content) = build_invalid_replica_message(hdfs, &role, dfs_replication) {
-            publish_event(
-                hdfs,
-                client,
-                "Reconcile",
-                "Invalid replicas",
-                content.as_ref(),
+        if let Some(message) = build_invalid_replica_message(hdfs, &role, dfs_replication) {
+            publish_warning_event(
+                &ctx,
+                &hdfs_obj_ref,
+                "Reconcile".to_owned(),
+                "Invalid replicas".to_owned(),
+                message,
             )
             .await
             .context(FailedToCreateClusterEventSnafu)?;
@@ -488,7 +495,7 @@ pub async fn reconcile_hdfs(
     let discovery_cm = build_discovery_configmap(
         hdfs,
         &client.kubernetes_cluster_info,
-        HDFS_CONTROLLER,
+        HDFS_CONTROLLER_NAME,
         &hdfs
             .namenode_listener_refs(client)
             .await

rust/operator-binary/Cargo.toml

@@ -2,7 +2,8 @@ use std::sync::Arc;
 
 use built_info::PKG_VERSION;
 use clap::{crate_description, crate_version, Parser};
-use futures::StreamExt;
+use futures::{pin_mut, StreamExt};
+use hdfs_controller::HDFS_FULL_CONTROLLER_NAME;
 use product_config::ProductConfigManager;
 use stackable_hdfs_crd::{constants::*, HdfsCluster};
 use stackable_operator::{
@@ -12,10 +13,13 @@ use stackable_operator::{
         apps::v1::StatefulSet,
         core::v1::{ConfigMap, Service},
     },
-    kube::core::DeserializeGuard,
     kube::{
         api::PartialObjectMeta,
-        runtime::{reflector, watcher, Controller},
+        core::DeserializeGuard,
+        runtime::{
+            events::{Recorder, Reporter},
+            reflector, watcher, Controller,
+        },
         Api,
     },
     kvp::ObjectLabels,
@@ -95,6 +99,14 @@ pub async fn create_controller(
 ) {
     let (store, store_w) = reflector::store();
 
+    let hdfs_event_recorder = Arc::new(Recorder::new(
+        client.as_kube_client(),
+        Reporter {
+            controller: HDFS_FULL_CONTROLLER_NAME.to_string(),
+            instance: None,
+        },
+    ));
+
     // The topology provider will need to build label information by querying kubernetes nodes and this
     // requires the clusterrole 'hdfs-clusterrole-nodes': this is bound to each deployed HDFS cluster
     // via a patch.
@@ -107,7 +119,8 @@ pub async fn create_controller(
     )
     .then(|ev| {
         hdfs_clusterrolebinding_nodes_controller::reconcile(client.as_kube_client(), &store, ev)
-    });
+    })
+    .collect::<()>();
 
     let hdfs_controller = Controller::new(
         namespace.get_api::<DeserializeGuard<HdfsCluster>>(&client),
@@ -132,14 +145,31 @@ pub async fn create_controller(
         Arc::new(hdfs_controller::Ctx {
             client: client.clone(),
             product_config,
+            event_recorder: hdfs_event_recorder.clone(),
         }),
     )
-    .map(|res| report_controller_reconciled(&client, CONTROLLER_NAME, &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 hdfs_event_recorder = hdfs_event_recorder.clone();
+            async move {
+                report_controller_reconciled(
+                    &hdfs_event_recorder,
+                    HDFS_FULL_CONTROLLER_NAME,
+                    &result,
+                )
+                .await;
+            }
+        },
+    )
     .instrument(info_span!("hdfs_controller"));
 
-    futures::stream::select(hdfs_controller, reflector)
-        .collect::<()>()
-        .await;
+    pin_mut!(hdfs_controller, reflector);
+    // kube-runtime's Controller will tokio::spawn each reconciliation, so this only concerns the internal watch machinery
+    futures::future::select(hdfs_controller, reflector).await;
 }
 
 /// Creates recommended `ObjectLabels` to be used in deployed resources