diff --git a/CHANGELOG.md b/CHANGELOG.md index 74832625..612d19bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,8 +10,11 @@ All notable changes to this project will be documented in this file. server reconcilers that assembles all relevant Kubernetes resources before anything is applied ([#721]). - Bump stackable-operator to 0.114.0 ([#732]). +- The RBAC ServiceAccounts and RoleBindings of the history and connect servers are now + built with the operator-rs `v2::rbac` functions and carry the recommended labels ([#727]). [#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721 +[#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727 [#732]: https://github.com/stackabletech/spark-k8s-operator/pull/732 ## [26.7.0] - 2026-07-21 diff --git a/rust/operator-binary/src/connect/common.rs b/rust/operator-binary/src/connect/common.rs index 85cd8b8e..2cba2620 100644 --- a/rust/operator-binary/src/connect/common.rs +++ b/rust/operator-binary/src/connect/common.rs @@ -1,11 +1,12 @@ -use std::collections::BTreeMap; +use std::{collections::BTreeMap, str::FromStr}; use snafu::{ResultExt, Snafu}; use stackable_operator::v2::{ config_file_writer::{PropertiesWriterError, to_java_properties_string}, role_utils::JavaCommonConfig, + types::operator::RoleName, }; -use strum::Display; +use strum::{Display, EnumIter}; use super::crd::CONNECT_EXECUTOR_ROLE_NAME; use crate::{ @@ -29,13 +30,30 @@ pub enum Error { MetricsProperties { source: PropertiesWriterError }, } -#[derive(Clone, Debug, Display)] +#[derive(Clone, Debug, Display, EnumIter)] #[strum(serialize_all = "lowercase")] pub(crate) enum SparkConnectRole { Server, Executor, } +impl From for RoleName { + fn from(value: SparkConnectRole) -> Self { + (&value).into() + } +} + +impl From<&SparkConnectRole> for RoleName { + fn from(value: &SparkConnectRole) -> Self { + match value { + SparkConnectRole::Server => RoleName::from_str(CONNECT_SERVER_ROLE_NAME) + .expect("CONNECT_SERVER_ROLE_NAME is a valid role name"), + SparkConnectRole::Executor => RoleName::from_str(CONNECT_EXECUTOR_ROLE_NAME) + .expect("CONNECT_EXECUTOR_ROLE_NAME is a valid role name"), + } + } +} + pub(crate) fn object_name(stacklet_name: &str, role: SparkConnectRole) -> String { match role { SparkConnectRole::Server => format!("{}-{}", stacklet_name, CONNECT_SERVER_ROLE_NAME), @@ -110,3 +128,21 @@ pub(crate) fn metrics_properties( to_java_properties_string(result.iter()).context(MetricsPropertiesSnafu) } + +#[cfg(test)] +mod tests { + use stackable_operator::v2::types::operator::RoleName; + use strum::IntoEnumIterator; + + use super::SparkConnectRole; + + /// Locks the invariant behind the `expect` in the `From for RoleName` + /// impls: every variant (present and future) must map to a valid `RoleName`. + #[test] + fn every_spark_connect_role_maps_to_a_valid_role_name() { + for role in SparkConnectRole::iter() { + let _: RoleName = (&role).into(); + let _: RoleName = role.into(); + } + } +} diff --git a/rust/operator-binary/src/connect/controller.rs b/rust/operator-binary/src/connect/controller.rs index 70b67ec4..33835b5e 100644 --- a/rust/operator-binary/src/connect/controller.rs +++ b/rust/operator-binary/src/connect/controller.rs @@ -3,7 +3,6 @@ use std::sync::Arc; use snafu::{ResultExt, Snafu}; use stackable_operator::{ cluster_resources::ClusterResourceApplyStrategy, - commons::rbac::build_rbac_resources, crd::listener, k8s_openapi::api::{ apps::v1::StatefulSet, @@ -25,7 +24,7 @@ use stackable_operator::{ }; use strum::{EnumDiscriminants, IntoStaticStr}; -use super::crd::{CONNECT_APP_NAME, v1alpha1}; +use super::crd::v1alpha1; use crate::{Ctx, connect::crd::SparkConnectServerStatus, crd::constants::OPERATOR_NAME}; pub mod build; @@ -44,16 +43,6 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display("failed to apply role ServiceAccount"))] - ApplyServiceAccount { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to apply global RoleBinding"))] - ApplyRoleBinding { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to update status of spark connect server {name}"))] ApplyStatus { source: stackable_operator::client::Error, @@ -65,22 +54,11 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display("failed to get required Labels"))] - GetRequiredLabels { - source: - stackable_operator::kvp::KeyValuePairError, - }, - #[snafu(display("SparkConnectServer object is invalid"))] InvalidSparkConnectServer { source: error_boundary::InvalidObject, }, - #[snafu(display("failed to build RBAC resources"))] - BuildRbacResources { - source: stackable_operator::commons::rbac::Error, - }, - #[snafu(display("failed to dereference SparkConnectServer"))] DereferenceSparkConnectServer { source: dereference::Error }, @@ -150,20 +128,7 @@ pub async fn reconcile( &scs.spec.object_overrides, ); - // Use a dedicated service account for connect server pods. Building the RBAC resources needs - // the cluster-resource labels, so it stays in the reconcile step; the built objects (whose - // names are deterministic) are handed to the client-free build step. - let (service_account, role_binding) = build_rbac_resources( - scs, - CONNECT_APP_NAME, - cluster_resources - .get_required_labels() - .context(GetRequiredLabelsSnafu)?, - ) - .context(BuildRbacResourcesSnafu)?; - - let resources = build::build(&validated, service_account, role_binding, &scs.spec.args) - .context(BuildResourcesSnafu)?; + let resources = build::build(&validated, &scs.spec.args).context(BuildResourcesSnafu)?; // Apply order: ServiceAccount and RoleBinding first, then the Services, ConfigMaps and // Listener, and finally the StatefulSet (it mounts the ConfigMaps and runs under the SA, so @@ -171,11 +136,11 @@ pub async fn reconcile( cluster_resources .add(client, resources.service_account) .await - .context(ApplyServiceAccountSnafu)?; + .context(ApplyResourceSnafu)?; cluster_resources .add(client, resources.role_binding) .await - .context(ApplyRoleBindingSnafu)?; + .context(ApplyResourceSnafu)?; for service in resources.services { cluster_resources .add(client, service) diff --git a/rust/operator-binary/src/connect/controller/build/executor.rs b/rust/operator-binary/src/connect/controller/build/executor.rs index 94748334..d74c300d 100644 --- a/rust/operator-binary/src/connect/controller/build/executor.rs +++ b/rust/operator-binary/src/connect/controller/build/executor.rs @@ -30,7 +30,7 @@ use stackable_operator::{ use crate::{ connect::{ common::{self, SparkConnectRole, object_name}, - controller::validate::ValidatedSparkConnectServer, + controller::{build::object_meta, validate::ValidatedSparkConnectServer}, crd::{ CONNECT_EXECUTOR_ROLE_NAME, DEFAULT_SPARK_CONNECT_GROUP_NAME, SparkConnectContainer, v1alpha1, @@ -351,11 +351,7 @@ pub(crate) fn executor_config_map( let mut cm_builder = ConfigMapBuilder::new(); cm_builder - .metadata( - validated - .object_meta(&cm_name, SparkConnectRole::Executor) - .build(), - ) + .metadata(object_meta(validated, &cm_name, SparkConnectRole::Executor).build()) .add_data(JVM_SECURITY_PROPERTIES_FILE, jvm_sec_props) .add_data(METRICS_PROPERTIES_FILE, metrics_props); diff --git a/rust/operator-binary/src/connect/controller/build/mod.rs b/rust/operator-binary/src/connect/controller/build/mod.rs index da79b9b9..324403cc 100644 --- a/rust/operator-binary/src/connect/controller/build/mod.rs +++ b/rust/operator-binary/src/connect/controller/build/mod.rs @@ -6,18 +6,23 @@ //! together in one module is clearer than scattering them across per-kind modules. pub(crate) mod executor; +pub(crate) mod rbac; pub(crate) mod server; pub(crate) mod service; use snafu::{ResultExt, Snafu}; use stackable_operator::{ - k8s_openapi::api::{core::v1::ServiceAccount, rbac::v1::RoleBinding}, - kube::ResourceExt, + builder::meta::ObjectMetaBuilder, kube::ResourceExt, + v2::builder::meta::ownerreference_from_resource, }; use crate::connect::{ - common, - controller::{SparkConnectResources, validate::ValidatedSparkConnectServer}, + common::{self, SparkConnectRole}, + controller::{ + SparkConnectResources, + build::rbac::{build_role_binding, build_service_account}, + validate::ValidatedSparkConnectServer, + }, }; #[derive(Snafu, Debug)] @@ -56,8 +61,6 @@ pub enum Error { /// Builds every Kubernetes resource for the given validated SparkConnectServer. pub(crate) fn build( validated: &ValidatedSparkConnectServer, - service_account: ServiceAccount, - role_binding: RoleBinding, user_args: &[String], ) -> Result { let resolved_s3 = &validated.cluster_config.resolved_s3; @@ -70,8 +73,7 @@ pub(crate) fn build( resolved_s3 .spark_properties() .context(S3SparkPropertiesSnafu)?, - server::server_properties(validated, &headless_service, &service_account) - .context(ServerPropertiesSnafu)?, + server::server_properties(validated, &headless_service).context(ServerPropertiesSnafu)?, executor::executor_properties(validated).context(ExecutorPropertiesSnafu)?, ]) .context(SerializePropertiesSnafu)?; @@ -97,21 +99,83 @@ pub(crate) fn build( let listener = server::build_listener(validated); let args = server::command_args(user_args); - let stateful_set = server::build_stateful_set( - validated, - &service_account, - &server_config_map, - &listener.name_any(), - args, - ) - .context(BuildServerStatefulSetSnafu)?; + let stateful_set = + server::build_stateful_set(validated, &server_config_map, &listener.name_any(), args) + .context(BuildServerStatefulSetSnafu)?; Ok(SparkConnectResources { - service_account, - role_binding, + service_account: build_service_account(validated), + role_binding: build_role_binding(validated), services: vec![headless_service, metrics_service], config_maps: vec![executor_config_map, server_config_map], listener, stateful_set, }) } + +/// Object metadata for a child resource named `name`, owned by the SparkConnectServer and +/// carrying the recommended labels for the given role. +pub(crate) fn object_meta( + validated: &ValidatedSparkConnectServer, + name: impl Into, + role: SparkConnectRole, +) -> ObjectMetaBuilder { + let mut builder = ObjectMetaBuilder::new(); + builder + .name_and_namespace(validated) + .name(name) + .ownerreference(ownerreference_from_resource(validated, None, Some(true))) + .with_labels(validated.recommended_labels(role)); + builder +} + +#[cfg(test)] +pub(crate) mod test_support { + use indoc::indoc; + use stackable_operator::{cli::OperatorEnvironmentOptions, utils::yaml_from_str_singleton_map}; + + use crate::connect::{ + controller::{ + dereference::DereferencedSparkConnectServer, + validate::{ValidatedSparkConnectServer, validate}, + }, + crd::v1alpha1, + s3::ResolvedS3, + }; + + /// Minimal S3-free `SparkConnectServer` fixture, keeping the dereference step client-free; + /// the `uid` allows owner references to be derived from it. + /// + /// The cluster name (`my-connect`) deliberately differs from the product name + /// (`spark-connect`), so tests asserting recommended labels catch swapped `name`/`instance` + /// values. + pub const CONNECT_YAML: &str = indoc! {r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkConnectServer + metadata: + name: my-connect + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 4.1.2 + "#}; + + /// Runs the real validate step against the minimal fixture. + pub fn minimal_validated_cluster() -> ValidatedSparkConnectServer { + let scs: v1alpha1::SparkConnectServer = yaml_from_str_singleton_map(CONNECT_YAML) + .expect("invalid test SparkConnectServer YAML"); + validate( + &scs, + DereferencedSparkConnectServer { + resolved_s3: ResolvedS3::none(), + }, + &OperatorEnvironmentOptions { + operator_namespace: "stackable-operators".to_string(), + operator_service_name: "spark-k8s-operator".to_string(), + image_repository: "oci.example.org/sdp".to_string(), + }, + ) + .expect("validate should succeed for the test fixture") + } +} diff --git a/rust/operator-binary/src/connect/controller/build/rbac.rs b/rust/operator-binary/src/connect/controller/build/rbac.rs new file mode 100644 index 00000000..78ff5cd3 --- /dev/null +++ b/rust/operator-binary/src/connect/controller/build/rbac.rs @@ -0,0 +1,133 @@ +//! Builds the RBAC resources (ServiceAccount + RoleBinding) shared by all role groups. + +use std::str::FromStr; + +use stackable_operator::{ + k8s_openapi::api::{core::v1::ServiceAccount, rbac::v1::RoleBinding}, + kvp::Labels, + v2::{ + rbac, + types::operator::{RoleGroupName, RoleName}, + }, +}; + +use crate::connect::controller::validate::ValidatedSparkConnectServer; + +stackable_operator::constant!(NONE_ROLE_NAME: RoleName = "none"); +stackable_operator::constant!(NONE_ROLE_GROUP_NAME: RoleGroupName = "none"); + +pub fn build_service_account(server: &ValidatedSparkConnectServer) -> ServiceAccount { + rbac::build_service_account( + server, + &server.cluster_resource_names(), + rbac_labels(server), + ) +} + +pub fn build_role_binding(server: &ValidatedSparkConnectServer) -> RoleBinding { + rbac::build_role_binding( + server, + &server.cluster_resource_names(), + rbac_labels(server), + ) +} + +fn rbac_labels(server: &ValidatedSparkConnectServer) -> Labels { + server.recommended_labels_for(&NONE_ROLE_NAME, &NONE_ROLE_GROUP_NAME) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::{ + connect::controller::build::test_support::minimal_validated_cluster, + test_support::app_version_label, + }; + + // `my-connect` vs `spark-connect`: see the swap-guard note on `CONNECT_YAML`. + + #[test] + fn test_service_account() { + let service_account = build_service_account(&minimal_validated_cluster()); + + assert_eq!( + json!({ + "apiVersion": "v1", + "kind": "ServiceAccount", + "metadata": { + // The RBAC resources are cluster-shared, so role and role group are `none`. + "labels": { + "app.kubernetes.io/component": "none", + "app.kubernetes.io/instance": "my-connect", + "app.kubernetes.io/managed-by": "spark.stackable.tech_connect", + "app.kubernetes.io/name": "spark-connect", + "app.kubernetes.io/role-group": "none", + "app.kubernetes.io/version": app_version_label("4.1.2"), + "stackable.tech/vendor": "Stackable" + }, + "name": "my-connect-serviceaccount", + "namespace": "default", + "ownerReferences": [ + { + "apiVersion": "spark.stackable.tech/v1alpha1", + "controller": true, + "kind": "SparkConnectServer", + "name": "my-connect", + "uid": "12345678-1234-1234-1234-123456789012" + } + ] + } + }), + serde_json::to_value(service_account).expect("must be serializable") + ); + } + + #[test] + fn test_role_binding() { + let role_binding = build_role_binding(&minimal_validated_cluster()); + + assert_eq!( + json!({ + "apiVersion": "rbac.authorization.k8s.io/v1", + "kind": "RoleBinding", + "metadata": { + "labels": { + "app.kubernetes.io/component": "none", + "app.kubernetes.io/instance": "my-connect", + "app.kubernetes.io/managed-by": "spark.stackable.tech_connect", + "app.kubernetes.io/name": "spark-connect", + "app.kubernetes.io/role-group": "none", + "app.kubernetes.io/version": app_version_label("4.1.2"), + "stackable.tech/vendor": "Stackable" + }, + "name": "my-connect-rolebinding", + "namespace": "default", + "ownerReferences": [ + { + "apiVersion": "spark.stackable.tech/v1alpha1", + "controller": true, + "kind": "SparkConnectServer", + "name": "my-connect", + "uid": "12345678-1234-1234-1234-123456789012" + } + ] + }, + "roleRef": { + "apiGroup": "rbac.authorization.k8s.io", + "kind": "ClusterRole", + "name": "spark-connect-clusterrole" + }, + "subjects": [ + { + "kind": "ServiceAccount", + "name": "my-connect-serviceaccount", + "namespace": "default" + } + ] + }), + serde_json::to_value(role_binding).expect("must be serializable") + ); + } +} diff --git a/rust/operator-binary/src/connect/controller/build/server.rs b/rust/operator-binary/src/connect/controller/build/server.rs index 8ab09233..34ccfbbd 100644 --- a/rust/operator-binary/src/connect/controller/build/server.rs +++ b/rust/operator-binary/src/connect/controller/build/server.rs @@ -23,10 +23,7 @@ use stackable_operator::{ DeepMerge, api::{ apps::v1::{StatefulSet, StatefulSetSpec}, - core::v1::{ - ConfigMap, EnvVar, HTTPGetAction, PodSecurityContext, Probe, Service, - ServiceAccount, - }, + core::v1::{ConfigMap, EnvVar, HTTPGetAction, PodSecurityContext, Probe, Service}, }, apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, }, @@ -48,7 +45,7 @@ use crate::{ connect::{ GRPC, HTTP, common::{self, SparkConnectRole, object_name}, - controller::validate::ValidatedSparkConnectServer, + controller::{build::object_meta, validate::ValidatedSparkConnectServer}, crd::{ CONNECT_GRPC_PORT, CONNECT_SERVER_ROLE_NAME, CONNECT_UI_PORT, DEFAULT_SPARK_CONNECT_GROUP_NAME, SparkConnectContainer, v1alpha1, @@ -147,11 +144,7 @@ pub(crate) fn server_config_map( let mut cm_builder = ConfigMapBuilder::new(); cm_builder - .metadata( - validated - .object_meta(&cm_name, SparkConnectRole::Server) - .build(), - ) + .metadata(object_meta(validated, &cm_name, SparkConnectRole::Server).build()) .add_data(SPARK_DEFAULTS_FILE_NAME, spark_properties) .add_data(POD_TEMPLATE_FILE, executor_pod_template_spec) .add_data(JVM_SECURITY_PROPERTIES_FILE, jvm_sec_props) @@ -176,7 +169,6 @@ pub(crate) fn server_config_map( pub(crate) fn build_stateful_set( validated: &ValidatedSparkConnectServer, - service_account: &ServiceAccount, config_map: &ConfigMap, listener_name: &str, args: Vec, @@ -194,7 +186,7 @@ pub(crate) fn build_stateful_set( let mut pb = PodBuilder::new(); - pb.service_account_name(service_account.name_unchecked()) + pb.service_account_name(validated.cluster_resource_names().service_account_name()) .metadata(metadata) .image_pull_secrets_from_product_image(resolved_product_image) .add_volume( @@ -323,12 +315,12 @@ pub(crate) fn build_stateful_set( pod_template.merge_from(validated.server_overrides.pod_overrides.clone()); Ok(StatefulSet { - metadata: validated - .object_meta( - object_name(&validated.name_any(), SparkConnectRole::Server), - SparkConnectRole::Server, - ) - .build(), + metadata: object_meta( + validated, + object_name(&validated.name_any(), SparkConnectRole::Server), + SparkConnectRole::Server, + ) + .build(), spec: Some(StatefulSetSpec { template: pod_template, replicas: Some(1), @@ -402,13 +394,12 @@ fn env(env_overrides: Option<&HashMap>) -> Result, E pub(crate) fn server_properties( validated: &ValidatedSparkConnectServer, driver_service: &Service, - service_account: &ServiceAccount, ) -> Result>, Error> { let config = &validated.server_config; let resolved_product_image = &validated.resolved_product_image; let spark_image = resolved_product_image.image.clone(); let spark_version = resolved_product_image.product_version.clone(); - let service_account_name = service_account.name_unchecked(); + let service_account_name = validated.cluster_resource_names().service_account_name(); let namespace = driver_service .namespace() .context(ObjectHasNoNamespaceSnafu)?; @@ -434,7 +425,7 @@ pub(crate) fn server_properties( ("spark.kubernetes.namespace".to_string(), Some(namespace)), ( "spark.kubernetes.authenticate.driver.serviceAccountName".to_string(), - Some(service_account_name), + Some(service_account_name.to_string()), ), ( "spark.kubernetes.driver.pod.name".to_string(), diff --git a/rust/operator-binary/src/connect/controller/build/service.rs b/rust/operator-binary/src/connect/controller/build/service.rs index a7dc6663..da01a657 100644 --- a/rust/operator-binary/src/connect/controller/build/service.rs +++ b/rust/operator-binary/src/connect/controller/build/service.rs @@ -7,7 +7,7 @@ use stackable_operator::{ use crate::connect::{ GRPC, HTTP, common::SparkConnectRole, - controller::validate::ValidatedSparkConnectServer, + controller::{build::object_meta, validate::ValidatedSparkConnectServer}, crd::{CONNECT_GRPC_PORT, CONNECT_UI_PORT}, }; @@ -23,9 +23,7 @@ pub(crate) fn build_headless_service(validated: &ValidatedSparkConnectServer) -> let selector = validated.role_selector(SparkConnectRole::Server).into(); Service { - metadata: validated - .object_meta(service_name, SparkConnectRole::Server) - .build(), + metadata: object_meta(validated, service_name, SparkConnectRole::Server).build(), spec: Some(ServiceSpec { type_: Some("ClusterIP".to_owned()), cluster_ip: Some("None".to_owned()), @@ -63,8 +61,7 @@ pub(crate) fn build_metrics_service(validated: &ValidatedSparkConnectServer) -> let selector = validated.role_selector(SparkConnectRole::Server).into(); Service { - metadata: validated - .object_meta(service_name, SparkConnectRole::Server) + metadata: object_meta(validated, service_name, SparkConnectRole::Server) .with_labels(prometheus_labels(&Scraping::Enabled)) .with_annotations(prometheus_annotations( &Scraping::Enabled, diff --git a/rust/operator-binary/src/connect/controller/validate.rs b/rust/operator-binary/src/connect/controller/validate.rs index c1a4de67..feb83e09 100644 --- a/rust/operator-binary/src/connect/controller/validate.rs +++ b/rust/operator-binary/src/connect/controller/validate.rs @@ -7,7 +7,6 @@ use std::{borrow::Cow, collections::HashMap, str::FromStr}; use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ - builder::meta::ObjectMetaBuilder, cli::OperatorEnvironmentOptions, commons::product_image_selection::{self, ResolvedProductImage}, k8s_openapi::{api::core::v1::PodTemplateSpec, apimachinery::pkg::apis::meta::v1::ObjectMeta}, @@ -16,13 +15,12 @@ use stackable_operator::{ product_logging::spec::Logging, v2::{ HasName, HasUid, NameIsValidLabelValue, - builder::meta::ownerreference_from_resource, controller_utils::{get_cluster_name, get_namespace, get_uid}, kvp::label::{recommended_labels, role_group_selector, role_selector}, product_logging::framework::{ VectorContainerLogConfig, validate_logging_configuration_for_container, }, - role_utils::JavaCommonConfig, + role_utils::{self, JavaCommonConfig}, types::{ kubernetes::{ConfigMapName, ListenerClassName, NamespaceName, Uid}, operator::{ @@ -38,9 +36,8 @@ use crate::{ common::SparkConnectRole, controller::dereference::DereferencedSparkConnectServer, crd::{ - self, CONNECT_APP_NAME, CONNECT_CONTROLLER_NAME, CONNECT_EXECUTOR_ROLE_NAME, - CONNECT_SERVER_ROLE_NAME, DEFAULT_SPARK_CONNECT_GROUP_NAME, SparkConnectContainer, - v1alpha1, + self, CONNECT_APP_NAME, CONNECT_CONTROLLER_NAME, DEFAULT_SPARK_CONNECT_GROUP_NAME, + SparkConnectContainer, v1alpha1, }, s3::ResolvedS3, }, @@ -118,6 +115,10 @@ fn validate_logging( type Result = std::result::Result; +stackable_operator::constant!( + DEFAULT_SPARK_CONNECT_ROLE_GROUP: RoleGroupName = DEFAULT_SPARK_CONNECT_GROUP_NAME +); + /// Validated logging configuration for the (optional) Vector container. /// /// Produced up-front by [`validate_logging`] so that an @@ -134,6 +135,7 @@ pub struct ValidatedSparkConnectServer { pub name: ClusterName, pub namespace: NamespaceName, pub uid: Uid, + pub product_version: ProductVersion, pub resolved_product_image: ResolvedProductImage, pub cluster_config: ValidatedClusterConfig, pub role_config: ValidatedRoleConfig, @@ -167,81 +169,79 @@ pub struct ValidatedRoleConfig { } impl ValidatedSparkConnectServer { + /// Recommended labels for a resource that is not tied to a concrete [`SparkConnectRole`] + /// (e.g. the cluster-shared RBAC resources), using a free-form role/role-group label value. + pub fn recommended_labels_for( + &self, + role_name: &RoleName, + role_group_name: &RoleGroupName, + ) -> Labels { + self.recommended_labels_with(&self.product_version, role_name, role_group_name) + } + /// Recommended labels for a resource of the given role. - pub(crate) fn recommended_labels(&self, role: SparkConnectRole) -> Labels { - // `app_version_label_value` is constructed to be a valid label value, so it is also a - // valid `ProductVersion`. - let product_version = - ProductVersion::from_str(&self.resolved_product_image.app_version_label_value) - .expect("the app version label value is a valid product version"); - let role_group = RoleGroupName::from_str(DEFAULT_SPARK_CONNECT_GROUP_NAME) - .expect("DEFAULT_SPARK_CONNECT_GROUP_NAME is a valid role group name"); + pub fn recommended_labels(&self, role: SparkConnectRole) -> Labels { + self.recommended_labels_for(&role.into(), &DEFAULT_SPARK_CONNECT_ROLE_GROUP) + } + + fn recommended_labels_with( + &self, + product_version: &ProductVersion, + role_name: &RoleName, + role_group_name: &RoleGroupName, + ) -> Labels { recommended_labels( self, &product_name(), - &product_version, + product_version, &operator_name(), &controller_name(), - &role_name(role), - &role_group, + role_name, + role_group_name, ) } /// Selector labels matching the pods of the given role. - pub(crate) fn role_selector(&self, role: SparkConnectRole) -> Labels { - role_selector(self, &product_name(), &role_name(role)) + pub fn role_selector(&self, role: SparkConnectRole) -> Labels { + role_selector(self, &product_name(), &role.into()) } /// Selector labels matching the pods of the given role's (single) role group. - pub(crate) fn role_group_selector(&self, role: SparkConnectRole) -> Labels { - let role_group = RoleGroupName::from_str(DEFAULT_SPARK_CONNECT_GROUP_NAME) - .expect("DEFAULT_SPARK_CONNECT_GROUP_NAME is a valid role group name"); - role_group_selector(self, &product_name(), &role_name(role), &role_group) + pub fn role_group_selector(&self, role: SparkConnectRole) -> Labels { + role_group_selector( + self, + &product_name(), + &role.into(), + &DEFAULT_SPARK_CONNECT_ROLE_GROUP, + ) } - /// Object metadata for a child resource named `name`, owned by this SparkConnectServer and - /// carrying the recommended labels for the given role. - pub(crate) fn object_meta( - &self, - name: impl Into, - role: SparkConnectRole, - ) -> ObjectMetaBuilder { - let mut builder = ObjectMetaBuilder::new(); - builder - .name_and_namespace(self) - .name(name) - .ownerreference(ownerreference_from_resource(self, None, Some(true))) - .with_labels(self.recommended_labels(role)); - builder + /// Type-safe names for the per-cluster RBAC resources: the ServiceAccount, + /// its (namespaced) RoleBinding, and the operator-deployed ClusterRole it binds. + pub fn cluster_resource_names(&self) -> role_utils::ResourceNames { + role_utils::ResourceNames { + cluster_name: self.name.clone(), + product_name: product_name(), + } } } /// The product name (`spark-connect`) as a type-safe label value. -pub(crate) fn product_name() -> ProductName { +pub fn product_name() -> ProductName { ProductName::from_str(CONNECT_APP_NAME).expect("CONNECT_APP_NAME is a valid product name") } /// The operator name as a type-safe label value. -pub(crate) fn operator_name() -> OperatorName { +pub fn operator_name() -> OperatorName { OperatorName::from_str(OPERATOR_NAME).expect("the operator name is a valid label value") } /// The controller name as a type-safe label value. -pub(crate) fn controller_name() -> ControllerName { +pub fn controller_name() -> ControllerName { ControllerName::from_str(CONNECT_CONTROLLER_NAME) .expect("the controller name is a valid label value") } -/// The role name for the given Spark Connect role as a type-safe label value. -fn role_name(role: SparkConnectRole) -> RoleName { - match role { - SparkConnectRole::Server => RoleName::from_str(CONNECT_SERVER_ROLE_NAME) - .expect("CONNECT_SERVER_ROLE_NAME is a valid role name"), - SparkConnectRole::Executor => RoleName::from_str(CONNECT_EXECUTOR_ROLE_NAME) - .expect("CONNECT_EXECUTOR_ROLE_NAME is a valid role name"), - } -} - impl NameIsValidLabelValue for ValidatedSparkConnectServer { fn to_label_value(&self) -> String { self.name.to_label_value() @@ -304,6 +304,9 @@ pub fn validate( ) .context(ResolveProductImageSnafu)?; + let product_version = ProductVersion::from_str(&resolved_product_image.app_version_label_value) + .expect("the app version label value is a valid product version"); + let server_config = scs.server_config().context(ServerConfigSnafu)?; let executor_config = scs.executor_config().context(ExecutorConfigSnafu)?; @@ -350,6 +353,7 @@ pub fn validate( name, namespace, uid, + product_version, resolved_product_image, cluster_config: ValidatedClusterConfig { resolved_s3: dereferenced.resolved_s3, @@ -365,3 +369,54 @@ pub fn validate( executor_logging, }) } + +#[cfg(test)] +mod tests { + use crate::{ + connect::controller::build::test_support::minimal_validated_cluster, + test_support::app_version_label, + }; + + /// Locks every value the validate step itself derives from the minimal fixture — so a + /// validation regression fails here, with a validate-shaped message, instead of surfacing as + /// a confusing build-test failure downstream. + #[test] + fn validate_ok_derives_expected_values() { + let validated = minimal_validated_cluster(); + + assert_eq!(validated.name.to_string(), "my-connect"); + assert_eq!(validated.namespace.to_string(), "default"); + assert_eq!( + validated.uid.to_string(), + "12345678-1234-1234-1234-123456789012" + ); + assert_eq!( + validated.resolved_product_image.image, + format!( + "oci.example.org/sdp/spark-k8s:{}", + app_version_label("4.1.2") + ) + ); + assert_eq!(validated.resolved_product_image.product_version, "4.1.2"); + assert_eq!( + validated.product_version.to_string(), + app_version_label("4.1.2") + ); + + // The role config falls back to the default cluster-internal listener. + assert_eq!( + validated.role_config.listener_class.to_string(), + "cluster-internal" + ); + + // The minimal fixture has no overrides and no Vector agent for either role. + for overrides in [&validated.server_overrides, &validated.executor_overrides] { + assert!(overrides.env_overrides.is_empty()); + assert_eq!(overrides.jvm_config, None); + } + for logging in [&validated.server_logging, &validated.executor_logging] { + assert!(!logging.enable_vector_agent); + assert_eq!(logging.vector_container, None); + } + } +} diff --git a/rust/operator-binary/src/connect/s3.rs b/rust/operator-binary/src/connect/s3.rs index 624720dc..0e46f89f 100644 --- a/rust/operator-binary/src/connect/s3.rs +++ b/rust/operator-binary/src/connect/s3.rs @@ -54,6 +54,16 @@ pub(crate) struct ResolvedS3 { } impl ResolvedS3 { + /// The resolved form of a SparkConnectServer without any S3 connectors, for tests that must + /// stay client-free. + #[cfg(test)] + pub(crate) fn none() -> Self { + Self { + s3_buckets: Vec::new(), + s3_connection: None, + } + } + pub(crate) async fn resolve( client: &stackable_operator::client::Client, connect_server: &crd::v1alpha1::SparkConnectServer, diff --git a/rust/operator-binary/src/history/controller.rs b/rust/operator-binary/src/history/controller.rs index 250e8d30..bc7f46de 100644 --- a/rust/operator-binary/src/history/controller.rs +++ b/rust/operator-binary/src/history/controller.rs @@ -3,7 +3,6 @@ use std::sync::Arc; use snafu::{ResultExt, Snafu}; use stackable_operator::{ cluster_resources::ClusterResourceApplyStrategy, - commons::rbac::build_rbac_resources, crd::listener, k8s_openapi::api::{ apps::v1::StatefulSet, @@ -21,10 +20,7 @@ use stackable_operator::{ }; use strum::{EnumDiscriminants, IntoStaticStr}; -use crate::{ - Ctx, - crd::{constants::HISTORY_APP_NAME, history::v1alpha1}, -}; +use crate::{Ctx, crd::history::v1alpha1}; pub mod build; pub mod dereference; @@ -34,11 +30,6 @@ pub mod validate; #[strum_discriminants(derive(IntoStaticStr))] #[allow(clippy::enum_variant_names)] pub enum Error { - #[snafu(display("failed to build RBAC resources"))] - BuildRbacResources { - source: stackable_operator::commons::rbac::Error, - }, - #[snafu(display("failed to build SparkHistoryServer resources"))] BuildSparkHistoryServer { source: build::Error }, @@ -47,16 +38,6 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display("failed to apply role ServiceAccount"))] - ApplyServiceAccount { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to apply global RoleBinding"))] - ApplyRoleBinding { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to dereference SparkHistoryServer"))] DereferenceSparkHistoryServer { source: dereference::Error }, @@ -68,12 +49,6 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display("failed to get required Labels"))] - GetRequiredLabels { - source: - stackable_operator::kvp::KeyValuePairError, - }, - #[snafu(display("SparkHistoryServer object is invalid"))] InvalidSparkHistoryServer { // boxed because otherwise Clippy warns about a large enum variant @@ -135,20 +110,7 @@ pub async fn reconcile( &shs.spec.object_overrides, ); - // Use a dedicated service account for history server pods. Building the RBAC resources needs - // the cluster-resource labels, so it stays in the reconcile step; the built objects (whose - // names are deterministic) are handed to the client-free build step. - let (service_account, role_binding) = build_rbac_resources( - shs, - HISTORY_APP_NAME, - cluster_resources - .get_required_labels() - .context(GetRequiredLabelsSnafu)?, - ) - .context(BuildRbacResourcesSnafu)?; - - let resources = build::build(&validated, service_account, role_binding) - .context(BuildSparkHistoryServerSnafu)?; + let resources = build::build(&validated).context(BuildSparkHistoryServerSnafu)?; // Apply order: ServiceAccount and RoleBinding first, then the ConfigMaps, metrics Services, // Listener and PodDisruptionBudget, and finally the StatefulSets (they mount the ConfigMaps @@ -156,11 +118,11 @@ pub async fn reconcile( cluster_resources .add(client, resources.service_account) .await - .context(ApplyServiceAccountSnafu)?; + .context(ApplyResourceSnafu)?; cluster_resources .add(client, resources.role_binding) .await - .context(ApplyRoleBindingSnafu)?; + .context(ApplyResourceSnafu)?; for config_map in resources.config_maps { cluster_resources .add(client, config_map) diff --git a/rust/operator-binary/src/history/controller/build/mod.rs b/rust/operator-binary/src/history/controller/build/mod.rs index e23d2609..1d3bf550 100644 --- a/rust/operator-binary/src/history/controller/build/mod.rs +++ b/rust/operator-binary/src/history/controller/build/mod.rs @@ -1,7 +1,10 @@ pub mod resource; use snafu::{ResultExt, Snafu}; -use stackable_operator::k8s_openapi::api::{core::v1::ServiceAccount, rbac::v1::RoleBinding}; +use stackable_operator::{ + builder::meta::ObjectMetaBuilder, + v2::{builder::meta::ownerreference_from_resource, types::operator::RoleGroupName}, +}; use crate::{ crd::constants::HISTORY_ROLE_NAME, @@ -11,6 +14,7 @@ use crate::{ config_map::{self, build_config_map}, listener::build_group_listener, pdb::build_pdb, + rbac::{build_role_binding, build_service_account}, service::build_rolegroup_metrics_service, statefulset::{self, build_stateful_set}, }, @@ -30,11 +34,7 @@ pub enum Error { type Result = std::result::Result; /// Builds every Kubernetes resource for the given validated SparkHistoryServer. -pub fn build( - validated: &ValidatedSparkHistoryServer, - service_account: ServiceAccount, - role_binding: RoleBinding, -) -> Result { +pub fn build(validated: &ValidatedSparkHistoryServer) -> Result { let log_dir = &validated.cluster_config.log_dir; let mut config_maps = vec![]; @@ -46,7 +46,7 @@ pub fn build( .push(build_config_map(validated, role_group_name, rg).context(BuildConfigMapSnafu)?); metrics_services.push(build_rolegroup_metrics_service(validated, role_group_name)); stateful_sets.push( - build_stateful_set(validated, role_group_name, rg, log_dir, &service_account) + build_stateful_set(validated, role_group_name, rg, log_dir) .context(BuildStatefulSetSnafu)?, ); } @@ -60,8 +60,8 @@ pub fn build( let pod_disruption_budget = build_pdb(&validated.role_config.pdb, validated); Ok(SparkHistoryResources { - service_account, - role_binding, + service_account: build_service_account(validated), + role_binding: build_role_binding(validated), config_maps, metrics_services, stateful_sets, @@ -69,3 +69,76 @@ pub fn build( pod_disruption_budget, }) } + +/// Object metadata for a child resource named `name`, owned by the SparkHistoryServer and +/// carrying the recommended labels for the given role group. Returns the builder so callers can +/// add extra labels (e.g. Prometheus annotations) before building. +pub(crate) fn object_meta( + validated: &ValidatedSparkHistoryServer, + name: impl Into, + role_group_name: &RoleGroupName, +) -> ObjectMetaBuilder { + let mut builder = ObjectMetaBuilder::new(); + builder + .name_and_namespace(validated) + .name(name) + .ownerreference(ownerreference_from_resource(validated, None, Some(true))) + .with_labels(validated.recommended_labels(role_group_name)); + builder +} + +#[cfg(test)] +pub(crate) mod test_support { + use indoc::indoc; + use stackable_operator::{cli::OperatorEnvironmentOptions, utils::yaml_from_str_singleton_map}; + + use crate::{ + crd::{history::v1alpha1, logdir::ResolvedLogDir}, + history::controller::{ + dereference::DereferencedSparkHistoryServer, + validate::{ValidatedSparkHistoryServer, validate}, + }, + }; + + /// Minimal custom-log-dir `SparkHistoryServer` fixture. The custom log directory keeps the + /// dereference step client-free; the `uid` allows owner references to be derived from it. + /// + /// The cluster name (`my-history`) deliberately differs from the product name + /// (`spark-history`), so tests asserting recommended labels catch swapped `name`/`instance` + /// values. + pub const HISTORY_YAML: &str = indoc! {r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkHistoryServer + metadata: + name: my-history + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.5.8 + logFileDirectory: + customLogDirectory: file:///stackable/spark/logs + nodes: + roleGroups: + default: + replicas: 1 + "#}; + + /// Runs the real validate step against the minimal fixture. + pub fn minimal_validated_cluster() -> ValidatedSparkHistoryServer { + let shs: v1alpha1::SparkHistoryServer = yaml_from_str_singleton_map(HISTORY_YAML) + .expect("invalid test SparkHistoryServer YAML"); + validate( + &shs, + DereferencedSparkHistoryServer { + log_dir: ResolvedLogDir::Custom("file:///stackable/spark/logs".to_string()), + }, + &OperatorEnvironmentOptions { + operator_namespace: "stackable-operators".to_string(), + operator_service_name: "spark-k8s-operator".to_string(), + image_repository: "oci.example.org/sdp".to_string(), + }, + ) + .expect("validate should succeed for the test fixture") + } +} diff --git a/rust/operator-binary/src/history/controller/build/resource/config_map.rs b/rust/operator-binary/src/history/controller/build/resource/config_map.rs index c19f18b2..cdfbb518 100644 --- a/rust/operator-binary/src/history/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/history/controller/build/resource/config_map.rs @@ -55,7 +55,7 @@ pub(crate) fn build_config_map( rg: &ValidatedHistoryRoleGroup, ) -> Result { let cm_name = validated - .resource_names(role_group_name) + .role_group_resource_names(role_group_name) .role_group_config_map() .to_string(); diff --git a/rust/operator-binary/src/history/controller/build/resource/mod.rs b/rust/operator-binary/src/history/controller/build/resource/mod.rs index 2923f3a2..a15a440e 100644 --- a/rust/operator-binary/src/history/controller/build/resource/mod.rs +++ b/rust/operator-binary/src/history/controller/build/resource/mod.rs @@ -1,5 +1,6 @@ pub mod config_map; pub mod listener; pub mod pdb; +pub mod rbac; pub mod service; pub mod statefulset; diff --git a/rust/operator-binary/src/history/controller/build/resource/rbac.rs b/rust/operator-binary/src/history/controller/build/resource/rbac.rs new file mode 100644 index 00000000..c252bfa8 --- /dev/null +++ b/rust/operator-binary/src/history/controller/build/resource/rbac.rs @@ -0,0 +1,133 @@ +//! Builds the RBAC resources (ServiceAccount + RoleBinding) shared by all role groups. + +use std::str::FromStr; + +use stackable_operator::{ + k8s_openapi::api::{core::v1::ServiceAccount, rbac::v1::RoleBinding}, + kvp::Labels, + v2::{ + rbac, + types::operator::{RoleGroupName, RoleName}, + }, +}; + +use crate::history::controller::validate::ValidatedSparkHistoryServer; + +stackable_operator::constant!(NONE_ROLE_NAME: RoleName = "none"); +stackable_operator::constant!(NONE_ROLE_GROUP_NAME: RoleGroupName = "none"); + +pub fn build_service_account(server: &ValidatedSparkHistoryServer) -> ServiceAccount { + rbac::build_service_account( + server, + &server.cluster_resource_names(), + rbac_labels(server), + ) +} + +pub fn build_role_binding(server: &ValidatedSparkHistoryServer) -> RoleBinding { + rbac::build_role_binding( + server, + &server.cluster_resource_names(), + rbac_labels(server), + ) +} + +fn rbac_labels(server: &ValidatedSparkHistoryServer) -> Labels { + server.recommended_labels_for(&NONE_ROLE_NAME, &NONE_ROLE_GROUP_NAME) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::{ + history::controller::build::test_support::minimal_validated_cluster, + test_support::app_version_label, + }; + + // `my-history` vs `spark-history`: see the swap-guard note on `HISTORY_YAML`. + + #[test] + fn test_service_account() { + let service_account = build_service_account(&minimal_validated_cluster()); + + assert_eq!( + json!({ + "apiVersion": "v1", + "kind": "ServiceAccount", + "metadata": { + // The RBAC resources are cluster-shared, so role and role group are `none`. + "labels": { + "app.kubernetes.io/component": "none", + "app.kubernetes.io/instance": "my-history", + "app.kubernetes.io/managed-by": "spark.stackable.tech_history", + "app.kubernetes.io/name": "spark-history", + "app.kubernetes.io/role-group": "none", + "app.kubernetes.io/version": app_version_label("3.5.8"), + "stackable.tech/vendor": "Stackable" + }, + "name": "my-history-serviceaccount", + "namespace": "default", + "ownerReferences": [ + { + "apiVersion": "spark.stackable.tech/v1alpha1", + "controller": true, + "kind": "SparkHistoryServer", + "name": "my-history", + "uid": "12345678-1234-1234-1234-123456789012" + } + ] + } + }), + serde_json::to_value(service_account).expect("must be serializable") + ); + } + + #[test] + fn test_role_binding() { + let role_binding = build_role_binding(&minimal_validated_cluster()); + + assert_eq!( + json!({ + "apiVersion": "rbac.authorization.k8s.io/v1", + "kind": "RoleBinding", + "metadata": { + "labels": { + "app.kubernetes.io/component": "none", + "app.kubernetes.io/instance": "my-history", + "app.kubernetes.io/managed-by": "spark.stackable.tech_history", + "app.kubernetes.io/name": "spark-history", + "app.kubernetes.io/role-group": "none", + "app.kubernetes.io/version": app_version_label("3.5.8"), + "stackable.tech/vendor": "Stackable" + }, + "name": "my-history-rolebinding", + "namespace": "default", + "ownerReferences": [ + { + "apiVersion": "spark.stackable.tech/v1alpha1", + "controller": true, + "kind": "SparkHistoryServer", + "name": "my-history", + "uid": "12345678-1234-1234-1234-123456789012" + } + ] + }, + "roleRef": { + "apiGroup": "rbac.authorization.k8s.io", + "kind": "ClusterRole", + "name": "spark-history-clusterrole" + }, + "subjects": [ + { + "kind": "ServiceAccount", + "name": "my-history-serviceaccount", + "namespace": "default" + } + ] + }), + serde_json::to_value(role_binding).expect("must be serializable") + ); + } +} diff --git a/rust/operator-binary/src/history/controller/build/resource/service.rs b/rust/operator-binary/src/history/controller/build/resource/service.rs index f1773fff..ddab39a9 100644 --- a/rust/operator-binary/src/history/controller/build/resource/service.rs +++ b/rust/operator-binary/src/history/controller/build/resource/service.rs @@ -7,7 +7,8 @@ use stackable_operator::{ }; use crate::{ - crd::constants::METRICS_PORT, history::controller::validate::ValidatedSparkHistoryServer, + crd::constants::METRICS_PORT, + history::controller::{build::object_meta, validate::ValidatedSparkHistoryServer}, }; /// The rolegroup metrics [`Service`] is a service that exposes metrics and a prometheus scraping label @@ -16,22 +17,22 @@ pub fn build_rolegroup_metrics_service( role_group_name: &RoleGroupName, ) -> Service { Service { - metadata: validated - .object_meta( - validated - .resource_names(role_group_name) - .metrics_service_name() - .to_string(), - role_group_name, - ) - .with_labels(prometheus_labels(&Scraping::Enabled)) - .with_annotations(prometheus_annotations( - &Scraping::Enabled, - &Scheme::Http, - "/metrics", - &METRICS_PORT, - )) - .build(), + metadata: object_meta( + validated, + validated + .role_group_resource_names(role_group_name) + .metrics_service_name() + .to_string(), + role_group_name, + ) + .with_labels(prometheus_labels(&Scraping::Enabled)) + .with_annotations(prometheus_annotations( + &Scraping::Enabled, + &Scheme::Http, + "/metrics", + &METRICS_PORT, + )) + .build(), spec: Some(ServiceSpec { // Internal communication does not need to be exposed type_: Some("ClusterIP".to_string()), @@ -53,3 +54,81 @@ fn metrics_ports() -> Vec { ..ServicePort::default() }] } + +#[cfg(test)] +mod tests { + use serde_json::json; + use stackable_operator::v2::types::operator::RoleGroupName; + + use super::*; + use crate::{ + history::controller::build::test_support::minimal_validated_cluster, + test_support::app_version_label, + }; + + /// Every metrics Service must carry the Prometheus scrape label and the + /// `prometheus.io/path|port|scheme|scrape` annotations, or Prometheus stops discovering the + /// endpoints. + #[test] + fn test_rolegroup_metrics_service() { + let validated = minimal_validated_cluster(); + let role_group_name: RoleGroupName = "default".parse().expect("valid role group name"); + + let service = build_rolegroup_metrics_service(&validated, &role_group_name); + + assert_eq!( + json!({ + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "annotations": { + "prometheus.io/path": "/metrics", + "prometheus.io/port": "18081", + "prometheus.io/scheme": "http", + "prometheus.io/scrape": "true" + }, + "labels": { + "app.kubernetes.io/component": "node", + "app.kubernetes.io/instance": "my-history", + "app.kubernetes.io/managed-by": "spark.stackable.tech_history", + "app.kubernetes.io/name": "spark-history", + "app.kubernetes.io/role-group": "default", + "app.kubernetes.io/version": app_version_label("3.5.8"), + "prometheus.io/scrape": "true", + "stackable.tech/vendor": "Stackable" + }, + "name": "my-history-node-default-metrics", + "namespace": "default", + "ownerReferences": [ + { + "apiVersion": "spark.stackable.tech/v1alpha1", + "controller": true, + "kind": "SparkHistoryServer", + "name": "my-history", + "uid": "12345678-1234-1234-1234-123456789012" + } + ] + }, + "spec": { + "clusterIP": "None", + "ports": [ + { + "name": "metrics", + "port": 18081, + "protocol": "TCP" + } + ], + "publishNotReadyAddresses": true, + "selector": { + "app.kubernetes.io/component": "node", + "app.kubernetes.io/instance": "my-history", + "app.kubernetes.io/name": "spark-history", + "app.kubernetes.io/role-group": "default" + }, + "type": "ClusterIP" + } + }), + serde_json::to_value(service).expect("must be serializable") + ); + } +} diff --git a/rust/operator-binary/src/history/controller/build/resource/statefulset.rs b/rust/operator-binary/src/history/controller/build/resource/statefulset.rs index 8ef1f5b2..00d9d512 100644 --- a/rust/operator-binary/src/history/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/history/controller/build/resource/statefulset.rs @@ -10,11 +10,10 @@ use stackable_operator::{ DeepMerge, api::{ apps::v1::{StatefulSet, StatefulSetSpec}, - core::v1::{PodSecurityContext, ServiceAccount}, + core::v1::PodSecurityContext, }, apimachinery::pkg::apis::meta::v1::LabelSelector, }, - kube::ResourceExt, product_logging::{ framework::calculate_log_volume_size_limit, spec::{ @@ -52,7 +51,7 @@ use crate::{ history::{ config::jvm::construct_history_jvm_args, controller::{ - build::resource::listener::group_listener_name, + build::{object_meta, resource::listener::group_listener_name}, validate::{self, ValidatedHistoryRoleGroup}, }, }, @@ -84,10 +83,9 @@ pub(crate) fn build_stateful_set( role_group_name: &RoleGroupName, rg: &ValidatedHistoryRoleGroup, log_dir: &ResolvedLogDir, - serviceaccount: &ServiceAccount, ) -> Result { let resolved_product_image = &validated.resolved_product_image; - let resource_names = validated.resource_names(role_group_name); + let resource_names = validated.role_group_resource_names(role_group_name); let log_config_map = if let Some(ContainerLogConfig { choice: @@ -119,40 +117,45 @@ pub(crate) fn build_stateful_set( .config .requested_secret_lifetime .context(MissingSecretLifetimeSnafu)?; - pb.service_account_name(serviceaccount.name_unchecked()) - .metadata(pb_metadata) - .image_pull_secrets_from_product_image(resolved_product_image) - .add_volume( - VolumeBuilder::new(VOLUME_MOUNT_NAME_CONFIG.as_ref()) - .with_config_map(resource_names.role_group_config_map().to_string()) - .build(), - ) - .context(AddVolumeSnafu)? - .add_volume( - VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG_CONFIG.as_ref()) - .with_config_map(log_config_map) - .build(), - ) - .context(AddVolumeSnafu)? - .add_volume( - VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG.as_ref()) - .with_empty_dir( - None::, - Some(calculate_log_volume_size_limit(&[MAX_SPARK_LOG_FILES_SIZE])), - ) - .build(), - ) - .context(AddVolumeSnafu)? - .add_volumes( - log_dir - .volumes(&requested_secret_lifetime) - .context(CreateLogDirVolumesSpecSnafu)?, - ) - .context(AddVolumeSnafu)? - .security_context(PodSecurityContext { - fs_group: Some(1000), - ..PodSecurityContext::default() - }); + pb.service_account_name( + validated + .cluster_resource_names() + .service_account_name() + .to_string(), + ) + .metadata(pb_metadata) + .image_pull_secrets_from_product_image(resolved_product_image) + .add_volume( + VolumeBuilder::new(VOLUME_MOUNT_NAME_CONFIG.as_ref()) + .with_config_map(resource_names.role_group_config_map().to_string()) + .build(), + ) + .context(AddVolumeSnafu)? + .add_volume( + VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG_CONFIG.as_ref()) + .with_config_map(log_config_map) + .build(), + ) + .context(AddVolumeSnafu)? + .add_volume( + VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG.as_ref()) + .with_empty_dir( + None::, + Some(calculate_log_volume_size_limit(&[MAX_SPARK_LOG_FILES_SIZE])), + ) + .build(), + ) + .context(AddVolumeSnafu)? + .add_volumes( + log_dir + .volumes(&requested_secret_lifetime) + .context(CreateLogDirVolumesSpecSnafu)?, + ) + .context(AddVolumeSnafu)? + .security_context(PodSecurityContext { + fs_group: Some(1000), + ..PodSecurityContext::default() + }); // Base environment variables, with the already-merged (role + role group) env overrides // layered on top (overrides win). The base names are static and known to be valid. @@ -244,12 +247,12 @@ pub(crate) fn build_stateful_set( let mut pod_template = pb.build_template(); pod_template.merge_from(rg.config.pod_overrides.clone()); - let sts_metadata = validated - .object_meta( - resource_names.stateful_set_name().to_string(), - role_group_name, - ) - .build(); + let sts_metadata = object_meta( + validated, + resource_names.stateful_set_name().to_string(), + role_group_name, + ) + .build(); Ok(StatefulSet { metadata: sts_metadata, diff --git a/rust/operator-binary/src/history/controller/validate.rs b/rust/operator-binary/src/history/controller/validate.rs index 83f6aedc..b0aaaab9 100644 --- a/rust/operator-binary/src/history/controller/validate.rs +++ b/rust/operator-binary/src/history/controller/validate.rs @@ -7,7 +7,6 @@ use std::{borrow::Cow, collections::BTreeMap, str::FromStr}; use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ - builder::meta::ObjectMetaBuilder, cli::OperatorEnvironmentOptions, commons::{ pdb::PdbConfig, @@ -20,17 +19,14 @@ use stackable_operator::{ product_logging::spec::Logging, v2::{ HasName, HasUid, NameIsValidLabelValue, - builder::{ - meta::ownerreference_from_resource, - pod::container::{EnvVarName, EnvVarSet}, - }, + builder::pod::container::{EnvVarName, EnvVarSet}, controller_utils::{get_cluster_name, get_namespace, get_uid}, kvp::label::{recommended_labels, role_group_selector}, product_logging::framework::{ VectorContainerLogConfig, validate_logging_configuration_for_container, }, role_group_utils::ResourceNames, - role_utils::{JavaCommonConfig, RoleGroupConfig, with_validated_config}, + role_utils::{self, JavaCommonConfig, RoleGroupConfig, with_validated_config}, types::{ kubernetes::{ConfigMapName, ListenerClassName, NamespaceName, Uid}, operator::{ @@ -202,8 +198,17 @@ impl ValidatedSparkHistoryServer { RoleName::from_str(HISTORY_ROLE_NAME).expect("HISTORY_ROLE_NAME is a valid role name") } + /// Type-safe names for the per-cluster RBAC resources: the ServiceAccount, + /// its (namespaced) RoleBinding, and the operator-deployed ClusterRole it binds. + pub fn cluster_resource_names(&self) -> role_utils::ResourceNames { + role_utils::ResourceNames { + cluster_name: self.name.clone(), + product_name: product_name(), + } + } + /// Type-safe names for the resources of a given role group. - pub(crate) fn resource_names(&self, role_group_name: &RoleGroupName) -> ResourceNames { + pub fn role_group_resource_names(&self, role_group_name: &RoleGroupName) -> ResourceNames { ResourceNames { cluster_name: self.name.clone(), role_name: Self::role_name(), @@ -211,10 +216,25 @@ impl ValidatedSparkHistoryServer { } } - /// Recommended labels for a role-group resource, using the given product version. - fn recommended_labels_for( + /// Recommended labels for a resource of the given role. + pub fn recommended_labels(&self, role_group_name: &RoleGroupName) -> Labels { + self.recommended_labels_for(&Self::role_name(), role_group_name) + } + + /// Recommended labels for a resource that is not tied to a concrete role + /// (e.g. the cluster-shared RBAC resources), using a free-form role/role-group label value. + pub fn recommended_labels_for( + &self, + role_name: &RoleName, + role_group_name: &RoleGroupName, + ) -> Labels { + self.recommended_labels_with(&self.product_version, role_name, role_group_name) + } + + fn recommended_labels_with( &self, product_version: &ProductVersion, + role_name: &RoleName, role_group_name: &RoleGroupName, ) -> Labels { recommended_labels( @@ -223,51 +243,29 @@ impl ValidatedSparkHistoryServer { product_version, &operator_name(), &controller_name(), - &Self::role_name(), + role_name, role_group_name, ) } - /// Recommended labels for a role-group resource. - pub fn recommended_labels(&self, role_group_name: &RoleGroupName) -> Labels { - self.recommended_labels_for(&self.product_version, role_group_name) - } - /// Selector labels matching the pods of a role group. pub fn role_group_selector(&self, role_group_name: &RoleGroupName) -> Labels { role_group_selector(self, &product_name(), &Self::role_name(), role_group_name) } - - /// Object metadata for a child resource named `name`, owned by this SparkHistoryServer and - /// carrying the recommended labels for the given role group. Returns the builder so callers can - /// add extra labels (e.g. Prometheus annotations) before building. - pub(crate) fn object_meta( - &self, - name: impl Into, - role_group_name: &RoleGroupName, - ) -> ObjectMetaBuilder { - let mut builder = ObjectMetaBuilder::new(); - builder - .name_and_namespace(self) - .name(name) - .ownerreference(ownerreference_from_resource(self, None, Some(true))) - .with_labels(self.recommended_labels(role_group_name)); - builder - } } /// The product name (`spark-history`) as a type-safe label value. -pub(crate) fn product_name() -> ProductName { +pub fn product_name() -> ProductName { ProductName::from_str(HISTORY_APP_NAME).expect("HISTORY_APP_NAME is a valid product name") } /// The operator name as a type-safe label value. -pub(crate) fn operator_name() -> OperatorName { +pub fn operator_name() -> OperatorName { OperatorName::from_str(OPERATOR_NAME).expect("the operator name is a valid label value") } /// The controller name as a type-safe label value. -pub(crate) fn controller_name() -> ControllerName { +pub fn controller_name() -> ControllerName { ControllerName::from_str(HISTORY_CONTROLLER_NAME) .expect("the controller name is a valid label value") } @@ -436,3 +434,84 @@ pub fn validate( role_groups, }) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + history::controller::build::test_support::minimal_validated_cluster, + test_support::app_version_label, + }; + + /// Locks every value the validate step itself derives from the minimal fixture — so a + /// validation regression fails here, with a validate-shaped message, instead of surfacing as + /// a confusing build-test failure downstream. + /// + /// The merged per-role-group config is produced by the config merge machinery, whose + /// contracts are tested in operator-rs; only the values this module derives on top are + /// re-asserted here. + #[test] + fn validate_ok_derives_expected_values() { + let validated = minimal_validated_cluster(); + + assert_eq!(validated.name.to_string(), "my-history"); + assert_eq!(validated.namespace.to_string(), "default"); + assert_eq!( + validated.uid.to_string(), + "12345678-1234-1234-1234-123456789012" + ); + assert_eq!( + validated.resolved_product_image.image, + format!( + "oci.example.org/sdp/spark-k8s:{}", + app_version_label("3.5.8") + ) + ); + assert_eq!(validated.resolved_product_image.product_version, "3.5.8"); + assert_eq!( + validated.product_version.to_string(), + app_version_label("3.5.8") + ); + + // The custom log directory is carried through, along with the event-log settings the + // history server derives from it; no cleaner role group and no extra Spark config. + let cluster_config = &validated.cluster_config; + assert!(matches!( + &cluster_config.log_dir, + ResolvedLogDir::Custom(dir) if dir == "file:///stackable/spark/logs" + )); + assert_eq!(cluster_config.cleaner_rolegroup_name, None); + assert!(cluster_config.spark_conf.is_empty()); + assert_eq!( + cluster_config.log_dir_settings, + BTreeMap::from([( + "spark.history.fs.logDirectory".to_string(), + "file:///stackable/spark/logs".to_string(), + )]) + ); + + // The role config falls back to its defaults: PDBs enabled, cluster-internal listener. + assert!(validated.role_config.pdb.enabled); + assert_eq!(validated.role_config.pdb.max_unavailable, None); + assert_eq!( + validated.role_config.listener_class.to_string(), + "cluster-internal" + ); + + // The single `default` role group; the Vector agent is off. + let role_group_names: Vec = validated + .role_groups + .keys() + .map(ToString::to_string) + .collect(); + assert_eq!(role_group_names, ["default"]); + let role_group = validated + .role_groups + .values() + .next() + .expect("the default role group exists"); + assert_eq!(role_group.config.replicas, Some(1)); + assert!(!role_group.logging.enable_vector_agent); + assert_eq!(role_group.logging.vector_container, None); + } +} diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 616f89e1..31499410 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -54,6 +54,8 @@ mod history; mod pod_driver_controller; mod product_logging; mod spark_k8s_controller; +#[cfg(test)] +pub(crate) mod test_support; mod webhooks; mod built_info { diff --git a/rust/operator-binary/src/spark_k8s_controller/build/mod.rs b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs index 9d2e136c..e08d138b 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/mod.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs @@ -3,6 +3,9 @@ pub mod resource; use resource::{config_map, job}; use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + builder::meta::ObjectMetaBuilder, v2::builder::meta::ownerreference_from_resource, +}; use crate::{ crd::roles::SparkApplicationRole, @@ -122,3 +125,20 @@ pub fn build(validated: &ValidatedSparkApplication) -> Result { job, }) } + +/// Object metadata for a child resource named `name`, owned by the SparkApplication and +/// carrying the recommended labels for the given `role`. Returns the builder so callers can add +/// extra labels before building. +pub(crate) fn object_meta( + validated: &ValidatedSparkApplication, + name: impl Into, + role: &str, +) -> ObjectMetaBuilder { + let mut builder = ObjectMetaBuilder::new(); + builder + .namespace(validated.namespace.clone()) + .name(name) + .ownerreference(ownerreference_from_resource(validated, None, Some(true))) + .with_labels(validated.recommended_labels(role)); + builder +} diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs index 002e85fe..21c208b3 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs @@ -20,7 +20,10 @@ use crate::{ }, product_logging::{self}, spark_k8s_controller::{ - build::pod::{self, pod_template}, + build::{ + object_meta, + pod::{self, pod_template}, + }, validate, }, }; @@ -108,7 +111,7 @@ pub(crate) fn pod_template_config_map( let mut cm_builder = ConfigMapBuilder::new(); cm_builder - .metadata(validated.object_meta(&cm_name, "pod-templates").build()) + .metadata(object_meta(validated, &cm_name, "pod-templates").build()) .add_data( POD_TEMPLATE_FILE, serde_yaml::to_string(&template).context(PodTemplateSerdeSnafu)?, @@ -152,7 +155,7 @@ pub(crate) fn submit_job_config_map( let mut cm_builder = ConfigMapBuilder::new(); - cm_builder.metadata(validated.object_meta(&cm_name, "spark-submit").build()); + cm_builder.metadata(object_meta(validated, &cm_name, "spark-submit").build()); cm_builder.add_data( SPARK_ENV_SH_FILE_NAME, diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs index 617a4666..6ff15f79 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs @@ -17,7 +17,10 @@ use crate::{ roles::{SparkApplicationRole, SparkContainer, SubmitConfig}, tlscerts, }, - spark_k8s_controller::{build::pod::security_context, validate}, + spark_k8s_controller::{ + build::{object_meta, pod::security_context}, + validate, + }, }; #[derive(Snafu, Debug)] @@ -139,9 +142,7 @@ pub(crate) fn spark_job( } let job = Job { - metadata: validated - .object_meta(validated.name.to_string(), "spark-job") - .build(), + metadata: object_meta(validated, validated.name.to_string(), "spark-job").build(), spec: Some(JobSpec { template: pod, ttl_seconds_after_finished: Some(600), diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs index 40ca4c46..2f88222c 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs @@ -6,7 +6,10 @@ use stackable_operator::k8s_openapi::{ }, }; -use crate::{crd::constants::*, spark_k8s_controller::validate}; +use crate::{ + crd::constants::*, + spark_k8s_controller::{build::object_meta, validate}, +}; /// For a given SparkApplication, we create a ServiceAccount with a RoleBinding to the ClusterRole /// that allows the driver to create pods etc. @@ -17,12 +20,12 @@ pub(crate) fn build_spark_role_serviceaccount( ) -> (ServiceAccount, RoleBinding) { let sa_name = validated.name.to_string(); let sa = ServiceAccount { - metadata: validated.object_meta(&sa_name, "service-account").build(), + metadata: object_meta(validated, &sa_name, "service-account").build(), ..ServiceAccount::default() }; let binding_name = &sa_name; let binding = RoleBinding { - metadata: validated.object_meta(binding_name, "role-binding").build(), + metadata: object_meta(validated, binding_name, "role-binding").build(), role_ref: RoleRef { api_group: Some(ClusterRole::GROUP.to_string()), kind: ClusterRole::KIND.to_string(), diff --git a/rust/operator-binary/src/spark_k8s_controller/validate.rs b/rust/operator-binary/src/spark_k8s_controller/validate.rs index 8c0d47bc..b48606eb 100644 --- a/rust/operator-binary/src/spark_k8s_controller/validate.rs +++ b/rust/operator-binary/src/spark_k8s_controller/validate.rs @@ -7,7 +7,6 @@ use std::{borrow::Cow, str::FromStr}; use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::meta::ObjectMetaBuilder, cli::OperatorEnvironmentOptions, commons::{ product_image_selection::{self, ResolvedProductImage}, @@ -19,7 +18,6 @@ use stackable_operator::{ kvp::Labels, v2::{ HasName, HasUid, NameIsValidLabelValue, - builder::meta::ownerreference_from_resource, controller_utils::{get_cluster_name, get_namespace, get_uid}, kvp::label::recommended_labels, types::{ @@ -170,19 +168,6 @@ impl ValidatedSparkApplication { &role_group, ) } - - /// Object metadata for a child resource named `name`, owned by this SparkApplication and - /// carrying the recommended labels for the given `role`. Returns the builder so callers can add - /// extra labels before building. - pub(crate) fn object_meta(&self, name: impl Into, role: &str) -> ObjectMetaBuilder { - let mut builder = ObjectMetaBuilder::new(); - builder - .namespace(self.namespace.clone()) - .name(name) - .ownerreference(ownerreference_from_resource(self, None, Some(true))) - .with_labels(self.recommended_labels(role)); - builder - } } /// The product name (`spark-k8s`) as a type-safe label value. @@ -256,3 +241,76 @@ fn reject_tls_no_verification(conn: &s3::v1alpha1::ConnectionSpec, context: &str } Ok(()) } + +#[cfg(test)] +mod tests { + use indoc::indoc; + use stackable_operator::cli::OperatorEnvironmentOptions; + + use super::*; + use crate::test_support::app_version_label; + + /// Locks every value the validate step itself derives from the minimal fixture. The raw + /// `SparkApplication` is deliberately retained on the validated type (see the field docs), + /// so only the resolved identity, image and cluster config are derived here. + #[test] + fn validate_ok_derives_expected_values() { + let yaml = indoc! {r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + "#}; + let deserializer = serde_yaml::Deserializer::from_str(yaml); + let spark_application: v1alpha1::SparkApplication = + serde_yaml::with::singleton_map_recursive::deserialize(deserializer) + .expect("invalid test SparkApplication YAML"); + + let validated = validate( + DereferencedSparkApplication { + spark_application, + resolved_template_refs: Vec::new(), + s3_connection: None, + log_dir: None, + }, + &OperatorEnvironmentOptions { + operator_namespace: "stackable-operators".to_string(), + operator_service_name: "spark-k8s-operator".to_string(), + image_repository: "oci.example.org/sdp".to_string(), + }, + ) + .expect("the minimal fixture validates"); + + assert_eq!(validated.name.to_string(), "spark-example"); + assert_eq!(validated.namespace.to_string(), "default"); + assert_eq!( + validated.uid.to_string(), + "12345678-1234-1234-1234-123456789012" + ); + assert_eq!( + validated.resolved_product_image.image, + format!( + "oci.example.org/sdp/spark-k8s:{}", + app_version_label("1.2.3") + ) + ); + assert_eq!(validated.resolved_product_image.product_version, "1.2.3"); + + // The minimal fixture references no templates, no S3 connection and no log directory, + // and the raw spec is retained for the pod builders. + assert!(validated.cluster_config.resolved_template_refs.is_empty()); + assert!(validated.cluster_config.s3_connection.is_none()); + assert!(validated.cluster_config.log_dir.is_none()); + assert_eq!( + validated.spark_application.metadata.name.as_deref(), + Some("spark-example") + ); + } +} diff --git a/rust/operator-binary/src/test_support.rs b/rust/operator-binary/src/test_support.rs new file mode 100644 index 00000000..96721cff --- /dev/null +++ b/rust/operator-binary/src/test_support.rs @@ -0,0 +1,13 @@ +//! Shared helpers for the crate's tests. + +/// The expected `app.kubernetes.io/version` label value for the given product version. +/// +/// The `-stackable` suffix carries the operator's own version, which is `0.0.0-dev` on main +/// but rewritten by the release process — so tests must derive it rather than hardcode it, +/// or they fail on release branches. +pub fn app_version_label(product_version: &str) -> String { + format!( + "{product_version}-stackable{}", + crate::built_info::PKG_VERSION + ) +}