Implement affinities

siegfriedweber · siegfriedweber · commit 239a379114bf · 2025-07-14T14:31:39.000+02:00
diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs
@@ -5,7 +5,7 @@ use build::build;
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     cluster_resources::ClusterResourceApplyStrategy,
-    commons::product_image_selection::ProductImage,
+    commons::{affinity::StackableAffinity, product_image_selection::ProductImage},
     k8s_openapi::api::{
         apps::v1::StatefulSet,
         core::v1::{ConfigMap, Service, ServiceAccount},
@@ -28,7 +28,7 @@ use crate::{
     },
     framework::{
         ClusterName, ControllerName, HasNamespace, HasObjectName, HasUid, IsLabelValue,
-        OperatorName, ProductName, ProductVersion, RoleGroupName,
+        OperatorName, ProductName, ProductVersion, RoleGroupName, RoleName,
         role_utils::{GenericProductSpecificCommonConfig, RoleGroupConfig},
     },
 };
@@ -103,6 +103,7 @@ type OpenSearchRoleGroupConfig =
 
 #[derive(Clone, Debug, PartialEq)]
 pub struct ValidatedOpenSearchConfig {
+    pub affinity: StackableAffinity,
     pub node_roles: NodeRoles,
     pub resources: stackable_operator::commons::resources::Resources<v1alpha1::StorageConfig>,
     pub termination_grace_period_seconds: i64,
@@ -125,6 +126,10 @@ pub struct ValidatedCluster {
 }
 
 impl ValidatedCluster {
+    pub fn role_name() -> RoleName {
+        RoleName::from_str("nodes").expect("should be a valid role name")
+    }
+
     pub fn is_single_node(&self) -> bool {
         self.node_count() == 1
     }
@@ -233,7 +238,7 @@ pub async fn reconcile(
     // dereference (client required)
 
     // validate (no client required)
-    let validated_cluster = validate(cluster).context(ValidateClusterSnafu)?;
+    let validated_cluster = validate(&context.names, cluster).context(ValidateClusterSnafu)?;
 
     // build (no client required; infallible)
     let prepared_resources = build(&context.names, validated_cluster.clone());
diff --git a/rust/operator-binary/src/controller/build.rs b/rust/operator-binary/src/controller/build.rs
@@ -1,9 +1,8 @@
-use std::{marker::PhantomData, str::FromStr};
+use std::marker::PhantomData;
 
 use role_builder::RoleBuilder;
 
 use super::{ContextNames, KubernetesResources, Prepared, ValidatedCluster};
-use crate::framework::RoleName;
 
 pub mod node_config;
 pub mod role_builder;
@@ -14,9 +13,7 @@ pub fn build(names: &ContextNames, cluster: ValidatedCluster) -> KubernetesResou
     let mut stateful_sets = vec![];
     let mut services = vec![];
 
-    let role_name = RoleName::from_str("nodes").expect("should be a valid role name");
-
-    let role_builder = RoleBuilder::new(role_name, cluster.clone(), names);
+    let role_builder = RoleBuilder::new(cluster.clone(), names);
 
     for role_group_builder in role_builder.role_group_builders() {
         config_maps.push(role_group_builder.build_config_map());
diff --git a/rust/operator-binary/src/controller/build/node_config.rs b/rust/operator-binary/src/controller/build/node_config.rs
@@ -5,7 +5,7 @@ use super::ValidatedCluster;
 use crate::{
     controller::OpenSearchRoleGroupConfig,
     crd::v1alpha1,
-    framework::{RoleName, builder::pod::container::EnvVarSet, role_group_utils},
+    framework::{builder::pod::container::EnvVarSet, role_group_utils},
 };
 
 pub const CONFIGURATION_FILE_OPENSEARCH_YML: &str = "opensearch.yml";
@@ -67,7 +67,6 @@ pub const CONFIG_OPTION_NODE_ROLES: &str = "node.roles";
 pub const CONFIG_OPTION_PLUGINS_SECURITY_NODES_DN: &str = "plugins.security.nodes_dn";
 
 pub struct NodeConfig {
-    role_name: RoleName,
     cluster: ValidatedCluster,
     role_group_config: OpenSearchRoleGroupConfig,
     discovery_service_name: String,
@@ -77,13 +76,11 @@ pub struct NodeConfig {
 // variables.
 impl NodeConfig {
     pub fn new(
-        role_name: RoleName,
         cluster: ValidatedCluster,
         role_group_config: OpenSearchRoleGroupConfig,
         discovery_service_name: String,
     ) -> Self {
         Self {
-            role_name,
             cluster,
             role_group_config,
             discovery_service_name,
@@ -221,7 +218,7 @@ impl NodeConfig {
             for (role_group_name, role_group_config) in cluster_manager_configs {
                 let role_group_resource_names = role_group_utils::ResourceNames {
                     cluster_name: self.cluster.name.clone(),
-                    role_name: self.role_name.clone(),
+                    role_name: ValidatedCluster::role_name(),
                     role_group_name,
                 };
 
@@ -249,14 +246,18 @@ mod tests {
     };
 
     use stackable_operator::{
-        commons::{product_image_selection::ProductImage, resources::Resources},
+        commons::{
+            affinity::StackableAffinity, product_image_selection::ProductImage,
+            resources::Resources,
+        },
         k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, ObjectFieldSelector, PodTemplateSpec},
         kube::api::ObjectMeta,
         role_utils::GenericRoleConfig,
     };
 
     use super::*;
     use crate::{
+        controller::ValidatedOpenSearchConfig,
         crd::NodeRoles,
         framework::{ClusterName, ProductVersion, role_utils::GenericProductSpecificCommonConfig},
     };
@@ -277,13 +278,14 @@ mod tests {
             role_config: GenericRoleConfig::default(),
             role_group_configs: BTreeMap::new(),
         };
-        let role_name = RoleName::from_str("nodes").expect("should be a valid role name");
 
         let role_group_config = OpenSearchRoleGroupConfig {
             replicas: 1,
-            config: v1alpha1::OpenSearchConfig {
+            config: ValidatedOpenSearchConfig {
+                affinity: StackableAffinity::default(),
                 node_roles: NodeRoles::default(),
                 resources: Resources::default(),
+                termination_grace_period_seconds: 30,
             },
             config_overrides: HashMap::default(),
             env_overrides: [("TEST".to_owned(), "value".to_owned())].into(),
@@ -293,7 +295,6 @@ mod tests {
         };
 
         let node_config = NodeConfig::new(
-            role_name,
             cluster,
             role_group_config,
             "my-opensearch-cluster-manager".to_owned(),
diff --git a/rust/operator-binary/src/controller/build/role_builder.rs b/rust/operator-binary/src/controller/build/role_builder.rs
@@ -22,7 +22,7 @@ use crate::{
     controller::{ContextNames, ValidatedCluster},
     crd::v1alpha1,
     framework::{
-        RoleName,
+        IsLabelValue,
         builder::{
             meta::ownerreference_from_resource, pdb::pod_disruption_budget_builder_with_role,
         },
@@ -33,20 +33,14 @@ use crate::{
 const PDB_DEFAULT_MAX_UNAVAILABLE: u16 = 1;
 
 pub struct RoleBuilder<'a> {
-    role_name: RoleName,
     cluster: ValidatedCluster,
     context_names: &'a ContextNames,
     resource_names: ResourceNames,
 }
 
 impl<'a> RoleBuilder<'a> {
-    pub fn new(
-        role_name: RoleName,
-        cluster: ValidatedCluster,
-        context_names: &'a ContextNames,
-    ) -> RoleBuilder<'a> {
+    pub fn new(cluster: ValidatedCluster, context_names: &'a ContextNames) -> RoleBuilder<'a> {
         RoleBuilder {
-            role_name: role_name.clone(),
             cluster: cluster.clone(),
             context_names,
             resource_names: ResourceNames {
@@ -64,7 +58,6 @@ impl<'a> RoleBuilder<'a> {
             .iter()
             .map(|(role_group_name, role_group_config)| {
                 RoleGroupBuilder::new(
-                    self.role_name.clone(),
                     self.resource_names.service_account_name(),
                     self.cluster.clone(),
                     role_group_name.clone(),
@@ -154,7 +147,7 @@ impl<'a> RoleBuilder<'a> {
                 pod_disruption_budget_builder_with_role(
                     &self.cluster,
                     &self.context_names.product_name,
-                    &self.role_name,
+                    &ValidatedCluster::role_name(),
                     &self.context_names.operator_name,
                     &self.context_names.controller_name,
                 )
@@ -184,8 +177,8 @@ impl<'a> RoleBuilder<'a> {
         // Well-known Kubernetes labels
         let mut labels = Labels::role_selector(
             &self.cluster,
-            &self.context_names.product_name.to_string(),
-            &self.role_name.to_string(),
+            &self.context_names.product_name.to_label_value(),
+            &ValidatedCluster::role_name().to_label_value(),
         )
         .unwrap();
 
diff --git a/rust/operator-binary/src/controller/build/role_group_builder.rs b/rust/operator-binary/src/controller/build/role_group_builder.rs
@@ -5,9 +5,9 @@ use stackable_operator::{
         api::{
             apps::v1::{StatefulSet, StatefulSetSpec},
             core::v1::{
-                ConfigMap, ConfigMapVolumeSource, Container, ContainerPort, PodSecurityContext,
-                PodSpec, PodTemplateSpec, Probe, Service, ServicePort, ServiceSpec,
-                TCPSocketAction, Volume, VolumeMount,
+                Affinity, ConfigMap, ConfigMapVolumeSource, Container, ContainerPort,
+                PodSecurityContext, PodSpec, PodTemplateSpec, Probe, Service, ServicePort,
+                ServiceSpec, TCPSocketAction, Volume, VolumeMount,
             },
         },
         apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString},
@@ -20,7 +20,7 @@ use super::node_config::{CONFIGURATION_FILE_OPENSEARCH_YML, NodeConfig};
 use crate::{
     controller::{ContextNames, OpenSearchRoleGroupConfig, ValidatedCluster},
     framework::{
-        RoleGroupName, RoleName,
+        RoleGroupName,
         builder::meta::ownerreference_from_resource,
         kvp::label::{recommended_labels, role_group_selector},
         role_group_utils::ResourceNames,
@@ -39,7 +39,6 @@ const DATA_VOLUME_NAME: &str = "data";
 const OPENSEARCH_BASE_PATH: &str = "/usr/share/opensearch";
 
 pub struct RoleGroupBuilder<'a> {
-    role_name: RoleName,
     service_account_name: String,
     cluster: ValidatedCluster,
     node_config: NodeConfig,
@@ -51,7 +50,6 @@ pub struct RoleGroupBuilder<'a> {
 
 impl<'a> RoleGroupBuilder<'a> {
     pub fn new(
-        role_name: RoleName,
         service_account_name: String,
         cluster: ValidatedCluster,
         role_group_name: RoleGroupName,
@@ -60,11 +58,9 @@ impl<'a> RoleGroupBuilder<'a> {
         discovery_service_name: String,
     ) -> RoleGroupBuilder<'a> {
         RoleGroupBuilder {
-            role_name: role_name.clone(),
             service_account_name,
             cluster: cluster.clone(),
             node_config: NodeConfig::new(
-                role_name.clone(),
                 cluster.clone(),
                 role_group_config.clone(),
                 discovery_service_name,
@@ -74,7 +70,7 @@ impl<'a> RoleGroupBuilder<'a> {
             context_names,
             resource_names: ResourceNames {
                 cluster_name: cluster.name.clone(),
-                role_name,
+                role_name: ValidatedCluster::role_name(),
                 role_group_name,
             },
         }
@@ -136,6 +132,7 @@ impl<'a> RoleGroupBuilder<'a> {
         let mut node_role_labels = Labels::new();
         for node_role in self.role_group_config.config.node_roles.iter() {
             node_role_labels.insert(
+                // TODO Prefix the key
                 Label::try_from((format!("{node_role}"), "true".to_string()))
                     .expect("should be a valid label"),
             );
@@ -155,7 +152,24 @@ impl<'a> RoleGroupBuilder<'a> {
         let mut pod_template = PodTemplateSpec {
             metadata: Some(metadata),
             spec: Some(PodSpec {
+                affinity: Some(Affinity {
+                    node_affinity: self.role_group_config.config.affinity.node_affinity.clone(),
+                    pod_affinity: self.role_group_config.config.affinity.pod_affinity.clone(),
+                    pod_anti_affinity: self
+                        .role_group_config
+                        .config
+                        .affinity
+                        .pod_anti_affinity
+                        .clone(),
+                }),
                 containers: vec![container],
+                node_selector: self
+                    .role_group_config
+                    .config
+                    .affinity
+                    .node_selector
+                    .clone()
+                    .map(|wrapped| wrapped.node_selector),
                 security_context: Some(PodSecurityContext {
                     fs_group: Some(1000),
                     ..PodSecurityContext::default()
@@ -326,7 +340,7 @@ impl<'a> RoleGroupBuilder<'a> {
             &self.cluster.product_version,
             &self.context_names.operator_name,
             &self.context_names.controller_name,
-            &self.role_name,
+            &ValidatedCluster::role_name(),
             &self.role_group_name,
         )
     }
@@ -335,7 +349,7 @@ impl<'a> RoleGroupBuilder<'a> {
         role_group_selector(
             &self.cluster,
             &self.context_names.product_name,
-            &self.role_name,
+            &ValidatedCluster::role_name(),
             &self.role_group_name,
         )
     }
diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs
@@ -8,7 +8,9 @@ use stackable_operator::{
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
-use super::{ProductVersion, RoleGroupName, ValidatedCluster, ValidatedOpenSearchConfig};
+use super::{
+    ContextNames, ProductVersion, RoleGroupName, ValidatedCluster, ValidatedOpenSearchConfig,
+};
 use crate::{
     crd::v1alpha1::{self, OpenSearchConfig},
     framework::{
@@ -54,7 +56,10 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 
 // TODO split
 // no client needed
-pub fn validate(cluster: &v1alpha1::OpenSearchCluster) -> Result<ValidatedCluster> {
+pub fn validate(
+    names: &ContextNames,
+    cluster: &v1alpha1::OpenSearchCluster,
+) -> Result<ValidatedCluster> {
     let raw_cluster_name = cluster.meta().name.clone().context(GetClusterNameSnafu)?;
     let cluster_name = ClusterName::from_str(&raw_cluster_name).context(ParseClusterNameSnafu)?;
 
@@ -73,7 +78,11 @@ pub fn validate(cluster: &v1alpha1::OpenSearchCluster) -> Result<ValidatedCluste
         let merged_role_group: RoleGroup<OpenSearchConfig, _> = with_validated_config(
             role_group_config,
             &cluster.spec.nodes,
-            &v1alpha1::OpenSearchConfig::default_config(),
+            &v1alpha1::OpenSearchConfig::default_config(
+                &names.product_name,
+                &cluster_name,
+                &ValidatedCluster::role_name(),
+            ),
         )
         .context(ValidateOpenSearchConfigSnafu)?;
 
@@ -87,6 +96,7 @@ pub fn validate(cluster: &v1alpha1::OpenSearchCluster) -> Result<ValidatedCluste
             })?;
 
         let validated_config = ValidatedOpenSearchConfig {
+            affinity: merged_role_group.config.config.affinity,
             node_roles: merged_role_group.config.config.node_roles,
             resources: merged_role_group.config.config.resources,
             termination_grace_period_seconds,
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
diff --git a/tests/templates/kuttl/smoke/10-assert.yaml b/tests/templates/kuttl/smoke/10-assert.yaml
diff --git a/tests/templates/kuttl/smoke/10-install-opensearch.yaml b/tests/templates/kuttl/smoke/10-install-opensearch.yaml
