chore: Update to stackable-operator 0.95.0

Cargo.toml

@@ -11,14 +11,14 @@ repository = "https://github.com/stackabletech/spark-k8s-operator"
 
 [workspace.dependencies]
 product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.7.0" }
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", features = ["telemetry", "versioned"], tag = "stackable-operator-0.94.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", features = ["telemetry", "versioned"], tag = "stackable-operator-0.95.0" }
 
 anyhow = "1.0"
 built = { version = "0.8", features = ["chrono", "git2"] }
 clap = "4.5"
 const_format = "0.2"
 futures = { version = "0.3", features = ["compat"] }
-rstest = "0.25"
+rstest = "0.26"
 semver = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

rust/operator-binary/src/connect/controller.rs

@@ -3,18 +3,18 @@ use std::sync::Arc;
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
-    commons::rbac::build_rbac_resources,
+    commons::{product_image_selection, rbac::build_rbac_resources},
     kube::{
         Resource, ResourceExt,
         core::{DeserializeGuard, error_boundary},
         runtime::controller::Action,
     },
     logging::controller::ReconcilerError,
+    shared::time::Duration,
     status::condition::{
         compute_conditions, operations::ClusterOperationsConditionBuilder,
         statefulset::StatefulSetConditionBuilder,
     },
-    time::Duration,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
@@ -140,6 +140,11 @@ pub enum Error {
 
     #[snafu(display("failed to serialize executor pod template"))]
     ExecutorPodTemplateSerde { source: serde_yaml::Error },
+
+    #[snafu(display("failed to resolve product image"))]
+    ResolveProductImage {
+        source: product_image_selection::Error,
+    },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -180,7 +185,8 @@ pub async fn reconcile(
     let resolved_product_image = scs
         .spec
         .image
-        .resolve(SPARK_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION);
+        .resolve(SPARK_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION)
+        .context(ResolveProductImageSnafu)?;
 
     // Use a dedicated service account for connect server pods.
     let (service_account, role_binding) = build_rbac_resources(
@@ -202,8 +208,9 @@ pub async fn reconcile(
         .context(ApplyRoleBindingSnafu)?;
 
     // Headless service used by executors connect back to the driver
-    let service = server::build_internal_service(scs, &resolved_product_image.app_version_label)
-        .context(BuildServiceSnafu)?;
+    let service =
+        server::build_internal_service(scs, &resolved_product_image.app_version_label_value)
+            .context(BuildServiceSnafu)?;
 
     let applied_internal_service = cluster_resources
         .add(client, service.clone())

rust/operator-binary/src/connect/crd.rs

@@ -26,8 +26,8 @@ use stackable_operator::{
     },
     role_utils::{CommonConfiguration, JavaCommonConfig},
     schemars::{self, JsonSchema},
+    shared::time::Duration,
     status::condition::{ClusterCondition, HasStatusCondition},
-    time::Duration,
     versioned::versioned,
 };
 use strum::{Display, EnumIter};

rust/operator-binary/src/connect/executor.rs

@@ -123,7 +123,7 @@ pub fn executor_pod_template(
     let metadata = ObjectMetaBuilder::new()
         .with_recommended_labels(common::labels(
             scs,
-            &resolved_product_image.app_version_label,
+            &resolved_product_image.app_version_label_value,
             &SparkConnectRole::Executor.to_string(),
         ))
         .context(PodTemplateMetadataBuildSnafu)?
@@ -343,7 +343,7 @@ pub(crate) fn executor_config_map(
                 .context(ObjectMissingMetadataForOwnerRefSnafu)?
                 .with_recommended_labels(common::labels(
                     scs,
-                    &resolved_product_image.app_version_label,
+                    &resolved_product_image.app_version_label_value,
                     &SparkConnectRole::Executor.to_string(),
                 ))
                 .context(ConfigMapMetadataBuildSnafu)?

rust/operator-binary/src/connect/server.rs

@@ -178,7 +178,7 @@ pub(crate) fn server_config_map(
                 .context(ObjectMissingMetadataForOwnerRefSnafu)?
                 .with_recommended_labels(common::labels(
                     scs,
-                    &resolved_product_image.app_version_label,
+                    &resolved_product_image.app_version_label_value,
                     &SparkConnectRole::Server.to_string(),
                 ))
                 .context(MetadataBuildSnafu)?
@@ -216,8 +216,11 @@ pub(crate) fn build_stateful_set(
     args: Vec<String>,
 ) -> Result<StatefulSet, Error> {
     let server_role = SparkConnectRole::Server.to_string();
-    let recommended_object_labels =
-        common::labels(scs, &resolved_product_image.app_version_label, &server_role);
+    let recommended_object_labels = common::labels(
+        scs,
+        &resolved_product_image.app_version_label_value,
+        &server_role,
+    );
 
     let recommended_labels =
         Labels::recommended(recommended_object_labels.clone()).context(LabelBuildSnafu)?;
@@ -365,7 +368,7 @@ pub(crate) fn build_stateful_set(
             .context(ObjectMissingMetadataForOwnerRefSnafu)?
             .with_recommended_labels(common::labels(
                 scs,
-                &resolved_product_image.app_version_label,
+                &resolved_product_image.app_version_label_value,
                 &SparkConnectRole::Server.to_string(),
             ))
             .context(MetadataBuildSnafu)?
@@ -642,7 +645,7 @@ pub(crate) fn build_listener(
     let listener_class = role_config.listener_class.clone();
     let role = SparkConnectRole::Server.to_string();
     let recommended_object_labels =
-        common::labels(scs, &resolved_product_image.app_version_label, &role);
+        common::labels(scs, &resolved_product_image.app_version_label_value, &role);
 
     let listener_ports = [
         listener::v1alpha1::ListenerPort {

rust/operator-binary/src/crd/history.rs

@@ -26,7 +26,7 @@ use stackable_operator::{
     product_logging::{self, spec::Logging},
     role_utils::{GenericRoleConfig, JavaCommonConfig, Role, RoleGroup, RoleGroupRef},
     schemars::{self, JsonSchema},
-    time::Duration,
+    shared::time::Duration,
     versioned::versioned,
 };
 use strum::{Display, EnumIter};

rust/operator-binary/src/crd/logdir.rs

@@ -9,7 +9,7 @@ use stackable_operator::{
     commons::secret_class::SecretClassVolume,
     crd::s3,
     k8s_openapi::api::core::v1::{Volume, VolumeMount},
-    time::Duration,
+    shared::time::Duration,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 

rust/operator-binary/src/crd/mod.rs

@@ -39,7 +39,7 @@ use stackable_operator::{
     product_logging,
     role_utils::{CommonConfiguration, GenericRoleConfig, JavaCommonConfig, Role, RoleGroup},
     schemars::{self, JsonSchema},
-    time::Duration,
+    shared::time::Duration,
     utils::crds::raw_object_list_schema,
     versioned::versioned,
 };
@@ -1153,7 +1153,6 @@ mod tests {
         },
         product_config_utils::ValidatedRoleConfigByPropertyKind,
         product_logging::spec::Logging,
-        time::Duration,
     };
 
     use super::*;
@@ -1384,7 +1383,8 @@ mod tests {
         let resolved_product_image = spark_application
             .spec
             .spark_image
-            .resolve("spark-k8s", "0.0.0-dev");
+            .resolve("spark-k8s", "0.0.0-dev")
+            .expect("test: resolved product image is always valid");
 
         let product_config =
             ProductConfigManager::from_yaml_file("../../deploy/config-spec/properties.yaml")

rust/operator-binary/src/crd/roles.rs

@@ -33,7 +33,7 @@ use stackable_operator::{
     product_config_utils::Configuration,
     product_logging::{self, spec::Logging},
     schemars::{self, JsonSchema},
-    time::Duration,
+    shared::time::Duration,
     utils::crds::raw_object_list_schema,
 };
 use strum::{Display, EnumIter};

rust/operator-binary/src/history/history_controller.rs

@@ -21,7 +21,10 @@ use stackable_operator::{
         },
     },
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
-    commons::{product_image_selection::ResolvedProductImage, rbac::build_rbac_resources},
+    commons::{
+        product_image_selection::{self, ResolvedProductImage},
+        rbac::build_rbac_resources,
+    },
     crd::listener,
     k8s_openapi::{
         DeepMerge,
@@ -46,7 +49,7 @@ use stackable_operator::{
         },
     },
     role_utils::RoleGroupRef,
-    time::Duration,
+    shared::time::Duration,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
@@ -225,6 +228,11 @@ pub enum Error {
     ApplyGroupListener {
         source: stackable_operator::cluster_resources::Error,
     },
+
+    #[snafu(display("failed to resolve product image"))]
+    ResolveProductImage {
+        source: product_image_selection::Error,
+    },
 }
 
 impl ReconcilerError for Error {
@@ -259,7 +267,8 @@ pub async fn reconcile(
     let resolved_product_image = shs
         .spec
         .image
-        .resolve(SPARK_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION);
+        .resolve(SPARK_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION)
+        .context(ResolveProductImageSnafu)?;
     let log_dir = ResolvedLogDir::resolve(
         &shs.spec.log_file_directory,
         shs.metadata.namespace.clone(),
@@ -307,7 +316,7 @@ pub async fn reconcile(
                 shs,
                 rolegroup_config,
                 &merged_config,
-                &resolved_product_image.app_version_label,
+                &resolved_product_image.app_version_label_value,
                 &rgr,
                 &log_dir,
             )?;
@@ -370,7 +379,8 @@ fn build_group_listener(
 ) -> Result<listener::v1alpha1::Listener, Error> {
     let listener_name = group_listener_name(shs, role);
 
-    let recommended_object_labels = labels(shs, &resolved_product_image.app_version_label, "none");
+    let recommended_object_labels =
+        labels(shs, &resolved_product_image.app_version_label_value, "none");
 
     let listener_ports = [listener::v1alpha1::ListenerPort {
         name: "http".to_string(),
@@ -408,7 +418,7 @@ fn build_config_map(
     shs: &v1alpha1::SparkHistoryServer,
     config: &HashMap<PropertyNameKind, BTreeMap<String, String>>,
     merged_config: &HistoryConfig,
-    app_version_label: &str,
+    app_version_label_value: &str,
     rolegroupref: &RoleGroupRef<v1alpha1::SparkHistoryServer>,
     log_dir: &ResolvedLogDir,
 ) -> Result<ConfigMap, Error> {
@@ -435,7 +445,11 @@ fn build_config_map(
                 .name(&cm_name)
                 .ownerreference_from_resource(shs, None, Some(true))
                 .context(ObjectMissingMetadataForOwnerRefSnafu)?
-                .with_recommended_labels(labels(shs, app_version_label, &rolegroupref.role_group))
+                .with_recommended_labels(labels(
+                    shs,
+                    app_version_label_value,
+                    &rolegroupref.role_group,
+                ))
                 .context(MetadataBuildSnafu)?
                 .build(),
         )
@@ -499,7 +513,7 @@ fn build_stateful_set(
 
     let recommended_object_labels = labels(
         shs,
-        &resolved_product_image.app_version_label,
+        &resolved_product_image.app_version_label_value,
         rolegroupref.role_group.as_ref(),
     );
     let recommended_labels =
@@ -720,13 +734,13 @@ fn command_args(logdir: &ResolvedLogDir) -> Vec<String> {
 
 fn labels<'a, T>(
     shs: &'a T,
-    app_version_label: &'a str,
+    app_version_label_value: &'a str,
     role_group: &'a str,
 ) -> ObjectLabels<'a, T> {
     ObjectLabels {
         owner: shs,
         app_name: HISTORY_APP_NAME,
-        app_version: app_version_label,
+        app_version: app_version_label_value,
         operator_name: OPERATOR_NAME,
         controller_name: HISTORY_CONTROLLER_NAME,
         role: HISTORY_ROLE_NAME,

rust/operator-binary/src/main.rs

@@ -85,15 +85,15 @@ async fn main() -> anyhow::Result<()> {
         Command::Run(ProductOperatorRun {
             product_config,
             watch_namespace,
-            telemetry_arguments,
-            cluster_info_opts,
+            operator_environment: _,
+            telemetry,
+            cluster_info,
         }) => {
             // NOTE (@NickLarsenNZ): Before stackable-telemetry was used:
             // - The console log level was set by `SPARK_K8S_OPERATOR_LOG`, and is now `CONSOLE_LOG` (when using Tracing::pre_configured).
             // - The file log level was set by `SPARK_K8S_OPERATOR_LOG`, and is now set via `FILE_LOG` (when using Tracing::pre_configured).
             // - The file log directory was set by `SPARK_K8S_OPERATOR_LOG_DIRECTORY`, and is now set by `ROLLING_LOGS_DIR` (or via `--rolling-logs <DIRECTORY>`).
-            let _tracing_guard =
-                Tracing::pre_configured(built_info::PKG_NAME, telemetry_arguments).init()?;
+            let _tracing_guard = Tracing::pre_configured(built_info::PKG_NAME, telemetry).init()?;
 
             tracing::info!(
                 built_info.pkg_version = built_info::PKG_VERSION,
@@ -107,7 +107,7 @@ async fn main() -> anyhow::Result<()> {
 
             let client = stackable_operator::client::initialize_operator(
                 Some(OPERATOR_NAME.to_string()),
-                &cluster_info_opts,
+                &cluster_info,
             )
             .await?;
 

rust/operator-binary/src/pod_driver_controller.rs

@@ -9,7 +9,7 @@ use stackable_operator::{
         runtime::controller::Action,
     },
     logging::controller::ReconcilerError,
-    time::Duration,
+    shared::time::Duration,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};