it builds

razvan · razvan · commit c97e5d871b77 · 2025-03-21T14:38:53.000+02:00
diff --git a/deploy/helm/spark-k8s-operator/crds/crds.yaml b/deploy/helm/spark-k8s-operator/crds/crds.yaml
diff --git a/rust/operator-binary/src/connect/affinity.rs b/rust/operator-binary/src/connect/affinity.rs
@@ -3,8 +3,7 @@ use stackable_operator::{
     k8s_openapi::api::core::v1::PodAntiAffinity,
 };
 
-use crate::connect::crd::CONNECT_ROLE_NAME;
-use crate::crd::constants::APP_NAME;
+use crate::{connect::crd::CONNECT_ROLE_NAME, crd::constants::APP_NAME};
 
 pub fn affinity(cluster_name: &str) -> StackableAffinityFragment {
     let affinity_between_role_pods =
diff --git a/rust/operator-binary/src/connect/controller.rs b/rust/operator-binary/src/connect/controller.rs
@@ -16,14 +16,13 @@ use stackable_operator::{
         },
     },
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
-    commons::product_image_selection::ResolvedProductImage,
+    commons::{product_image_selection::ResolvedProductImage, rbac::build_rbac_resources},
     k8s_openapi::{
         api::{
             apps::v1::{StatefulSet, StatefulSetSpec},
             core::v1::{
                 ConfigMap, PodSecurityContext, Service, ServiceAccount, ServicePort, ServiceSpec,
             },
-            rbac::v1::{ClusterRole, RoleBinding, RoleRef, Subject},
         },
         apimachinery::pkg::apis::meta::v1::LabelSelector,
         DeepMerge,
@@ -47,29 +46,28 @@ use stackable_operator::{
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
-use crate::{
-    connect::crd::{
+use super::{
+    crd::{
         v1alpha1, ConnectConfig, SparkConnectServerContainer, CONNECT_CONTAINER_NAME,
-        CONNECT_CONTROLLER_NAME, CONNECT_ROLE_NAME,
+        CONNECT_CONTROLLER_NAME, CONNECT_GRPC_PORT, CONNECT_ROLE_NAME, CONNECT_UI_PORT,
     },
-    connect::pdb::add_pdbs,
+    pdb::add_pdbs,
+};
+use crate::{
     crd::{
         constants::{
-            ACCESS_KEY_ID, APP_NAME, JVM_SECURITY_PROPERTIES_FILE, MAX_SPARK_LOG_FILES_SIZE,
-            METRICS_PORT, OPERATOR_NAME, SECRET_ACCESS_KEY, SPARK_CLUSTER_ROLE,
-            SPARK_DEFAULTS_FILE_NAME, SPARK_ENV_SH_FILE_NAME, SPARK_IMAGE_BASE_NAME, SPARK_UID,
-            STACKABLE_TRUST_STORE, VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG,
+            APP_NAME, JVM_SECURITY_PROPERTIES_FILE, MAX_SPARK_LOG_FILES_SIZE, METRICS_PORT,
+            OPERATOR_NAME, SPARK_DEFAULTS_FILE_NAME, SPARK_ENV_SH_FILE_NAME, SPARK_IMAGE_BASE_NAME,
+            SPARK_UID, VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG,
             VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG,
             VOLUME_MOUNT_PATH_LOG_CONFIG,
         },
-        tlscerts, to_spark_env_sh_string,
+        to_spark_env_sh_string,
     },
     product_logging::{self, resolve_vector_aggregator_address},
     Ctx,
 };
 
-use super::crd::{CONNECT_GRPC_PORT, CONNECT_UI_PORT};
-
 #[derive(Snafu, Debug, EnumDiscriminants)]
 #[strum_discriminants(derive(IntoStaticStr))]
 #[allow(clippy::enum_variant_names)]
@@ -127,12 +125,6 @@ pub enum Error {
     #[snafu(display("failed to resolve and merge config for role and role group"))]
     FailedToResolveConfig { source: crate::connect::crd::Error },
 
-    #[snafu(display("number of cleaner rolegroups exceeds 1"))]
-    TooManyCleanerRoleGroups,
-
-    #[snafu(display("number of cleaner replicas exceeds 1"))]
-    TooManyCleanerReplicas,
-
     #[snafu(display("failed to create cluster resources"))]
     CreateClusterResources {
         source: stackable_operator::cluster_resources::Error,
@@ -156,7 +148,7 @@ pub enum Error {
     ConfigureLogging { source: LoggingError },
 
     #[snafu(display("cannot retrieve role group"))]
-    CannotRetrieveRoleGroup { source: crate::connect::crd::Error },
+    CannotRetrieveRoleGroup { source: super::crd::Error },
 
     #[snafu(display(
         "Connect server : failed to serialize [{JVM_SECURITY_PROPERTIES_FILE}] for group {}",
@@ -200,7 +192,12 @@ pub enum Error {
     },
 
     #[snafu(display("failed to merge environment config and/or overrides"))]
-    MergeEnv { source: crate::connect::crd::Error },
+    MergeEnv { source: super::crd::Error },
+
+    #[snafu(display("failed to build RBAC resources"))]
+    BuildRbacResources {
+        source: stackable_operator::commons::rbac::Error,
+    },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -250,8 +247,15 @@ pub async fn reconcile(
     .context(ResolveVectorAggregatorAddressSnafu)?;
 
     // Use a dedicated service account for connect server pods.
-    let (serviceaccount, rolebinding) =
-        build_connect_role_serviceaccount(scs, &resolved_product_image.app_version_label)?;
+    let (serviceaccount, rolebinding) = build_rbac_resources(
+        scs,
+        APP_NAME,
+        cluster_resources
+            .get_required_labels()
+            .context(GetRequiredLabelsSnafu)?,
+    )
+    .context(BuildRbacResourcesSnafu)?;
+
     let serviceaccount = cluster_resources
         .add(client, serviceaccount)
         .await
@@ -587,7 +591,7 @@ fn build_stateful_set(
 
 #[allow(clippy::result_large_err)]
 fn build_service(
-    shs: &v1alpha1::SparkConnectServer,
+    scs: &v1alpha1::SparkConnectServer,
     app_version_label: &str,
     role: &str,
     group: Option<&RoleGroupRef<v1alpha1::SparkConnectServer>>,
@@ -604,38 +608,43 @@ fn build_service(
             Some("None".to_string()),
         ),
         None => (
-            format!("{}-{}", shs.metadata.name.as_ref().unwrap(), role),
-            shs.spec.cluster_config.listener_class.k8s_service_type(),
+            format!("{}-{}", scs.metadata.name.as_ref().unwrap(), role),
+            scs.spec.cluster_config.listener_class.k8s_service_type(),
             None,
         ),
     };
 
     let selector = match group {
-        Some(rgr) => Labels::role_group_selector(shs, APP_NAME, &rgr.role, &rgr.role_group)
+        Some(rgr) => Labels::role_group_selector(scs, APP_NAME, &rgr.role, &rgr.role_group)
             .context(LabelBuildSnafu)?
             .into(),
-        None => Labels::role_selector(shs, APP_NAME, role)
+        None => Labels::role_selector(scs, APP_NAME, role)
             .context(LabelBuildSnafu)?
             .into(),
     };
 
     Ok(Service {
         metadata: ObjectMetaBuilder::new()
-            .name_and_namespace(shs)
+            .name_and_namespace(scs)
             .name(service_name)
-            .ownerreference_from_resource(shs, None, Some(true))
+            .ownerreference_from_resource(scs, None, Some(true))
             .context(ObjectMissingMetadataForOwnerRefSnafu)?
-            .with_recommended_labels(labels(shs, app_version_label, &group_name))
+            .with_recommended_labels(labels(scs, app_version_label, &group_name))
             .context(MetadataBuildSnafu)?
             .with_label(Label::try_from(("prometheus.io/scrape", "true")).context(LabelBuildSnafu)?)
             .build(),
         spec: Some(ServiceSpec {
             type_: Some(service_type),
             cluster_ip: service_cluster_ip,
             ports: Some(vec![
+                ServicePort {
+                    name: Some(String::from("grpc")),
+                    port: CONNECT_GRPC_PORT,
+                    ..ServicePort::default()
+                },
                 ServicePort {
                     name: Some(String::from("http")),
-                    port: 18080,
+                    port: CONNECT_UI_PORT,
                     ..ServicePort::default()
                 },
                 ServicePort {
@@ -651,50 +660,8 @@ fn build_service(
     })
 }
 
-// TODO: This function should be replaced with operator-rs build_rbac_resources.
-// See: https://github.com/stackabletech/spark-k8s-operator/issues/499
-#[allow(clippy::result_large_err)]
-fn build_connect_role_serviceaccount(
-    shs: &v1alpha1::SparkConnectServer,
-    app_version_label: &str,
-) -> Result<(ServiceAccount, RoleBinding)> {
-    let sa = ServiceAccount {
-        metadata: ObjectMetaBuilder::new()
-            .name_and_namespace(shs)
-            .ownerreference_from_resource(shs, None, Some(true))
-            .context(ObjectMissingMetadataForOwnerRefSnafu)?
-            .with_recommended_labels(labels(shs, app_version_label, connect_CONTROLLER_NAME))
-            .context(MetadataBuildSnafu)?
-            .build(),
-        ..ServiceAccount::default()
-    };
-    let binding = RoleBinding {
-        metadata: ObjectMetaBuilder::new()
-            .name_and_namespace(shs)
-            .ownerreference_from_resource(shs, None, Some(true))
-            .context(ObjectMissingMetadataForOwnerRefSnafu)?
-            .with_recommended_labels(labels(shs, app_version_label, connect_CONTROLLER_NAME))
-            .context(MetadataBuildSnafu)?
-            .build(),
-        role_ref: RoleRef {
-            api_group: <ClusterRole as stackable_operator::k8s_openapi::Resource>::GROUP // need to fully qualify because of "Resource" name clash
-                .to_string(),
-            kind: <ClusterRole as stackable_operator::k8s_openapi::Resource>::KIND.to_string(),
-            name: SPARK_CLUSTER_ROLE.to_string(),
-        },
-        subjects: Some(vec![Subject {
-            api_group: Some(
-                <ServiceAccount as stackable_operator::k8s_openapi::Resource>::GROUP.to_string(),
-            ),
-            kind: <ServiceAccount as stackable_operator::k8s_openapi::Resource>::KIND.to_string(),
-            name: sa.name_any(),
-            namespace: sa.namespace(),
-        }]),
-    };
-    Ok((sa, binding))
-}
-
 // TODO: revisit this
+#[allow(clippy::result_large_err)]
 fn spark_defaults(
     _cs: &v1alpha1::SparkConnectServer,
     _rolegroupref: &RoleGroupRef<v1alpha1::SparkConnectServer>,
diff --git a/rust/operator-binary/src/connect/crd.rs b/rust/operator-binary/src/connect/crd.rs
@@ -31,11 +31,13 @@ use stackable_operator::{
 use stackable_versioned::versioned;
 use strum::{Display, EnumIter};
 
-use crate::crd::constants::{
-    JVM_SECURITY_PROPERTIES_FILE, OPERATOR_NAME, SPARK_DEFAULTS_FILE_NAME, SPARK_ENV_SH_FILE_NAME,
-    VOLUME_MOUNT_PATH_LOG,
+use crate::{
+    connect::{affinity::affinity, jvm::construct_jvm_args},
+    crd::constants::{
+        JVM_SECURITY_PROPERTIES_FILE, OPERATOR_NAME, SPARK_DEFAULTS_FILE_NAME,
+        SPARK_ENV_SH_FILE_NAME, VOLUME_MOUNT_PATH_LOG,
+    },
 };
-use crate::{connect::affinity::affinity, connect::jvm::construct_jvm_args};
 
 pub const CONNECT_CONTROLLER_NAME: &str = "connect";
 pub const CONNECT_FULL_CONTROLLER_NAME: &str =
diff --git a/rust/operator-binary/src/connect/jvm.rs b/rust/operator-binary/src/connect/jvm.rs
@@ -3,10 +3,12 @@ use stackable_operator::role_utils::{
     self, GenericRoleConfig, JavaCommonConfig, JvmArgumentOverrides, Role,
 };
 
-use crate::connect::crd::ConnectConfigFragment;
-use crate::crd::constants::{
-    JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, METRICS_PORT, VOLUME_MOUNT_PATH_CONFIG,
-    VOLUME_MOUNT_PATH_LOG_CONFIG,
+use crate::{
+    connect::crd::ConnectConfigFragment,
+    crd::constants::{
+        JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, METRICS_PORT, VOLUME_MOUNT_PATH_CONFIG,
+        VOLUME_MOUNT_PATH_LOG_CONFIG,
+    },
 };
 
 #[derive(Snafu, Debug)]
diff --git a/rust/operator-binary/src/connect/pdb.rs b/rust/operator-binary/src/connect/pdb.rs
@@ -4,8 +4,10 @@ use stackable_operator::{
     commons::pdb::PdbConfig, kube::ResourceExt,
 };
 
-use crate::connect::crd::{v1alpha1, CONNECT_CONTROLLER_NAME, CONNECT_ROLE_NAME};
-use crate::crd::constants::{APP_NAME, OPERATOR_NAME};
+use crate::{
+    connect::crd::{v1alpha1, CONNECT_CONTROLLER_NAME, CONNECT_ROLE_NAME},
+    crd::constants::{APP_NAME, OPERATOR_NAME},
+};
 
 #[derive(Snafu, Debug)]
 pub enum Error {
diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs
@@ -1,7 +1,8 @@
 use std::sync::Arc;
 
 use clap::{crate_description, crate_version, Parser};
-use futures::{pin_mut, StreamExt};
+use connect::crd::{SparkConnectServer, CONNECT_FULL_CONTROLLER_NAME};
+use futures::{pin_mut, select, StreamExt};
 use history::history_controller;
 use product_config::ProductConfigManager;
 use stackable_operator::{
@@ -70,6 +71,8 @@ async fn main() -> anyhow::Result<()> {
                 .print_yaml_schema(built_info::PKG_VERSION, SerializeOptions::default())?;
             SparkHistoryServer::merged_crd(SparkHistoryServer::V1Alpha1)?
                 .print_yaml_schema(built_info::PKG_VERSION, SerializeOptions::default())?;
+            SparkConnectServer::merged_crd(SparkConnectServer::V1Alpha1)?
+                .print_yaml_schema(built_info::PKG_VERSION, SerializeOptions::default())?;
         }
         Command::Run(ProductOperatorRun {
             product_config,
@@ -246,13 +249,84 @@ async fn main() -> anyhow::Result<()> {
                 },
             );
 
-            pin_mut!(app_controller, pod_driver_controller, history_controller);
-            // kube-runtime's Controller will tokio::spawn each reconciliation, so this only concerns the internal watch machinery
-            futures::future::select(
-                futures::future::select(app_controller, pod_driver_controller),
-                history_controller,
+            // ==============================
+            // Create new object because Ctx cannot be cloned
+            let ctx = Ctx {
+                client: client.clone(),
+                product_config: product_config.load(&PRODUCT_CONFIG_PATHS)?,
+            };
+            let connect_event_recorder = Arc::new(Recorder::new(
+                client.as_kube_client(),
+                Reporter {
+                    controller: CONNECT_FULL_CONTROLLER_NAME.to_string(),
+                    instance: None,
+                },
+            ));
+            let connect_controller = Controller::new(
+                watch_namespace
+                    .get_api::<DeserializeGuard<connect::crd::v1alpha1::SparkConnectServer>>(
+                        &client,
+                    ),
+                watcher::Config::default(),
+            )
+            .owns(
+                watch_namespace
+                    .get_api::<DeserializeGuard<connect::crd::v1alpha1::SparkConnectServer>>(
+                        &client,
+                    ),
+                watcher::Config::default(),
+            )
+            .owns(
+                watch_namespace.get_api::<DeserializeGuard<StatefulSet>>(&client),
+                watcher::Config::default(),
+            )
+            .owns(
+                watch_namespace.get_api::<DeserializeGuard<Service>>(&client),
+                watcher::Config::default(),
+            )
+            .owns(
+                watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
+                watcher::Config::default(),
             )
-            .await;
+            .shutdown_on_signal()
+            .run(
+                connect::controller::reconcile,
+                connect::controller::error_policy,
+                Arc::new(ctx),
+            )
+            .instrument(info_span!("connect_controller"))
+            // 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 connect_event_recorder = connect_event_recorder.clone();
+                    async move {
+                        report_controller_reconciled(
+                            &connect_event_recorder,
+                            CONNECT_FULL_CONTROLLER_NAME,
+                            &result,
+                        )
+                        .await;
+                    }
+                },
+            );
+
+            //
+            pin_mut!(
+                app_controller,
+                pod_driver_controller,
+                history_controller,
+                connect_controller
+            );
+            // kube-runtime's Controller will tokio::spawn each reconciliation, so this only concerns the internal watch machinery
+            select! {
+                r1 = app_controller => r1,
+                r2 = pod_driver_controller => r2,
+                r3 = history_controller => r3,
+                r4 = connect_controller => r4,
+            };
         }
     }
     Ok(())
