implement executor affinity and resource properties

Author: razvan

Cargo.lock

@@ -1922,6 +1922,36 @@ spec:
                     config:
                       default: {}
                       properties:
+                        affinity:
+                          default:
+                            nodeAffinity: null
+                            nodeSelector: null
+                            podAffinity: null
+                            podAntiAffinity: null
+                          description: These configuration settings control [Pod placement](https://docs.stackable.tech/home/nightly/concepts/operations/pod_placement).
+                          properties:
+                            nodeAffinity:
+                              description: Same as the `spec.affinity.nodeAffinity` field on the Pod, see the [Kubernetes docs](https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node)
+                              nullable: true
+                              type: object
+                              x-kubernetes-preserve-unknown-fields: true
+                            nodeSelector:
+                              additionalProperties:
+                                type: string
+                              description: Simple key-value pairs forming a nodeSelector, see the [Kubernetes docs](https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node)
+                              nullable: true
+                              type: object
+                            podAffinity:
+                              description: Same as the `spec.affinity.podAffinity` field on the Pod, see the [Kubernetes docs](https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node)
+                              nullable: true
+                              type: object
+                              x-kubernetes-preserve-unknown-fields: true
+                            podAntiAffinity:
+                              description: Same as the `spec.affinity.podAntiAffinity` field on the Pod, see the [Kubernetes docs](https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node)
+                              nullable: true
+                              type: object
+                              x-kubernetes-preserve-unknown-fields: true
+                          type: object
                         logging:
                           default:
                             containers: {}

Cargo.nix

@@ -282,8 +282,13 @@ pub async fn reconcile(
         })?;
 
     let executor_pod_template = serde_yaml::to_string(
-        &executor::executor_pod_template(scs, &executor_config, &executor_config_map)
-            .context(ExecutorPodTemplateSnafu)?,
+        &executor::executor_pod_template(
+            scs,
+            &executor_config,
+            &resolved_product_image,
+            &executor_config_map,
+        )
+        .context(ExecutorPodTemplateSnafu)?,
     )
     .context(ExecutorPodTemplateSerdeSnafu)?;
 

deploy/helm/spark-k8s-operator/crds/crds.yaml

