peter-toth commented on code in PR #823: URL: https://github.com/apache/spark-kubernetes-operator/pull/823#discussion_r4007335410
########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java: ########## @@ -0,0 +1,475 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodSpecBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSet; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.SparkClusterResourceSpec; +import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta1.WorkloadSpec; +import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.StringUtils; +import org.apache.spark.network.util.JavaUtils; + +/** + * Factory for creating Kueue Workload resources from Spark custom resources. + * This factory supports both {@link SparkApplication} (driver and executor pod sets) + * and {@link SparkCluster} (master and worker pod sets). + */ +@SuppressWarnings("PMD.GodClass") +public final class KueueWorkloadFactory { + + public static final String PODSET_DRIVER = "driver"; + public static final String PODSET_EXECUTOR = "executor"; + public static final String PODSET_MASTER = "master"; + public static final String PODSET_WORKER = "worker"; + + public static final String DEFAULT_CORES = "1"; + public static final String DEFAULT_MEMORY = "1g"; + public static final String DEFAULT_MIN_MEMORY_OVERHEAD = "384m"; + public static final double DEFAULT_MEMORY_OVERHEAD_FACTOR = 0.10; + public static final double NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.40; + public static final int DEFAULT_EXECUTOR_INSTANCES = 2; + + private KueueWorkloadFactory() {} + + /** + * Builds a Kueue Workload from a SparkApplication resource. + * + * @param app The SparkApplication. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkApplication app) { + String queueName = getQueueName(app); + ApplicationSpec appSpec = app.getSpec(); + Map<String, String> sparkConf = + appSpec != null && appSpec.getSparkConf() != null + ? appSpec.getSparkConf() + : Map.of(); + + PodSet driverPodSet = buildDriverPodSet(app, sparkConf); + PodSet executorPodSet = buildExecutorPodSet(app, sparkConf); + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(driverPodSet); + podSets.add(executorPodSet); + + boolean active = appSpec == null || !appSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (app.getMetadata().getLabels() != null) { + labels.putAll(app.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, app.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(app); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(app)) + .withNamespace(app.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + /** + * Builds a Kueue Workload from a SparkCluster resource. + * + * @param cluster The SparkCluster. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkCluster cluster) { + String queueName = getQueueName(cluster); + ClusterSpec clusterSpec = cluster.getSpec(); + + // Use the same StatefulSets which the operator creates for the cluster. + SparkClusterResourceSpec resourceSpec = + SparkClusterResourceSpecFactory.buildResourceSpec( + cluster, new SparkClusterSubmissionWorker()); + if (resourceSpec.getHorizontalPodAutoscaler().isPresent()) { + throw new UnsupportedOperationException( + "Kueue does not support SparkCluster with HorizontalPodAutoscaler " + + "(minWorkers < maxWorkers) yet."); + } + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(buildPodSet(PODSET_MASTER, resourceSpec.getMasterStatefulSet())); + podSets.add(buildPodSet(PODSET_WORKER, resourceSpec.getWorkerStatefulSet())); + + boolean active = clusterSpec == null || !clusterSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (cluster.getMetadata().getLabels() != null) { + labels.putAll(cluster.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_CLUSTER_NAME, cluster.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(cluster); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(cluster)) + .withNamespace(cluster.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + private static PodSet buildPodSet(final String name, final StatefulSet statefulSet) { + return PodSet.builder() + .name(name) + .count(statefulSet.getSpec().getReplicas()) + .template(statefulSet.getSpec().getTemplate()) + .build(); + } + + /** + * Returns the Workload name prefixed with the lower-cased kind of the owner resource, like Kueue + * built-in integrations, to avoid name collisions between SparkApplication and SparkCluster. + */ + private static String getWorkloadName(final HasMetadata resource) { + return resource.getKind().toLowerCase(Locale.ROOT) + "-" + resource.getMetadata().getName(); + } + + /** + * Extracts the Kueue queue name from the resource metadata labels. + * + * @param resource The Kubernetes resource (e.g. SparkApplication or SparkCluster). + * @return The queue name, or null if not found. + */ + public static String getQueueName(final HasMetadata resource) { + if (resource != null + && resource.getMetadata() != null + && resource.getMetadata().getLabels() != null) { + String queue = resource.getMetadata().getLabels().get(Constants.LABEL_QUEUE_NAME); + if (StringUtils.isNotEmpty(queue)) { + return queue; + } + } + return null; + } + + /** + * Checks whether the resource is configured to use Kueue. + * + * @param resource The Kubernetes resource (SparkApplication or SparkCluster). + * @return true if a Kueue queue name is specified. + */ + public static boolean hasQueueName(final HasMetadata resource) { + return StringUtils.isNotEmpty(getQueueName(resource)); + } + + /** + * Builds the driver PodSet. + */ + public static PodSet buildDriverPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getDriverSpec() != null + && app.getSpec().getDriverSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getDriverSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.driver.request.cores", + sparkConf.getOrDefault("spark.driver.cores", DEFAULT_CORES)); + long memoryMiB = calculateDriverMemoryMiB(sparkConf, isNonJvmApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.driver.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.driver.resource.gpu.vendor"); + + decorateTemplateResources( Review Comment: **Finding 13.** The PodSet template omits the node selectors Spark injects, so Kueue sees a less constrained pod than the one that will actually be created. `BasicDriverFeatureStep.configurePod` does `.addToNodeSelector(conf.nodeSelector).addToNodeSelector(conf.driverNodeSelector)` (`BasicDriverFeatureStep.scala:153-154` on `branch-4.2`), where those two maps are `spark.kubernetes.node.selector.*` and `spark.kubernetes.driver.node.selector.*` with the prefix stripped (`KubernetesConf.scala:69` and `:101`). `BasicExecutorFeatureStep` does the same with the executor prefix (`BasicExecutorFeatureStep.scala:310-311`). Here only what the user wrote into `podTemplateSpec` survives. Kueue's own field doc on `PodSet.template` says the template's `nodeSelector` and required node affinity are matched against a `ResourceFlavor`'s node labels during admission to filter the eligible flavors. A pod with no `nodeSelector` conflicts with no flavor, so it matches all of them. Measured on `0ddd85a` with `spark.kubernetes.node.selector.karpenter.sh/nodepool=gpu` plus `spark.kubernetes.{driver,executor}.node.selector.node.kubernetes.io/instance-type=p4d.24xlarge`: DRIVER nodeSelector = {} EXEC nodeSelector = {} Both should carry all three entries. Fix shape, as a sibling of `decorateTemplateResources`: ```java private static void decorateTemplateNodeSelector( final PodTemplateSpec templateSpec, final Map<String, String> sparkConf, final String rolePrefix) { PodSpec podSpec = templateSpec.getSpec(); if (podSpec.getNodeSelector() == null) { podSpec.setNodeSelector(new HashMap<>()); } // Like Spark, the role-specific prefix is applied last so that it wins. putPrefixedKeyValuePairs(podSpec.getNodeSelector(), sparkConf, NODE_SELECTOR_PREFIX); putPrefixedKeyValuePairs(podSpec.getNodeSelector(), sparkConf, rolePrefix); } ``` Tolerations and affinity are fine as they are, since Spark only ever takes those from the pod template. If you would rather defer this, please list it under Unsupported Features in the description. Dropping it silently produces a plausible-looking Workload rather than an error, so nothing surfaces it. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java: ########## @@ -0,0 +1,475 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodSpecBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSet; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.SparkClusterResourceSpec; +import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta1.WorkloadSpec; +import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.StringUtils; +import org.apache.spark.network.util.JavaUtils; + +/** + * Factory for creating Kueue Workload resources from Spark custom resources. + * This factory supports both {@link SparkApplication} (driver and executor pod sets) + * and {@link SparkCluster} (master and worker pod sets). + */ +@SuppressWarnings("PMD.GodClass") +public final class KueueWorkloadFactory { + + public static final String PODSET_DRIVER = "driver"; + public static final String PODSET_EXECUTOR = "executor"; + public static final String PODSET_MASTER = "master"; + public static final String PODSET_WORKER = "worker"; + + public static final String DEFAULT_CORES = "1"; + public static final String DEFAULT_MEMORY = "1g"; + public static final String DEFAULT_MIN_MEMORY_OVERHEAD = "384m"; + public static final double DEFAULT_MEMORY_OVERHEAD_FACTOR = 0.10; + public static final double NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.40; + public static final int DEFAULT_EXECUTOR_INSTANCES = 2; + + private KueueWorkloadFactory() {} + + /** + * Builds a Kueue Workload from a SparkApplication resource. + * + * @param app The SparkApplication. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkApplication app) { + String queueName = getQueueName(app); + ApplicationSpec appSpec = app.getSpec(); + Map<String, String> sparkConf = + appSpec != null && appSpec.getSparkConf() != null + ? appSpec.getSparkConf() + : Map.of(); + + PodSet driverPodSet = buildDriverPodSet(app, sparkConf); + PodSet executorPodSet = buildExecutorPodSet(app, sparkConf); + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(driverPodSet); + podSets.add(executorPodSet); + + boolean active = appSpec == null || !appSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (app.getMetadata().getLabels() != null) { + labels.putAll(app.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, app.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(app); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(app)) + .withNamespace(app.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + /** + * Builds a Kueue Workload from a SparkCluster resource. + * + * @param cluster The SparkCluster. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkCluster cluster) { + String queueName = getQueueName(cluster); + ClusterSpec clusterSpec = cluster.getSpec(); + + // Use the same StatefulSets which the operator creates for the cluster. + SparkClusterResourceSpec resourceSpec = + SparkClusterResourceSpecFactory.buildResourceSpec( + cluster, new SparkClusterSubmissionWorker()); + if (resourceSpec.getHorizontalPodAutoscaler().isPresent()) { + throw new UnsupportedOperationException( + "Kueue does not support SparkCluster with HorizontalPodAutoscaler " + + "(minWorkers < maxWorkers) yet."); + } + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(buildPodSet(PODSET_MASTER, resourceSpec.getMasterStatefulSet())); + podSets.add(buildPodSet(PODSET_WORKER, resourceSpec.getWorkerStatefulSet())); + + boolean active = clusterSpec == null || !clusterSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (cluster.getMetadata().getLabels() != null) { + labels.putAll(cluster.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_CLUSTER_NAME, cluster.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(cluster); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(cluster)) + .withNamespace(cluster.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + private static PodSet buildPodSet(final String name, final StatefulSet statefulSet) { + return PodSet.builder() + .name(name) + .count(statefulSet.getSpec().getReplicas()) + .template(statefulSet.getSpec().getTemplate()) + .build(); + } + + /** + * Returns the Workload name prefixed with the lower-cased kind of the owner resource, like Kueue + * built-in integrations, to avoid name collisions between SparkApplication and SparkCluster. + */ + private static String getWorkloadName(final HasMetadata resource) { + return resource.getKind().toLowerCase(Locale.ROOT) + "-" + resource.getMetadata().getName(); + } + + /** + * Extracts the Kueue queue name from the resource metadata labels. + * + * @param resource The Kubernetes resource (e.g. SparkApplication or SparkCluster). + * @return The queue name, or null if not found. + */ + public static String getQueueName(final HasMetadata resource) { + if (resource != null + && resource.getMetadata() != null + && resource.getMetadata().getLabels() != null) { + String queue = resource.getMetadata().getLabels().get(Constants.LABEL_QUEUE_NAME); + if (StringUtils.isNotEmpty(queue)) { + return queue; + } + } + return null; + } + + /** + * Checks whether the resource is configured to use Kueue. + * + * @param resource The Kubernetes resource (SparkApplication or SparkCluster). + * @return true if a Kueue queue name is specified. + */ + public static boolean hasQueueName(final HasMetadata resource) { + return StringUtils.isNotEmpty(getQueueName(resource)); + } + + /** + * Builds the driver PodSet. + */ + public static PodSet buildDriverPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getDriverSpec() != null + && app.getSpec().getDriverSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getDriverSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.driver.request.cores", + sparkConf.getOrDefault("spark.driver.cores", DEFAULT_CORES)); + long memoryMiB = calculateDriverMemoryMiB(sparkConf, isNonJvmApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.driver.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.driver.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get(Constants.DRIVER_SPARK_CONTAINER_PROP_KEY), + "spark-driver", + cpu, + memoryMiB, + gpuAmount, + gpuVendor); + + return PodSet.builder() + .name(PODSET_DRIVER) + .count(1) + .template(templateSpec) + .build(); + } + + /** + * Builds the executor PodSet. + */ + public static PodSet buildExecutorPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + if ("true".equalsIgnoreCase(sparkConf.get("spark.dynamicAllocation.enabled"))) { + throw new UnsupportedOperationException( + "Kueue does not support SparkApplication with dynamic allocation " + + "(spark.dynamicAllocation.enabled=true) yet."); + } + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getExecutorSpec() != null + && app.getSpec().getExecutorSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getExecutorSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.executor.request.cores", + sparkConf.getOrDefault("spark.executor.cores", DEFAULT_CORES)); + long memoryMiB = + calculateExecutorMemoryMiB( + sparkConf, isNonJvmApp(app.getSpec()), isPythonApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.executor.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.executor.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get("spark.kubernetes.executor.podTemplateContainerName"), + "spark-executor", + cpu, + memoryMiB, + gpuAmount, + gpuVendor); + + return PodSet.builder() + .name(PODSET_EXECUTOR) + .count(parseInt(sparkConf.get("spark.executor.instances"), DEFAULT_EXECUTOR_INSTANCES)) + .template(templateSpec) + .build(); + } + + /** + * Calculates total driver memory in MiB including overhead. + */ + public static long calculateDriverMemoryMiB( + final Map<String, String> sparkConf, final boolean isNonJvm) { + long memMiB = + JavaUtils.byteStringAsMb(sparkConf.getOrDefault("spark.driver.memory", DEFAULT_MEMORY)); + return memMiB + calculateMemoryOverheadMiB(sparkConf, "spark.driver", memMiB, isNonJvm); + } + + /** + * Calculates total executor memory in MiB including overhead, off-heap memory and PySpark + * memory. + */ + public static long calculateExecutorMemoryMiB( + final Map<String, String> sparkConf, final boolean isNonJvm, final boolean isPython) { + long memMiB = + JavaUtils.byteStringAsMb(sparkConf.getOrDefault("spark.executor.memory", DEFAULT_MEMORY)); + long total = + memMiB + calculateMemoryOverheadMiB(sparkConf, "spark.executor", memMiB, isNonJvm); + if ("true".equalsIgnoreCase(sparkConf.get("spark.memory.offHeap.enabled"))) { + // `spark.memory.offHeap.size` is in bytes unless otherwise specified. + total += + JavaUtils.byteStringAsBytes(sparkConf.getOrDefault("spark.memory.offHeap.size", "0")) + / 1024 + / 1024; + } + if (isPython && sparkConf.containsKey("spark.executor.pyspark.memory")) { + total += JavaUtils.byteStringAsMb(sparkConf.get("spark.executor.pyspark.memory")); + } + return total; + } + + private static long calculateMemoryOverheadMiB( + final Map<String, String> sparkConf, + final String prefix, + final long memMiB, + final boolean isNonJvm) { + String overhead = sparkConf.get(prefix + ".memoryOverhead"); + if (StringUtils.isNotEmpty(overhead)) { + return JavaUtils.byteStringAsMb(overhead); + } + // Like Spark's BasicDriverFeatureStep, the deprecated `spark.kubernetes.memoryOverheadFactor` + // or the default factor (0.4 for non-JVM applications) is used if not set explicitly. + double defaultFactor = + parseDouble( + sparkConf.get("spark.kubernetes.memoryOverheadFactor"), + isNonJvm ? NON_JVM_MEMORY_OVERHEAD_FACTOR : DEFAULT_MEMORY_OVERHEAD_FACTOR); + double factor = parseDouble(sparkConf.get(prefix + ".memoryOverheadFactor"), defaultFactor); + long minOverheadMiB = + JavaUtils.byteStringAsMb( + sparkConf.getOrDefault(prefix + ".minMemoryOverhead", DEFAULT_MIN_MEMORY_OVERHEAD)); + return Math.max((int) (factor * memMiB), minOverheadMiB); + } + + /** Follows the main application resource selection of SparkAppSubmissionWorker. */ + private static boolean isPythonApp(final ApplicationSpec spec) { + return StringUtils.isEmpty(spec.getJars()) + && ("org.apache.spark.deploy.PythonRunner".equals(spec.getMainClass()) + || StringUtils.isNotEmpty(spec.getPyFiles())); + } + + private static boolean isNonJvmApp(final ApplicationSpec spec) { + return isPythonApp(spec) + || (StringUtils.isEmpty(spec.getJars()) && StringUtils.isNotEmpty(spec.getSparkRFiles())); + } + + private static void ensurePodSpec(final PodTemplateSpec templateSpec) { + if (templateSpec.getSpec() == null) { + templateSpec.setSpec(new PodSpecBuilder().build()); + } + if (templateSpec.getSpec().getContainers() == null) { + templateSpec.getSpec().setContainers(new ArrayList<>()); + } + } + + private static void decorateTemplateResources( + final PodTemplateSpec templateSpec, + final String containerName, + final String defaultContainerName, + final String cpu, + final long memoryMiB, + final String gpuAmount, + final String gpuVendor) { + PodSpec podSpec = templateSpec.getSpec(); + Container container; + if (podSpec.getContainers().isEmpty()) { + container = new ContainerBuilder().withName(defaultContainerName).build(); + podSpec.getContainers().add(container); + } else { + // Like Spark's KubernetesUtils.selectSparkContainer, select the container by name and + // fall back to the first container. + container = + podSpec.getContainers().stream() + .filter(c -> containerName != null && containerName.equals(c.getName())) + .findFirst() + .orElse(podSpec.getContainers().get(0)); + } + + ResourceRequirements resources = container.getResources(); + if (resources == null) { + resources = new ResourceRequirementsBuilder().build(); + container.setResources(resources); + } + + Map<String, Quantity> requests = resources.getRequests(); + if (requests == null) { + requests = new HashMap<>(); + resources.setRequests(requests); + } + + // Like Spark, overwrite the requests of the pod template. + requests.put("cpu", new Quantity(cpu)); + requests.put("memory", new Quantity(memoryMiB + "Mi")); + + if (StringUtils.isNotEmpty(gpuAmount)) { + if (StringUtils.isEmpty(gpuVendor)) { + throw new IllegalArgumentException( + "Resource: gpu was requested, but vendor was not specified."); + } + // Like Spark's KubernetesConf.buildKubernetesResourceName, e.g., `nvidia.com/gpu`. + String gpuResourceName = gpuVendor + "/gpu"; Review Comment: **Finding 14.** `gpu` is not the only resource name Spark accepts here. `spark.{driver,executor}.resource.<resourceName>.amount` / `.vendor` is generic, and Spark turns each pair into `<vendor>/<resourceName>` (`KubernetesUtils.buildResourcesQuantities` for the driver, `BasicExecutorFeatureStep.buildExecutorResourcesQuantities` for the executor, both via `KubernetesConf.buildKubernetesResourceName`). Measured on `0ddd85a`, with `spark.executor.resource.gpu.{amount,vendor} = 1, nvidia.com` and `spark.executor.resource.fpga.{amount,vendor} = 2, xilinx.com`: EXEC requests = {cpu=1, memory=1408Mi, nvidia.com/gpu=1} The executor pod that runs requests `xilinx.com/fpga: 2` as well, so Kueue admits against a quota that never accounts for it. The generic form needs no new dependency, and it also lets the two `spark.{driver,executor}.resource.gpu.*` reads in the callers go away: ```java // Like Spark's `KubernetesConf.buildKubernetesResourceName`, e.g., `nvidia.com/gpu`. String prefix = rolePrefix + ".resource."; // e.g., `spark.executor.resource.` for (Map.Entry<String, String> e : sparkConf.entrySet()) { if (!e.getKey().startsWith(prefix) || !e.getKey().endsWith(".amount")) { continue; } String name = e.getKey().substring(prefix.length(), e.getKey().length() - ".amount".length()); String vendor = sparkConf.get(prefix + name + ".vendor"); if (StringUtils.isEmpty(vendor)) { throw new IllegalArgumentException( "Resource: " + name + " was requested, but vendor was not specified."); } requests.put(vendor + "/" + name, new Quantity(e.getValue())); limits.put(vendor + "/" + name, new Quantity(e.getValue())); } ``` ########## spark-operator/src/test/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactoryTest.java: ########## @@ -0,0 +1,497 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.HashMap; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSetSpecBuilder; +import org.junit.jupiter.api.Test; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.BaseApplicationTemplateSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.spec.ClusterTolerations; +import org.apache.spark.k8s.operator.spec.MasterSpec; +import org.apache.spark.k8s.operator.spec.RuntimeVersions; +import org.apache.spark.k8s.operator.spec.WorkerInstanceConfig; +import org.apache.spark.k8s.operator.spec.WorkerSpec; + +class KueueWorkloadFactoryTest { + + @Test + void testCalculateDriverMemoryMiB() { + // 1g memory -> 1024MiB. Overhead factor 0.10 -> 102MiB -> min 384MiB. + // Total = 1024 + 384 = 1408 + Map<String, String> conf = new HashMap<>(); + conf.put("spark.driver.memory", "1g"); + assertEquals(1408L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, false)); + + // Explicit overhead 512m -> 1024 + 512 = 1536 + conf.put("spark.driver.memoryOverhead", "512m"); + assertEquals(1536L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, false)); + + // Custom overhead factor 0.50 -> 1024 * 0.50 = 512 -> 1024 + 512 = 1536 + conf.remove("spark.driver.memoryOverhead"); + conf.put("spark.driver.memoryOverheadFactor", "0.50"); + assertEquals(1536L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, false)); + + // Custom minimum overhead 1g -> 1024 + max(512, 1024) = 2048 + conf.put("spark.driver.minMemoryOverhead", "1g"); + assertEquals(2048L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, false)); + } + + @Test + void testCalculateDriverMemoryMiBForNonJvmApp() { + Map<String, String> conf = new HashMap<>(); + conf.put("spark.driver.memory", "8g"); + // Non-JVM: 8192 + 0.40 * 8192 = 8192 + 3276 = 11468 + assertEquals(11468L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, true)); + // JVM: 8192 + 0.10 * 8192 = 8192 + 819 = 9011 + assertEquals(9011L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, false)); + + // The deprecated `spark.kubernetes.memoryOverheadFactor` 0.20 -> 8192 + 1638 = 9830 + conf.put("spark.kubernetes.memoryOverheadFactor", "0.20"); + assertEquals(9830L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, true)); + assertEquals(9830L, KueueWorkloadFactory.calculateDriverMemoryMiB(conf, false)); + } + + @Test + void testCalculateExecutorMemoryMiB() { + Map<String, String> conf = new HashMap<>(); + // 2048MiB. Overhead: max(204, 384) = 384. Total = 2432 + conf.put("spark.executor.memory", "2g"); + assertEquals(2432L, KueueWorkloadFactory.calculateExecutorMemoryMiB(conf, false, false)); + + // PySpark memory is added only for Python applications + conf.put("spark.executor.pyspark.memory", "1g"); + assertEquals(2432L, KueueWorkloadFactory.calculateExecutorMemoryMiB(conf, false, false)); + // Non-JVM overhead: max(819, 384) = 819. Total = 2048 + 819 + 1024 = 3891 + assertEquals(3891L, KueueWorkloadFactory.calculateExecutorMemoryMiB(conf, true, true)); + } + + @Test + void testCalculateExecutorMemoryMiBWithOffHeap() { + Map<String, String> conf = new HashMap<>(); + conf.put("spark.executor.memory", "4g"); + conf.put("spark.memory.offHeap.size", "4g"); + // Off-heap memory is ignored if disabled. 4096 + 409 = 4505 + assertEquals(4505L, KueueWorkloadFactory.calculateExecutorMemoryMiB(conf, false, false)); + + // 4096 + 409 + 4096 = 8601 + conf.put("spark.memory.offHeap.enabled", "true"); + assertEquals(8601L, KueueWorkloadFactory.calculateExecutorMemoryMiB(conf, false, false)); + + // `spark.memory.offHeap.size` is in bytes unless otherwise specified + conf.put("spark.memory.offHeap.size", "4294967296"); + assertEquals(8601L, KueueWorkloadFactory.calculateExecutorMemoryMiB(conf, false, false)); + } + + @Test + void testCalculateMemoryWithUnits() { + // 2048 + 384 = 2432 + assertEquals( + 2432L, + KueueWorkloadFactory.calculateExecutorMemoryMiB( + Map.of("spark.executor.memory", "2Gi"), false, false)); + assertThrows( + NumberFormatException.class, + () -> + KueueWorkloadFactory.calculateDriverMemoryMiB( + Map.of("spark.driver.memory", "10zz"), false)); + assertThrows( + NumberFormatException.class, + () -> + KueueWorkloadFactory.calculateExecutorMemoryMiB( + Map.of("spark.executor.memory", "1.5g"), false, false)); + } + + @Test + void testHasAndGetQueueName() { + SparkApplication app = new SparkApplication(); + assertFalse(KueueWorkloadFactory.hasQueueName(app)); + assertNull(KueueWorkloadFactory.getQueueName(app)); + + // From label + app.setMetadata( + new ObjectMetaBuilder() + .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "test-queue")) + .build()); + assertTrue(KueueWorkloadFactory.hasQueueName(app)); + assertEquals("test-queue", KueueWorkloadFactory.getQueueName(app)); + + // Empty label + SparkApplication app2 = new SparkApplication(); + app2.setMetadata( + new ObjectMetaBuilder() + .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "")) + .build()); + assertFalse(KueueWorkloadFactory.hasQueueName(app2)); + assertNull(KueueWorkloadFactory.getQueueName(app2)); + } + + @Test + void testBuildWorkloadStaticAllocation() { + SparkApplication app = new SparkApplication(); + app.setMetadata( + new ObjectMetaBuilder() + .withName("spark-pi") + .withNamespace("spark-jobs") + .withUid("app-uid-123") + .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "team-a-queue")) + .build()); + + ApplicationSpec spec = new ApplicationSpec(); + spec.setSparkConf( + Map.of( + "spark.driver.cores", "2", + "spark.driver.memory", "2g", + "spark.executor.cores", "4", + "spark.executor.memory", "4g", + "spark.executor.instances", "3")); + app.setSpec(spec); + + Workload workload = KueueWorkloadFactory.buildWorkload(app); + assertNotNull(workload); + assertEquals("sparkapplication-spark-pi", workload.getMetadata().getName()); + assertEquals("spark-jobs", workload.getMetadata().getNamespace()); + assertEquals("team-a-queue", workload.getSpec().getQueueName()); + assertTrue(workload.getSpec().getActive()); + assertEquals(1, workload.getMetadata().getOwnerReferences().size()); + assertEquals("spark-pi", workload.getMetadata().getOwnerReferences().get(0).getName()); + assertTrue(workload.getMetadata().getOwnerReferences().get(0).getController()); + + assertEquals(2, workload.getSpec().getPodSets().size()); + + PodSet driverPodSet = workload.getSpec().getPodSets().get(0); + assertEquals("driver", driverPodSet.getName()); + assertEquals(1, driverPodSet.getCount()); + assertNull(driverPodSet.getMinCount()); + Container driverContainer = + driverPodSet.getTemplate().getSpec().getContainers().get(0); + assertEquals(new Quantity("2"), driverContainer.getResources().getRequests().get("cpu")); + assertEquals( + new Quantity("2432Mi"), + driverContainer.getResources().getRequests().get("memory")); // 2048 + 384 = 2432 + + PodSet executorPodSet = workload.getSpec().getPodSets().get(1); + assertEquals("executor", executorPodSet.getName()); + assertEquals(3, executorPodSet.getCount()); + assertNull(executorPodSet.getMinCount()); + Container executorContainer = + executorPodSet.getTemplate().getSpec().getContainers().get(0); + assertEquals(new Quantity("4"), executorContainer.getResources().getRequests().get("cpu")); + assertEquals( + new Quantity("4505Mi"), + executorContainer.getResources().getRequests().get("memory")); // 4096 + 409 = 4505 + } + + @Test + void testBuildWorkloadDefaultExecutorInstances() { + SparkApplication app = new SparkApplication(); + app.setMetadata( + new ObjectMetaBuilder().withName("spark-default").withNamespace("default").build()); + + Workload workload = KueueWorkloadFactory.buildWorkload(app); + PodSet executorPodSet = workload.getSpec().getPodSets().get(1); + assertEquals(2, executorPodSet.getCount()); + assertNull(executorPodSet.getMinCount()); + } + + @Test + void testBuildWorkloadDynamicAllocation() { + SparkApplication app = new SparkApplication(); + app.setMetadata( + new ObjectMetaBuilder() + .withName("spark-elastic") + .withNamespace("default") + .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "elastic-queue")) + .build()); + + ApplicationSpec spec = new ApplicationSpec(); + spec.setSparkConf( + Map.of( + "spark.dynamicAllocation.enabled", "true", + "spark.dynamicAllocation.minExecutors", "2", + "spark.dynamicAllocation.maxExecutors", "10")); + app.setSpec(spec); + + assertThrows( + UnsupportedOperationException.class, () -> KueueWorkloadFactory.buildWorkload(app)); + } + + @Test + void testBuildWorkloadForPythonApp() { + SparkApplication app = new SparkApplication(); + app.setMetadata( + new ObjectMetaBuilder().withName("pi-python").withNamespace("default").build()); + + ApplicationSpec spec = new ApplicationSpec(); + spec.setPyFiles("local:///opt/spark/examples/src/main/python/pi.py"); + spec.setSparkConf( + Map.of( + "spark.driver.memory", "8g", + "spark.executor.memory", "8g", + "spark.executor.pyspark.memory", "1g")); + app.setSpec(spec); + + Workload workload = KueueWorkloadFactory.buildWorkload(app); + Container driverContainer = + workload.getSpec().getPodSets().get(0).getTemplate().getSpec().getContainers().get(0); + assertEquals( + new Quantity("11468Mi"), + driverContainer.getResources().getRequests().get("memory")); // 8192 + 3276 = 11468 + Container executorContainer = + workload.getSpec().getPodSets().get(1).getTemplate().getSpec().getContainers().get(0); + assertEquals( + new Quantity("12492Mi"), + executorContainer.getResources().getRequests().get("memory")); // 8192 + 3276 + 1024 + } + + @Test + void testBuildWorkloadWithGpuAndTemplateSpec() { + SparkApplication app = new SparkApplication(); + app.setMetadata( + new ObjectMetaBuilder() + .withName("spark-gpu") + .withNamespace("default") + .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "gpu-queue")) + .build()); + + PodTemplateSpec driverTemplate = + new PodTemplateSpecBuilder() + .withNewSpec() + .withContainers( + new ContainerBuilder() + .withName("sidecar") + .withResources( + new ResourceRequirementsBuilder() + .withRequests(Map.of("cpu", new Quantity("100m"))) + .build()) + .build(), + new ContainerBuilder() + .withName("custom-driver") + .withResources( + new ResourceRequirementsBuilder() + .withRequests(Map.of("cpu", new Quantity("3"))) + .build()) + .build()) + .endSpec() + .build(); + + ApplicationSpec spec = new ApplicationSpec(); + spec.setDriverSpec(new BaseApplicationTemplateSpec(driverTemplate)); + spec.setSparkConf( + Map.of( + Constants.DRIVER_SPARK_CONTAINER_PROP_KEY, "custom-driver", Review Comment: **Finding 16.** This covers the driver key only. The executor path reads a bare `"spark.kubernetes.executor.podTemplateContainerName"` literal (`KueueWorkloadFactory.java:308`) with no constant and no test, so a typo there falls back to `containers.get(0)` and the whole executor request lands on a sidecar. Mirroring this test for the executor is a few lines: an `executorSpec` template with `sidecar` first and `custom-executor` second, `spark.kubernetes.executor.podTemplateContainerName: custom-executor`, then assert the computed `cpu` / `memory` landed on `custom-executor` and the sidecar's own `requests` are untouched. `spark.kubernetes.{driver,executor}.request.cores` is untested too, and it is the other branch of the same `getOrDefault` chain. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java: ########## @@ -0,0 +1,475 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodSpecBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSet; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.SparkClusterResourceSpec; +import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta1.WorkloadSpec; +import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.StringUtils; +import org.apache.spark.network.util.JavaUtils; + +/** + * Factory for creating Kueue Workload resources from Spark custom resources. + * This factory supports both {@link SparkApplication} (driver and executor pod sets) + * and {@link SparkCluster} (master and worker pod sets). + */ +@SuppressWarnings("PMD.GodClass") +public final class KueueWorkloadFactory { + + public static final String PODSET_DRIVER = "driver"; + public static final String PODSET_EXECUTOR = "executor"; + public static final String PODSET_MASTER = "master"; + public static final String PODSET_WORKER = "worker"; + + public static final String DEFAULT_CORES = "1"; + public static final String DEFAULT_MEMORY = "1g"; + public static final String DEFAULT_MIN_MEMORY_OVERHEAD = "384m"; + public static final double DEFAULT_MEMORY_OVERHEAD_FACTOR = 0.10; + public static final double NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.40; + public static final int DEFAULT_EXECUTOR_INSTANCES = 2; + + private KueueWorkloadFactory() {} + + /** + * Builds a Kueue Workload from a SparkApplication resource. + * + * @param app The SparkApplication. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkApplication app) { + String queueName = getQueueName(app); + ApplicationSpec appSpec = app.getSpec(); + Map<String, String> sparkConf = + appSpec != null && appSpec.getSparkConf() != null + ? appSpec.getSparkConf() + : Map.of(); + + PodSet driverPodSet = buildDriverPodSet(app, sparkConf); + PodSet executorPodSet = buildExecutorPodSet(app, sparkConf); + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(driverPodSet); + podSets.add(executorPodSet); + + boolean active = appSpec == null || !appSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (app.getMetadata().getLabels() != null) { + labels.putAll(app.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, app.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(app); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(app)) + .withNamespace(app.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + /** + * Builds a Kueue Workload from a SparkCluster resource. + * + * @param cluster The SparkCluster. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkCluster cluster) { + String queueName = getQueueName(cluster); + ClusterSpec clusterSpec = cluster.getSpec(); + + // Use the same StatefulSets which the operator creates for the cluster. + SparkClusterResourceSpec resourceSpec = + SparkClusterResourceSpecFactory.buildResourceSpec( + cluster, new SparkClusterSubmissionWorker()); + if (resourceSpec.getHorizontalPodAutoscaler().isPresent()) { + throw new UnsupportedOperationException( + "Kueue does not support SparkCluster with HorizontalPodAutoscaler " + + "(minWorkers < maxWorkers) yet."); + } + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(buildPodSet(PODSET_MASTER, resourceSpec.getMasterStatefulSet())); + podSets.add(buildPodSet(PODSET_WORKER, resourceSpec.getWorkerStatefulSet())); + + boolean active = clusterSpec == null || !clusterSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (cluster.getMetadata().getLabels() != null) { + labels.putAll(cluster.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_CLUSTER_NAME, cluster.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(cluster); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(cluster)) + .withNamespace(cluster.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + private static PodSet buildPodSet(final String name, final StatefulSet statefulSet) { + return PodSet.builder() + .name(name) + .count(statefulSet.getSpec().getReplicas()) + .template(statefulSet.getSpec().getTemplate()) + .build(); + } + + /** + * Returns the Workload name prefixed with the lower-cased kind of the owner resource, like Kueue + * built-in integrations, to avoid name collisions between SparkApplication and SparkCluster. + */ + private static String getWorkloadName(final HasMetadata resource) { + return resource.getKind().toLowerCase(Locale.ROOT) + "-" + resource.getMetadata().getName(); + } + + /** + * Extracts the Kueue queue name from the resource metadata labels. + * + * @param resource The Kubernetes resource (e.g. SparkApplication or SparkCluster). + * @return The queue name, or null if not found. + */ + public static String getQueueName(final HasMetadata resource) { + if (resource != null + && resource.getMetadata() != null + && resource.getMetadata().getLabels() != null) { + String queue = resource.getMetadata().getLabels().get(Constants.LABEL_QUEUE_NAME); + if (StringUtils.isNotEmpty(queue)) { + return queue; + } + } + return null; + } + + /** + * Checks whether the resource is configured to use Kueue. + * + * @param resource The Kubernetes resource (SparkApplication or SparkCluster). + * @return true if a Kueue queue name is specified. + */ + public static boolean hasQueueName(final HasMetadata resource) { + return StringUtils.isNotEmpty(getQueueName(resource)); + } + + /** + * Builds the driver PodSet. + */ + public static PodSet buildDriverPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getDriverSpec() != null + && app.getSpec().getDriverSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getDriverSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.driver.request.cores", + sparkConf.getOrDefault("spark.driver.cores", DEFAULT_CORES)); + long memoryMiB = calculateDriverMemoryMiB(sparkConf, isNonJvmApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.driver.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.driver.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get(Constants.DRIVER_SPARK_CONTAINER_PROP_KEY), + "spark-driver", + cpu, + memoryMiB, + gpuAmount, + gpuVendor); + + return PodSet.builder() + .name(PODSET_DRIVER) + .count(1) + .template(templateSpec) + .build(); + } + + /** + * Builds the executor PodSet. + */ + public static PodSet buildExecutorPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + if ("true".equalsIgnoreCase(sparkConf.get("spark.dynamicAllocation.enabled"))) { + throw new UnsupportedOperationException( + "Kueue does not support SparkApplication with dynamic allocation " + + "(spark.dynamicAllocation.enabled=true) yet."); + } + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getExecutorSpec() != null + && app.getSpec().getExecutorSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getExecutorSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.executor.request.cores", + sparkConf.getOrDefault("spark.executor.cores", DEFAULT_CORES)); + long memoryMiB = + calculateExecutorMemoryMiB( + sparkConf, isNonJvmApp(app.getSpec()), isPythonApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.executor.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.executor.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get("spark.kubernetes.executor.podTemplateContainerName"), + "spark-executor", + cpu, + memoryMiB, + gpuAmount, + gpuVendor); + + return PodSet.builder() + .name(PODSET_EXECUTOR) + .count(parseInt(sparkConf.get("spark.executor.instances"), DEFAULT_EXECUTOR_INSTANCES)) + .template(templateSpec) + .build(); + } + + /** + * Calculates total driver memory in MiB including overhead. + */ + public static long calculateDriverMemoryMiB( + final Map<String, String> sparkConf, final boolean isNonJvm) { + long memMiB = + JavaUtils.byteStringAsMb(sparkConf.getOrDefault("spark.driver.memory", DEFAULT_MEMORY)); + return memMiB + calculateMemoryOverheadMiB(sparkConf, "spark.driver", memMiB, isNonJvm); + } + + /** + * Calculates total executor memory in MiB including overhead, off-heap memory and PySpark + * memory. + */ + public static long calculateExecutorMemoryMiB( + final Map<String, String> sparkConf, final boolean isNonJvm, final boolean isPython) { + long memMiB = + JavaUtils.byteStringAsMb(sparkConf.getOrDefault("spark.executor.memory", DEFAULT_MEMORY)); + long total = + memMiB + calculateMemoryOverheadMiB(sparkConf, "spark.executor", memMiB, isNonJvm); + if ("true".equalsIgnoreCase(sparkConf.get("spark.memory.offHeap.enabled"))) { + // `spark.memory.offHeap.size` is in bytes unless otherwise specified. + total += + JavaUtils.byteStringAsBytes(sparkConf.getOrDefault("spark.memory.offHeap.size", "0")) + / 1024 + / 1024; + } + if (isPython && sparkConf.containsKey("spark.executor.pyspark.memory")) { + total += JavaUtils.byteStringAsMb(sparkConf.get("spark.executor.pyspark.memory")); + } + return total; + } + + private static long calculateMemoryOverheadMiB( + final Map<String, String> sparkConf, + final String prefix, + final long memMiB, + final boolean isNonJvm) { + String overhead = sparkConf.get(prefix + ".memoryOverhead"); + if (StringUtils.isNotEmpty(overhead)) { + return JavaUtils.byteStringAsMb(overhead); + } + // Like Spark's BasicDriverFeatureStep, the deprecated `spark.kubernetes.memoryOverheadFactor` + // or the default factor (0.4 for non-JVM applications) is used if not set explicitly. + double defaultFactor = + parseDouble( + sparkConf.get("spark.kubernetes.memoryOverheadFactor"), + isNonJvm ? NON_JVM_MEMORY_OVERHEAD_FACTOR : DEFAULT_MEMORY_OVERHEAD_FACTOR); + double factor = parseDouble(sparkConf.get(prefix + ".memoryOverheadFactor"), defaultFactor); + long minOverheadMiB = + JavaUtils.byteStringAsMb( + sparkConf.getOrDefault(prefix + ".minMemoryOverhead", DEFAULT_MIN_MEMORY_OVERHEAD)); + return Math.max((int) (factor * memMiB), minOverheadMiB); + } + + /** Follows the main application resource selection of SparkAppSubmissionWorker. */ + private static boolean isPythonApp(final ApplicationSpec spec) { + return StringUtils.isEmpty(spec.getJars()) + && ("org.apache.spark.deploy.PythonRunner".equals(spec.getMainClass()) + || StringUtils.isNotEmpty(spec.getPyFiles())); + } + + private static boolean isNonJvmApp(final ApplicationSpec spec) { + return isPythonApp(spec) + || (StringUtils.isEmpty(spec.getJars()) && StringUtils.isNotEmpty(spec.getSparkRFiles())); + } + + private static void ensurePodSpec(final PodTemplateSpec templateSpec) { + if (templateSpec.getSpec() == null) { + templateSpec.setSpec(new PodSpecBuilder().build()); + } + if (templateSpec.getSpec().getContainers() == null) { + templateSpec.getSpec().setContainers(new ArrayList<>()); + } + } + + private static void decorateTemplateResources( + final PodTemplateSpec templateSpec, + final String containerName, + final String defaultContainerName, + final String cpu, + final long memoryMiB, + final String gpuAmount, + final String gpuVendor) { + PodSpec podSpec = templateSpec.getSpec(); + Container container; + if (podSpec.getContainers().isEmpty()) { + container = new ContainerBuilder().withName(defaultContainerName).build(); + podSpec.getContainers().add(container); + } else { + // Like Spark's KubernetesUtils.selectSparkContainer, select the container by name and + // fall back to the first container. + container = + podSpec.getContainers().stream() + .filter(c -> containerName != null && containerName.equals(c.getName())) + .findFirst() + .orElse(podSpec.getContainers().get(0)); + } + + ResourceRequirements resources = container.getResources(); + if (resources == null) { + resources = new ResourceRequirementsBuilder().build(); + container.setResources(resources); + } + + Map<String, Quantity> requests = resources.getRequests(); + if (requests == null) { + requests = new HashMap<>(); + resources.setRequests(requests); + } + + // Like Spark, overwrite the requests of the pod template. + requests.put("cpu", new Quantity(cpu)); + requests.put("memory", new Quantity(memoryMiB + "Mi")); + + if (StringUtils.isNotEmpty(gpuAmount)) { + if (StringUtils.isEmpty(gpuVendor)) { + throw new IllegalArgumentException( + "Resource: gpu was requested, but vendor was not specified."); + } + // Like Spark's KubernetesConf.buildKubernetesResourceName, e.g., `nvidia.com/gpu`. + String gpuResourceName = gpuVendor + "/gpu"; + requests.put(gpuResourceName, new Quantity(gpuAmount)); + Map<String, Quantity> limits = resources.getLimits(); + if (limits == null) { + limits = new HashMap<>(); + resources.setLimits(limits); + } + limits.put(gpuResourceName, new Quantity(gpuAmount)); + } + } + + private static int parseInt(final String str, final int defaultValue) { Review Comment: **Finding 15.** Both helpers swallow a malformed value and return a plausible number, which is the failure mode finding 12 removed from the memory path. Measured on `0ddd85a`: spark.executor.instances=not-a-number -> executor PodSet count 2 spark.driver.memoryOverheadFactor=not-a-number -> 9011Mi (factor silently 0.1) Spark rejects both: `spark.executor.instances` is an `intConf` and `spark.driver.memoryOverheadFactor` is a `doubleConf` with a `> 0` check, so the driver never starts. Meanwhile `spark.driver.memory=10zz` throws here, and `testCalculateMemoryWithUnits` pins that. Letting `Integer.parseInt` / `Double.parseDouble` throw makes all three consistent. The `StringUtils.isEmpty` guard already covers the only input that legitimately needs a default. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java: ########## @@ -0,0 +1,475 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodSpecBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSet; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.SparkClusterResourceSpec; +import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta1.WorkloadSpec; +import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.StringUtils; +import org.apache.spark.network.util.JavaUtils; + +/** + * Factory for creating Kueue Workload resources from Spark custom resources. + * This factory supports both {@link SparkApplication} (driver and executor pod sets) + * and {@link SparkCluster} (master and worker pod sets). + */ +@SuppressWarnings("PMD.GodClass") +public final class KueueWorkloadFactory { + + public static final String PODSET_DRIVER = "driver"; + public static final String PODSET_EXECUTOR = "executor"; + public static final String PODSET_MASTER = "master"; + public static final String PODSET_WORKER = "worker"; + + public static final String DEFAULT_CORES = "1"; + public static final String DEFAULT_MEMORY = "1g"; + public static final String DEFAULT_MIN_MEMORY_OVERHEAD = "384m"; + public static final double DEFAULT_MEMORY_OVERHEAD_FACTOR = 0.10; + public static final double NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.40; + public static final int DEFAULT_EXECUTOR_INSTANCES = 2; + + private KueueWorkloadFactory() {} + + /** + * Builds a Kueue Workload from a SparkApplication resource. + * + * @param app The SparkApplication. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkApplication app) { + String queueName = getQueueName(app); + ApplicationSpec appSpec = app.getSpec(); + Map<String, String> sparkConf = + appSpec != null && appSpec.getSparkConf() != null + ? appSpec.getSparkConf() + : Map.of(); + + PodSet driverPodSet = buildDriverPodSet(app, sparkConf); + PodSet executorPodSet = buildExecutorPodSet(app, sparkConf); + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(driverPodSet); + podSets.add(executorPodSet); + + boolean active = appSpec == null || !appSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (app.getMetadata().getLabels() != null) { + labels.putAll(app.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, app.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(app); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(app)) + .withNamespace(app.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + /** + * Builds a Kueue Workload from a SparkCluster resource. + * + * @param cluster The SparkCluster. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkCluster cluster) { + String queueName = getQueueName(cluster); + ClusterSpec clusterSpec = cluster.getSpec(); + + // Use the same StatefulSets which the operator creates for the cluster. + SparkClusterResourceSpec resourceSpec = + SparkClusterResourceSpecFactory.buildResourceSpec( + cluster, new SparkClusterSubmissionWorker()); + if (resourceSpec.getHorizontalPodAutoscaler().isPresent()) { + throw new UnsupportedOperationException( + "Kueue does not support SparkCluster with HorizontalPodAutoscaler " + + "(minWorkers < maxWorkers) yet."); + } + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(buildPodSet(PODSET_MASTER, resourceSpec.getMasterStatefulSet())); + podSets.add(buildPodSet(PODSET_WORKER, resourceSpec.getWorkerStatefulSet())); + + boolean active = clusterSpec == null || !clusterSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (cluster.getMetadata().getLabels() != null) { + labels.putAll(cluster.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_CLUSTER_NAME, cluster.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(cluster); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(cluster)) + .withNamespace(cluster.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + private static PodSet buildPodSet(final String name, final StatefulSet statefulSet) { + return PodSet.builder() + .name(name) + .count(statefulSet.getSpec().getReplicas()) + .template(statefulSet.getSpec().getTemplate()) + .build(); + } + + /** + * Returns the Workload name prefixed with the lower-cased kind of the owner resource, like Kueue + * built-in integrations, to avoid name collisions between SparkApplication and SparkCluster. + */ + private static String getWorkloadName(final HasMetadata resource) { + return resource.getKind().toLowerCase(Locale.ROOT) + "-" + resource.getMetadata().getName(); + } + + /** + * Extracts the Kueue queue name from the resource metadata labels. + * + * @param resource The Kubernetes resource (e.g. SparkApplication or SparkCluster). + * @return The queue name, or null if not found. + */ + public static String getQueueName(final HasMetadata resource) { + if (resource != null + && resource.getMetadata() != null + && resource.getMetadata().getLabels() != null) { + String queue = resource.getMetadata().getLabels().get(Constants.LABEL_QUEUE_NAME); + if (StringUtils.isNotEmpty(queue)) { + return queue; + } + } + return null; + } + + /** + * Checks whether the resource is configured to use Kueue. + * + * @param resource The Kubernetes resource (SparkApplication or SparkCluster). + * @return true if a Kueue queue name is specified. + */ + public static boolean hasQueueName(final HasMetadata resource) { + return StringUtils.isNotEmpty(getQueueName(resource)); + } + + /** + * Builds the driver PodSet. + */ + public static PodSet buildDriverPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getDriverSpec() != null + && app.getSpec().getDriverSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getDriverSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.driver.request.cores", + sparkConf.getOrDefault("spark.driver.cores", DEFAULT_CORES)); + long memoryMiB = calculateDriverMemoryMiB(sparkConf, isNonJvmApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.driver.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.driver.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get(Constants.DRIVER_SPARK_CONTAINER_PROP_KEY), + "spark-driver", Review Comment: **Finding 18.** Spark's defaults are `spark-kubernetes-driver` and `spark-kubernetes-executor` (`Constants.scala:106-107` on `branch-4.2`); `BasicDriverFeatureStep` applies the first via `Option(pod.container.getName).getOrElse(DEFAULT_DRIVER_CONTAINER_NAME)`. The comment at line 412 says this block follows `KubernetesUtils.selectSparkContainer`, so the name it synthesizes should follow too. It is what shows up in the Workload a user reads, and what any later PodSet-update or topology code would match on. ```suggestion "spark-kubernetes-driver", ``` Line 309 needs the same change to `"spark-kubernetes-executor"`. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java: ########## @@ -0,0 +1,475 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodSpecBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSet; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.SparkClusterResourceSpec; +import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta1.WorkloadSpec; +import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.StringUtils; +import org.apache.spark.network.util.JavaUtils; + +/** + * Factory for creating Kueue Workload resources from Spark custom resources. + * This factory supports both {@link SparkApplication} (driver and executor pod sets) + * and {@link SparkCluster} (master and worker pod sets). + */ +@SuppressWarnings("PMD.GodClass") +public final class KueueWorkloadFactory { + + public static final String PODSET_DRIVER = "driver"; + public static final String PODSET_EXECUTOR = "executor"; + public static final String PODSET_MASTER = "master"; + public static final String PODSET_WORKER = "worker"; + + public static final String DEFAULT_CORES = "1"; + public static final String DEFAULT_MEMORY = "1g"; + public static final String DEFAULT_MIN_MEMORY_OVERHEAD = "384m"; + public static final double DEFAULT_MEMORY_OVERHEAD_FACTOR = 0.10; + public static final double NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.40; + public static final int DEFAULT_EXECUTOR_INSTANCES = 2; + + private KueueWorkloadFactory() {} + + /** + * Builds a Kueue Workload from a SparkApplication resource. + * + * @param app The SparkApplication. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkApplication app) { + String queueName = getQueueName(app); + ApplicationSpec appSpec = app.getSpec(); + Map<String, String> sparkConf = + appSpec != null && appSpec.getSparkConf() != null + ? appSpec.getSparkConf() + : Map.of(); + + PodSet driverPodSet = buildDriverPodSet(app, sparkConf); + PodSet executorPodSet = buildExecutorPodSet(app, sparkConf); + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(driverPodSet); + podSets.add(executorPodSet); + + boolean active = appSpec == null || !appSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (app.getMetadata().getLabels() != null) { + labels.putAll(app.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, app.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(app); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(app)) + .withNamespace(app.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + /** + * Builds a Kueue Workload from a SparkCluster resource. + * + * @param cluster The SparkCluster. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkCluster cluster) { + String queueName = getQueueName(cluster); + ClusterSpec clusterSpec = cluster.getSpec(); + + // Use the same StatefulSets which the operator creates for the cluster. + SparkClusterResourceSpec resourceSpec = + SparkClusterResourceSpecFactory.buildResourceSpec( + cluster, new SparkClusterSubmissionWorker()); + if (resourceSpec.getHorizontalPodAutoscaler().isPresent()) { + throw new UnsupportedOperationException( + "Kueue does not support SparkCluster with HorizontalPodAutoscaler " + + "(minWorkers < maxWorkers) yet."); + } + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(buildPodSet(PODSET_MASTER, resourceSpec.getMasterStatefulSet())); + podSets.add(buildPodSet(PODSET_WORKER, resourceSpec.getWorkerStatefulSet())); + + boolean active = clusterSpec == null || !clusterSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (cluster.getMetadata().getLabels() != null) { + labels.putAll(cluster.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_CLUSTER_NAME, cluster.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(cluster); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(cluster)) + .withNamespace(cluster.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + private static PodSet buildPodSet(final String name, final StatefulSet statefulSet) { + return PodSet.builder() + .name(name) + .count(statefulSet.getSpec().getReplicas()) + .template(statefulSet.getSpec().getTemplate()) + .build(); + } + + /** + * Returns the Workload name prefixed with the lower-cased kind of the owner resource, like Kueue + * built-in integrations, to avoid name collisions between SparkApplication and SparkCluster. + */ + private static String getWorkloadName(final HasMetadata resource) { + return resource.getKind().toLowerCase(Locale.ROOT) + "-" + resource.getMetadata().getName(); + } + + /** + * Extracts the Kueue queue name from the resource metadata labels. + * + * @param resource The Kubernetes resource (e.g. SparkApplication or SparkCluster). + * @return The queue name, or null if not found. + */ + public static String getQueueName(final HasMetadata resource) { + if (resource != null + && resource.getMetadata() != null + && resource.getMetadata().getLabels() != null) { + String queue = resource.getMetadata().getLabels().get(Constants.LABEL_QUEUE_NAME); + if (StringUtils.isNotEmpty(queue)) { + return queue; + } + } + return null; + } + + /** + * Checks whether the resource is configured to use Kueue. + * + * @param resource The Kubernetes resource (SparkApplication or SparkCluster). + * @return true if a Kueue queue name is specified. + */ + public static boolean hasQueueName(final HasMetadata resource) { + return StringUtils.isNotEmpty(getQueueName(resource)); + } + + /** + * Builds the driver PodSet. + */ + public static PodSet buildDriverPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getDriverSpec() != null + && app.getSpec().getDriverSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getDriverSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.driver.request.cores", + sparkConf.getOrDefault("spark.driver.cores", DEFAULT_CORES)); + long memoryMiB = calculateDriverMemoryMiB(sparkConf, isNonJvmApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.driver.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.driver.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get(Constants.DRIVER_SPARK_CONTAINER_PROP_KEY), + "spark-driver", + cpu, + memoryMiB, + gpuAmount, + gpuVendor); + + return PodSet.builder() + .name(PODSET_DRIVER) + .count(1) + .template(templateSpec) + .build(); + } + + /** + * Builds the executor PodSet. + */ + public static PodSet buildExecutorPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + if ("true".equalsIgnoreCase(sparkConf.get("spark.dynamicAllocation.enabled"))) { Review Comment: **Finding 19.** `buildDriverPodSet`, `buildExecutorPodSet`, `calculateDriverMemoryMiB` and `calculateExecutorMemoryMiB` are `public` only so `KueueWorkloadFactoryTest` can reach them, and that test is in the same package — package-private is enough, and it also settles the missing `@param` / `@return` tags on those four. That public surface is also why this guard sits here rather than at the top of `buildWorkload`. As written it runs after `buildDriverPodSet` has already built a PodSet, and `buildDriverPodSet` called on its own skips the check entirely. Moving it to the first line of `buildWorkload(SparkApplication)` puts it alongside the `SparkCluster` HPA rejection at line 148, so both unsupported cases are rejected in the same place. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java: ########## @@ -0,0 +1,475 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodSpecBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSet; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.SparkClusterResourceSpec; +import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta1.WorkloadSpec; +import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.StringUtils; +import org.apache.spark.network.util.JavaUtils; + +/** + * Factory for creating Kueue Workload resources from Spark custom resources. + * This factory supports both {@link SparkApplication} (driver and executor pod sets) + * and {@link SparkCluster} (master and worker pod sets). + */ +@SuppressWarnings("PMD.GodClass") +public final class KueueWorkloadFactory { + + public static final String PODSET_DRIVER = "driver"; + public static final String PODSET_EXECUTOR = "executor"; + public static final String PODSET_MASTER = "master"; + public static final String PODSET_WORKER = "worker"; + + public static final String DEFAULT_CORES = "1"; + public static final String DEFAULT_MEMORY = "1g"; + public static final String DEFAULT_MIN_MEMORY_OVERHEAD = "384m"; + public static final double DEFAULT_MEMORY_OVERHEAD_FACTOR = 0.10; + public static final double NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.40; + public static final int DEFAULT_EXECUTOR_INSTANCES = 2; + + private KueueWorkloadFactory() {} + + /** + * Builds a Kueue Workload from a SparkApplication resource. + * + * @param app The SparkApplication. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkApplication app) { + String queueName = getQueueName(app); + ApplicationSpec appSpec = app.getSpec(); + Map<String, String> sparkConf = + appSpec != null && appSpec.getSparkConf() != null + ? appSpec.getSparkConf() + : Map.of(); + + PodSet driverPodSet = buildDriverPodSet(app, sparkConf); + PodSet executorPodSet = buildExecutorPodSet(app, sparkConf); + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(driverPodSet); + podSets.add(executorPodSet); + + boolean active = appSpec == null || !appSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (app.getMetadata().getLabels() != null) { + labels.putAll(app.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, app.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(app); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(app)) + .withNamespace(app.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + /** + * Builds a Kueue Workload from a SparkCluster resource. + * + * @param cluster The SparkCluster. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkCluster cluster) { + String queueName = getQueueName(cluster); + ClusterSpec clusterSpec = cluster.getSpec(); + + // Use the same StatefulSets which the operator creates for the cluster. + SparkClusterResourceSpec resourceSpec = + SparkClusterResourceSpecFactory.buildResourceSpec( + cluster, new SparkClusterSubmissionWorker()); + if (resourceSpec.getHorizontalPodAutoscaler().isPresent()) { + throw new UnsupportedOperationException( + "Kueue does not support SparkCluster with HorizontalPodAutoscaler " + + "(minWorkers < maxWorkers) yet."); + } + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(buildPodSet(PODSET_MASTER, resourceSpec.getMasterStatefulSet())); + podSets.add(buildPodSet(PODSET_WORKER, resourceSpec.getWorkerStatefulSet())); + + boolean active = clusterSpec == null || !clusterSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (cluster.getMetadata().getLabels() != null) { + labels.putAll(cluster.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_CLUSTER_NAME, cluster.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); + } + + OwnerReference ownerReference = ModelUtils.buildOwnerReferenceTo(cluster); + ownerReference.setController(true); + + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(getWorkloadName(cluster)) + .withNamespace(cluster.getMetadata().getNamespace()) + .withLabels(labels) + .withOwnerReferences(ownerReference) + .build()); + + workload.setSpec( + WorkloadSpec.builder() + .queueName(queueName) + .active(active) + .podSets(podSets) + .build()); + + return workload; + } + + private static PodSet buildPodSet(final String name, final StatefulSet statefulSet) { + return PodSet.builder() + .name(name) + .count(statefulSet.getSpec().getReplicas()) + .template(statefulSet.getSpec().getTemplate()) + .build(); + } + + /** + * Returns the Workload name prefixed with the lower-cased kind of the owner resource, like Kueue + * built-in integrations, to avoid name collisions between SparkApplication and SparkCluster. + */ + private static String getWorkloadName(final HasMetadata resource) { + return resource.getKind().toLowerCase(Locale.ROOT) + "-" + resource.getMetadata().getName(); + } + + /** + * Extracts the Kueue queue name from the resource metadata labels. + * + * @param resource The Kubernetes resource (e.g. SparkApplication or SparkCluster). + * @return The queue name, or null if not found. + */ + public static String getQueueName(final HasMetadata resource) { + if (resource != null + && resource.getMetadata() != null + && resource.getMetadata().getLabels() != null) { + String queue = resource.getMetadata().getLabels().get(Constants.LABEL_QUEUE_NAME); + if (StringUtils.isNotEmpty(queue)) { + return queue; + } + } + return null; + } + + /** + * Checks whether the resource is configured to use Kueue. + * + * @param resource The Kubernetes resource (SparkApplication or SparkCluster). + * @return true if a Kueue queue name is specified. + */ + public static boolean hasQueueName(final HasMetadata resource) { + return StringUtils.isNotEmpty(getQueueName(resource)); + } + + /** + * Builds the driver PodSet. + */ + public static PodSet buildDriverPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getDriverSpec() != null + && app.getSpec().getDriverSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getDriverSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.driver.request.cores", + sparkConf.getOrDefault("spark.driver.cores", DEFAULT_CORES)); + long memoryMiB = calculateDriverMemoryMiB(sparkConf, isNonJvmApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.driver.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.driver.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get(Constants.DRIVER_SPARK_CONTAINER_PROP_KEY), + "spark-driver", + cpu, + memoryMiB, + gpuAmount, + gpuVendor); + + return PodSet.builder() + .name(PODSET_DRIVER) + .count(1) + .template(templateSpec) + .build(); + } + + /** + * Builds the executor PodSet. + */ + public static PodSet buildExecutorPodSet( + final SparkApplication app, final Map<String, String> sparkConf) { + if ("true".equalsIgnoreCase(sparkConf.get("spark.dynamicAllocation.enabled"))) { + throw new UnsupportedOperationException( + "Kueue does not support SparkApplication with dynamic allocation " + + "(spark.dynamicAllocation.enabled=true) yet."); + } + PodTemplateSpec templateSpec = null; + if (app.getSpec() != null + && app.getSpec().getExecutorSpec() != null + && app.getSpec().getExecutorSpec().getPodTemplateSpec() != null) { + templateSpec = ReconcilerUtils.clone(app.getSpec().getExecutorSpec().getPodTemplateSpec()); + } + if (templateSpec == null) { + templateSpec = new PodTemplateSpecBuilder().build(); + } + ensurePodSpec(templateSpec); + + String cpu = + sparkConf.getOrDefault( + "spark.kubernetes.executor.request.cores", + sparkConf.getOrDefault("spark.executor.cores", DEFAULT_CORES)); + long memoryMiB = + calculateExecutorMemoryMiB( + sparkConf, isNonJvmApp(app.getSpec()), isPythonApp(app.getSpec())); + String gpuAmount = sparkConf.get("spark.executor.resource.gpu.amount"); + String gpuVendor = sparkConf.get("spark.executor.resource.gpu.vendor"); + + decorateTemplateResources( + templateSpec, + sparkConf.get("spark.kubernetes.executor.podTemplateContainerName"), Review Comment: **Finding 17.** The driver reads this from `Constants.DRIVER_SPARK_CONTAINER_PROP_KEY` at line 261. `Constants` already carries the driver and executor `podTemplateFile` keys as a pair (`spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java:93-99`), so the executor container-name key is the one member missing from that set. ```java /** The property key for the executor Spark container name. */ public static final String EXECUTOR_SPARK_CONTAINER_PROP_KEY = "spark.kubernetes.executor.podTemplateContainerName"; ``` ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java: ########## @@ -0,0 +1,475 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.k8s.operator.kueue; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ContainerBuilder; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodSpecBuilder; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder; +import io.fabric8.kubernetes.api.model.Quantity; +import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.apps.StatefulSet; + +import org.apache.spark.k8s.operator.Constants; +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.SparkCluster; +import org.apache.spark.k8s.operator.SparkClusterResourceSpec; +import org.apache.spark.k8s.operator.SparkClusterSubmissionWorker; +import org.apache.spark.k8s.operator.kueue.v1beta1.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta1.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta1.WorkloadSpec; +import org.apache.spark.k8s.operator.reconciler.SparkClusterResourceSpecFactory; +import org.apache.spark.k8s.operator.spec.ApplicationSpec; +import org.apache.spark.k8s.operator.spec.ClusterSpec; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; +import org.apache.spark.k8s.operator.utils.StringUtils; +import org.apache.spark.network.util.JavaUtils; + +/** + * Factory for creating Kueue Workload resources from Spark custom resources. + * This factory supports both {@link SparkApplication} (driver and executor pod sets) + * and {@link SparkCluster} (master and worker pod sets). + */ +@SuppressWarnings("PMD.GodClass") +public final class KueueWorkloadFactory { + + public static final String PODSET_DRIVER = "driver"; + public static final String PODSET_EXECUTOR = "executor"; + public static final String PODSET_MASTER = "master"; + public static final String PODSET_WORKER = "worker"; + + public static final String DEFAULT_CORES = "1"; + public static final String DEFAULT_MEMORY = "1g"; + public static final String DEFAULT_MIN_MEMORY_OVERHEAD = "384m"; + public static final double DEFAULT_MEMORY_OVERHEAD_FACTOR = 0.10; + public static final double NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.40; + public static final int DEFAULT_EXECUTOR_INSTANCES = 2; + + private KueueWorkloadFactory() {} + + /** + * Builds a Kueue Workload from a SparkApplication resource. + * + * @param app The SparkApplication. + * @return The constructed Kueue Workload. + */ + public static Workload buildWorkload(final SparkApplication app) { + String queueName = getQueueName(app); + ApplicationSpec appSpec = app.getSpec(); + Map<String, String> sparkConf = + appSpec != null && appSpec.getSparkConf() != null + ? appSpec.getSparkConf() + : Map.of(); + + PodSet driverPodSet = buildDriverPodSet(app, sparkConf); + PodSet executorPodSet = buildExecutorPodSet(app, sparkConf); + + List<PodSet> podSets = new ArrayList<>(); + podSets.add(driverPodSet); + podSets.add(executorPodSet); + + boolean active = appSpec == null || !appSpec.isSuspend(); + + Map<String, String> labels = new HashMap<>(); + if (app.getMetadata().getLabels() != null) { + labels.putAll(app.getMetadata().getLabels()); + } + labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, app.getMetadata().getName()); + if (StringUtils.isNotEmpty(queueName)) { + labels.put(Constants.LABEL_QUEUE_NAME, queueName); Review Comment: **Finding 20.** This `put` cannot change anything. `queueName` was read from `Constants.LABEL_QUEUE_NAME` on this same object (line 87 into line 217), and line 105 already copied every one of the CR's labels. Line 166 is the same for `SparkCluster`. Dropping both leaves `labels` identical. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