@@ -3,6 +3,7 @@ use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     commons::{
+        affinity::{affinity_between_role_pods, StackableAffinity, StackableAffinityFragment},
         cluster_operation::ClusterOperation,
         product_image_selection::ProductImage,
         resources::{
@@ -14,8 +15,8 @@ use stackable_operator::{
         fragment::{self, Fragment, ValidationError},
         merge::Merge,
     },
-    k8s_openapi::apimachinery::pkg::api::resource::Quantity,
-    kube::CustomResource,
+    k8s_openapi::{api::core::v1::PodAntiAffinity, apimachinery::pkg::api::resource::Quantity},
+    kube::{CustomResource, ResourceExt},
     product_logging::{
         self,
         spec::{
@@ -31,6 +32,9 @@ use stackable_operator::{
 use stackable_versioned::versioned;
 use strum::{Display, EnumIter};
 
+use super::common::SparkConnectRole;
+use crate::crd::constants::APP_NAME;
+
 pub const CONNECT_CONTROLLER_NAME: &str = "connect";
 pub const CONNECT_FULL_CONTROLLER_NAME: &str = concatcp!(
     CONNECT_CONTROLLER_NAME,
@@ -68,6 +72,7 @@ pub enum Error {
 
 #[versioned(version(name = "v1alpha1"))]
 pub mod versioned {
+
     /// A Spark cluster connect server component. This resource is managed by the Stackable operator
     /// for Apache Spark. Find more information on how to use it in the
     /// [operator documentation](DOCS_BASE_URL_PLACEHOLDER/spark-k8s/usage-guide/connect-server).
@@ -174,6 +179,8 @@ pub mod versioned {
         pub resources: Resources<crate::connect::crd::ConnectStorageConfig, NoRuntimeLimits>,
         #[fragment_attrs(serde(default))]
         pub logging: Logging<crate::connect::crd::SparkConnectContainer>,
+        #[fragment_attrs(serde(default))]
+        pub affinity: StackableAffinity,
 
         /// Request secret (currently only autoTls certificates) lifetime from the secret operator, e.g. `7d`, or `30d`.
         /// This can be shortened by the `maxCertificateLifetime` setting on the SecretClass issuing the TLS certificate.
@@ -304,7 +311,7 @@ impl v1alpha1::SparkConnectServer {
     }
 
     pub fn executor_config(&self) -> Result<v1alpha1::ExecutorConfig, Error> {
-        let defaults = v1alpha1::ExecutorConfig::default_config();
+        let defaults = v1alpha1::ExecutorConfig::default_config(&self.name_unchecked());
         fragment::validate(
             match self.spec.executor.as_ref().map(|cc| cc.config.clone()) {
                 Some(fragment) => {
@@ -339,20 +346,22 @@ impl v1alpha1::ExecutorConfig {
     // Auto TLS certificate lifetime
     const DEFAULT_CONNECT_SECRET_LIFETIME: Duration = Duration::from_days_unchecked(1);
 
-    fn default_config() -> v1alpha1::ExecutorConfigFragment {
+    fn default_config(cluster_name: &str) -> v1alpha1::ExecutorConfigFragment {
         v1alpha1::ExecutorConfigFragment {
             resources: ResourcesFragment {
                 cpu: CpuLimitsFragment {
-                    min: Some(Quantity("250m".to_owned())),
+                    min: Some(Quantity("1".to_owned())),
                     max: Some(Quantity("1".to_owned())),
                 },
                 memory: MemoryLimitsFragment {
-                    limit: Some(Quantity("1024Mi".to_owned())),
+                    limit: Some(Quantity("1024M".to_owned())),
                     runtime_limits: NoRuntimeLimitsFragment {},
                 },
                 storage: ConnectStorageConfigFragment {},
             },
             logging: product_logging::spec::default_logging(),
+            affinity: v1alpha1::ExecutorConfig::affinity(cluster_name),
+
             requested_secret_lifetime: Some(Self::DEFAULT_CONNECT_SECRET_LIFETIME),
         }
     }
@@ -374,4 +383,25 @@ impl v1alpha1::ExecutorConfig {
             _ => None,
         }
     }
+
+    fn affinity(cluster_name: &str) -> StackableAffinityFragment {
+        let affinity_between_role_pods = affinity_between_role_pods(
+            APP_NAME,
+            cluster_name,
+            &SparkConnectRole::Executor.to_string(),
+            70,
+        );
+
+        StackableAffinityFragment {
+            pod_affinity: None,
+            pod_anti_affinity: Some(PodAntiAffinity {
+                preferred_during_scheduling_ignored_during_execution: Some(vec![
+                    affinity_between_role_pods,
+                ]),
+                required_during_scheduling_ignored_during_execution: None,
+            }),
+            node_affinity: None,
+            node_selector: None,
+        }
+    }
 }

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

@@ -8,7 +8,10 @@ use stackable_operator::{
         meta::ObjectMetaBuilder,
         pod::{container::ContainerBuilder, volume::VolumeBuilder, PodBuilder},
     },
-    commons::product_image_selection::ResolvedProductImage,
+    commons::{
+        product_image_selection::ResolvedProductImage,
+        resources::{CpuLimits, MemoryLimits, Resources},
+    },
     k8s_openapi::{
         api::core::v1::{ConfigMap, EnvVar, PodTemplateSpec},
         DeepMerge,
@@ -53,11 +56,8 @@ pub enum Error {
         source: stackable_operator::builder::meta::Error,
     },
 
-    #[snafu(display("failed to build metadata for spark connect pod template config map {name}"))]
-    MetadataBuild {
-        source: builder::meta::Error,
-        name: String,
-    },
+    #[snafu(display("failed to build metadata for spark connect executor pod template"))]
+    PodTemplateMetadataBuild { source: builder::meta::Error },
 
     #[snafu(display("invalid connect container name"))]
     InvalidContainerName {
@@ -109,6 +109,7 @@ pub enum Error {
 pub fn executor_pod_template(
     scs: &v1alpha1::SparkConnectServer,
     config: &v1alpha1::ExecutorConfig,
+    pi: &ResolvedProductImage,
     config_map: &ConfigMap,
 ) -> Result<PodTemplateSpec, Error> {
     let container_env = executor_env(
@@ -128,8 +129,19 @@ pub fn executor_pod_template(
         .add_volume_mount(VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_PATH_LOG)
         .context(AddVolumeMountSnafu)?;
 
+    let metadata = ObjectMetaBuilder::new()
+        .with_recommended_labels(common::labels(
+            scs,
+            &pi.app_version_label,
+            &SparkConnectRole::Executor.to_string(),
+        ))
+        .context(PodTemplateMetadataBuildSnafu)?
+        .build();
+
     let mut template = PodBuilder::new();
     template
+        .metadata(metadata)
+        .affinity(&config.affinity)
         .add_volume(
             VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG)
                 .with_empty_dir(
@@ -226,6 +238,37 @@ pub fn executor_properties(
     ]
     .into();
 
+    // ========================================
+    // Add executor resource properties
+    let Resources {
+        cpu: CpuLimits { min, max },
+        memory: MemoryLimits {
+            limit,
+            runtime_limits: _,
+        },
+        storage: _,
+    } = &config.resources;
+    result.insert(
+        "spark.kubernetes.executor.limit.cores".to_string(),
+        max.clone().map(|v| v.0),
+    );
+    result.insert("spark.executor.cores".to_string(), min.clone().map(|v| v.0));
+    result.insert(
+        "spark.executor.memory".to_string(),
+        limit.clone().map(|v| v.0),
+    );
+    // This ensures that the pod's memory limit is exactly the value
+    // in `config.resources.memory.limit`.
+    // By default, Spark computes an `executor.memoryOverhead` as 6-10% from the
+    // `executor.memory`.
+    result.insert(
+        "spark.executor.memoryOverhead".to_string(),
+        Some("0".to_string()),
+    );
+
+    // ========================================
+    // Add the user provided executor properties
+
     let config_overrides = scs
         .spec
         .executor

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

@@ -53,3 +53,14 @@ spec:
           spark:
             custom:
               configMap: spark-connect-log-config
+  executor:
+    configOverrides:
+      spark-defaults.conf:
+        spark.executor.memoryOverhead: 1m
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+        containers:
+          spark:
+            custom:
+              configMap: spark-connect-log-config