fix(sparkconnect): use server.cores as server pod CPU request

api/openapi-spec/swagger.json

@@ -12688,7 +12688,7 @@
             ]
           },
           "cores": {
-            "description": "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.",
+            "description": "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.",
             "type": "integer",
             "format": "int32"
           },
@@ -12761,7 +12761,7 @@
             ]
           },
           "cores": {
-            "description": "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.",
+            "description": "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.",
             "type": "integer",
             "format": "int32"
           },
@@ -13044,7 +13044,7 @@
             ]
           },
           "cores": {
-            "description": "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.",
+            "description": "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.",
             "type": "integer",
             "format": "int32"
           },

api/python_api/kubeflow_spark_api/models/spark_v1alpha1_executor_spec.py

@@ -45,7 +45,7 @@ class SparkV1alpha1ExecutorSpec(BaseModel):
     """ # noqa: E501
     core_limit: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreLimit is the physical CPU core limit for the pod.", alias="coreLimit")
     core_request: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreRequest is the physical CPU core request for the pod.", alias="coreRequest")
-    cores: Optional[StrictInt] = Field(default=None, description="Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.")
+    cores: Optional[StrictInt] = Field(default=None, description="Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.")
     gpu: Optional[SparkV1alpha1GPUSpec] = Field(default=None, description="GPU specifies GPU resources for the pod and Spark resource scheduler. GPU discovery and per-task resource settings are configured through SparkConf.")
     instances: Optional[StrictInt] = Field(default=None, description="Instances is the number of executor instances.")
     memory: Optional[StrictStr] = Field(default=None, description="Memory is the amount of memory to request for the pod.")

api/python_api/kubeflow_spark_api/models/spark_v1alpha1_server_spec.py

@@ -46,7 +46,7 @@ class SparkV1alpha1ServerSpec(BaseModel):
     """ # noqa: E501
     core_limit: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreLimit is the physical CPU core limit for the pod.", alias="coreLimit")
     core_request: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreRequest is the physical CPU core request for the pod.", alias="coreRequest")
-    cores: Optional[StrictInt] = Field(default=None, description="Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.")
+    cores: Optional[StrictInt] = Field(default=None, description="Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.")
     gpu: Optional[SparkV1alpha1GPUSpec] = Field(default=None, description="GPU specifies GPU resources for the pod and Spark resource scheduler. GPU discovery and per-task resource settings are configured through SparkConf.")
     memory: Optional[StrictStr] = Field(default=None, description="Memory is the amount of memory to request for the pod.")
     service: Optional[IoK8sApiCoreV1Service] = Field(default=None, description="Service exposes the Spark connect server.")

api/python_api/kubeflow_spark_api/models/spark_v1alpha1_spark_pod_spec.py

@@ -45,7 +45,7 @@ class SparkV1alpha1SparkPodSpec(BaseModel):
     """ # noqa: E501
     core_limit: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreLimit is the physical CPU core limit for the pod.", alias="coreLimit")
     core_request: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreRequest is the physical CPU core request for the pod.", alias="coreRequest")
-    cores: Optional[StrictInt] = Field(default=None, description="Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.")
+    cores: Optional[StrictInt] = Field(default=None, description="Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.")
     gpu: Optional[SparkV1alpha1GPUSpec] = Field(default=None, description="GPU specifies GPU resources for the pod and Spark resource scheduler. GPU discovery and per-task resource settings are configured through SparkConf.")
     memory: Optional[StrictStr] = Field(default=None, description="Memory is the amount of memory to request for the pod.")
     template: Optional[IoK8sApiCoreV1PodTemplateSpec] = Field(default=None, description="Template is a pod template that can be used to define the driver or executor pod configurations that Spark configurations do not support. Spark version >= 3.0.0 is required. Ref: https://spark.apache.org/docs/latest/running-on-kubernetes.html#pod-template.")

api/v1alpha1/sparkconnect_types.go

@@ -111,7 +111,7 @@ type ExecutorSpec struct {
 
 // SparkPodSpec defines common things that can be customized for a Spark driver or executor pod.
 type SparkPodSpec struct {
-	// Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.
+	// Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.
 	// +optional
 	// +kubebuilder:validation:Minimum=1
 	Cores *int32 `json:"cores,omitempty"`

charts/spark-operator-chart/crds/sparkoperator.k8s.io_sparkconnects.yaml

@@ -159,8 +159,8 @@ spec:
                     pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
                     x-kubernetes-int-or-string: true
                   cores:
-                    description: Cores maps to `spark.driver.cores` or `spark.executor.cores`
-                      for the driver and executors, respectively.
+                    description: Cores is a shorthand for coreRequest on the server,
+                      and maps to `spark.executor.cores` for executors.
                     format: int32
                     minimum: 1
                     type: integer
@@ -236,8 +236,8 @@ spec:
                     pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
                     x-kubernetes-int-or-string: true
                   cores:
-                    description: Cores maps to `spark.driver.cores` or `spark.executor.cores`
-                      for the driver and executors, respectively.
+                    description: Cores is a shorthand for coreRequest on the server,
+                      and maps to `spark.executor.cores` for executors.
                     format: int32
                     minimum: 1
                     type: integer

config/crd/bases/sparkoperator.k8s.io_sparkconnects.yaml

@@ -159,8 +159,8 @@ spec:
                     pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
                     x-kubernetes-int-or-string: true
                   cores:
-                    description: Cores maps to `spark.driver.cores` or `spark.executor.cores`
-                      for the driver and executors, respectively.
+                    description: Cores is a shorthand for coreRequest on the server,
+                      and maps to `spark.executor.cores` for executors.
                     format: int32
                     minimum: 1
                     type: integer
@@ -236,8 +236,8 @@ spec:
                     pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
                     x-kubernetes-int-or-string: true
                   cores:
-                    description: Cores maps to `spark.driver.cores` or `spark.executor.cores`
-                      for the driver and executors, respectively.
+                    description: Cores is a shorthand for coreRequest on the server,
+                      and maps to `spark.executor.cores` for executors.
                     format: int32
                     minimum: 1
                     type: integer

docs/api-docs.md

@@ -926,7 +926,7 @@ int32
 </td>
 <td>
 <em>(Optional)</em>
-<p>Cores maps to <code>spark.driver.cores</code> or <code>spark.executor.cores</code> for the driver and executors, respectively.</p>
+<p>Cores is a shorthand for coreRequest on the server, and maps to <code>spark.executor.cores</code> for executors.</p>
 </td>
 </tr>
 <tr>

docs/website/user-guide/spark-connect.md

@@ -113,7 +113,8 @@ The main `SparkConnect` fields are:
 | --- | --- |
 | `.spec.sparkVersion` | Declares the Spark version used by the server. |
 | `.spec.image` | Sets a common server and executor image when templates do not provide component-specific images. |
-| `.spec.server.cores` and `.spec.server.memory` | Set the server's Spark CPU and memory settings, passed as `spark.driver.cores` and `spark.driver.memory`. |
+| `.spec.server.cores` | Shorthand for `.spec.server.coreRequest` in whole CPUs; `coreRequest` takes precedence when both are set. |
+| `.spec.server.memory` | Sets the server's Spark memory setting, passed as `spark.driver.memory`. |
 | `.spec.server.coreRequest` and `.spec.server.coreLimit` | Set the physical Kubernetes CPU request and limit on the container of the operator-created server pod. |
 | `.spec.server.template` | Customizes the server pod, including its image, service account, volumes, and security context. If no service account is set here, the operator falls back to the `--default-service-account` controller flag (`controller.defaultServiceAccount` in the Helm chart). |
 | `.spec.executor.instances` | Sets the number of static executors. |
@@ -172,18 +173,24 @@ for the full rules.
 
 ## CPU request and limit
 
-`cores` and `coreRequest` / `coreLimit` are independent. `cores` is the number of
-Spark task slots, while `coreRequest` and `coreLimit` are the physical Kubernetes
-CPU request and limit on the container, so setting `cores: 2` together with
-`coreRequest: 500m` is valid and common.
+`coreRequest` and `coreLimit` are the physical Kubernetes CPU request and limit on
+the container. On executors, `cores` is independent of them: it is the number of
+Spark task slots, so setting `cores: 2` together with `coreRequest: 500m` is valid
+and common.
+
+On the server, `cores` is a shorthand for `coreRequest` in whole CPUs. The server
+runs as the Spark driver in client mode, where `spark.driver.cores` has no effect,
+so the operator does not pass it to Spark. When both `cores` and `coreRequest` are
+set, `coreRequest` wins.
 
 The two are applied differently, because the server and executor pods are created
 by different actors:
 
 - **Server** — the operator creates the server pod, so
-  `.spec.server.coreRequest` and `.spec.server.coreLimit` are written directly onto
-  its container's `resources`. A CPU value in `.spec.server.template` is used only
-  where the corresponding field is unset.
+  `.spec.server.coreRequest` (or `.spec.server.cores` when `coreRequest` is unset)
+  and `.spec.server.coreLimit` are written directly onto its container's
+  `resources`. A CPU value in `.spec.server.template` is used only where the
+  corresponding field is unset.
 - **Executor** — Spark creates the executor pods. The operator passes the fields to
   Spark as `spark.kubernetes.executor.request.cores` and
   `spark.kubernetes.executor.limit.cores`, and Spark sets the container resources

examples/sparkconnect/spark-connect-custom-resource.yaml

@@ -41,7 +41,8 @@ spec:
           resources:
             requests:
               # Overridden by `spec.server.coreRequest` above. Remove that field
-              # and this value becomes the server pod's CPU request.
+              # and `spec.server.cores`, and this value becomes the server pod's
+              # CPU request.
               cpu: 100m
               # Not overridden: the operator only manages the server pod's CPU,
               # so this memory request is applied to the pod as-is.

internal/controller/sparkconnect/options.go

@@ -196,12 +196,6 @@ func driverConfOption(conn *v1alpha1.SparkConnect) ([]string, error) {
 		args = append(args, "--conf", fmt.Sprintf("%s=%s", property, value))
 	}
 
-	// Driver cores
-	if conn.Spec.Server.Cores != nil {
-		args = append(args, "--conf",
-			fmt.Sprintf("%s=%d", common.SparkDriverCores, *conn.Spec.Server.Cores))
-	}
-
 	// Driver memory
 	if conn.Spec.Server.Memory != nil {
 		args = append(args, "--conf",

internal/controller/sparkconnect/options_test.go

@@ -131,7 +131,7 @@ var _ = Describe("Options functions", func() {
 	})
 
 	Context("driverConfOption and executorConfOption with CPU resources", func() {
-		It("does not emit driver SparkConf keys for server CoreRequest and CoreLimit", func() {
+		It("does not emit driver CPU SparkConf keys for server Cores, CoreRequest and CoreLimit", func() {
 			cores := int32(4)
 			coreRequest := resource.MustParse("3500m")
 			coreLimit := resource.MustParse("4")
@@ -157,7 +157,7 @@ var _ = Describe("Options functions", func() {
 			Expect(err).NotTo(HaveOccurred())
 			config := parsedSparkConfig(args)
 
-			Expect(config).To(HaveKeyWithValue("spark.driver.cores", "4"))
+			Expect(config).NotTo(HaveKey(common.SparkDriverCores))
 			Expect(config).NotTo(HaveKey(common.SparkKubernetesDriverRequestCores))
 			Expect(config).NotTo(HaveKey(common.SparkKubernetesDriverLimitCores))
 		})
@@ -217,7 +217,7 @@ var _ = Describe("Options functions", func() {
 			Expect(err).NotTo(HaveOccurred())
 			driverConfig := parsedSparkConfig(driverArgs)
 
-			Expect(driverConfig).To(HaveKeyWithValue("spark.driver.cores", "4"))
+			Expect(driverConfig).NotTo(HaveKey(common.SparkDriverCores))
 			Expect(driverConfig).NotTo(HaveKey(common.SparkKubernetesDriverRequestCores))
 			Expect(driverConfig).NotTo(HaveKey(common.SparkKubernetesDriverLimitCores))
 		})

internal/controller/sparkconnect/reconciler.go

@@ -397,9 +397,10 @@ func (r *Reconciler) mutateServerPod(ctx context.Context, conn *v1alpha1.SparkCo
 		)
 
 		// Setup Kubernetes CPU resources for the Connect server container.
+		// spark.driver.cores has no effect because the Connect server runs in client mode.
 		container = util.SetContainerCPUResources(
 			container,
-			conn.Spec.Server.CoreRequest,
+			util.CPURequestOrCores(conn.Spec.Server.CoreRequest, conn.Spec.Server.Cores),
 			conn.Spec.Server.CoreLimit,
 		)
 

internal/controller/sparkconnect/reconciler_test.go

@@ -537,7 +537,7 @@ var _ = Describe("mutateServerPod with CPU resources", func() {
 		Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(2000)))
 	})
 
-	It("keeps spec.server.cores independent from the Kubernetes CPU resources", func() {
+	It("prefers server.coreRequest over server.cores for the CPU request", func() {
 		cores := int32(4)
 		coreRequest := resource.MustParse("500m")
 		conn.Spec.Server.Cores = &cores
@@ -555,6 +555,73 @@ var _ = Describe("mutateServerPod with CPU resources", func() {
 		Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(500)))
 	})
 
+	It("uses server.cores as the CPU request when server.coreRequest is omitted", func() {
+		cores := int32(2)
+		conn.Spec.Server.Cores = &cores
+		pod := &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: conn.Namespace,
+			},
+		}
+
+		err := reconciler.mutateServerPod(context.TODO(), conn, pod)
+		Expect(err).NotTo(HaveOccurred())
+
+		container := pod.Spec.Containers[0]
+		Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(2000)))
+		Expect(container.Resources.Limits).NotTo(HaveKey(corev1.ResourceCPU))
+	})
+
+	It("overrides template CPU request with server.cores when server.coreRequest is omitted", func() {
+		cores := int32(2)
+		conn.Spec.Server.Cores = &cores
+		conn.Spec.Server.Template = &corev1.PodTemplateSpec{
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name:  common.SparkDriverContainerName,
+						Image: image,
+						Resources: corev1.ResourceRequirements{
+							Requests: corev1.ResourceList{
+								corev1.ResourceCPU: resource.MustParse("1"),
+							},
+							Limits: corev1.ResourceList{
+								corev1.ResourceCPU: resource.MustParse("3"),
+							},
+						},
+					},
+				},
+			},
+		}
+		pod := &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: conn.Namespace,
+			},
+		}
+
+		err := reconciler.mutateServerPod(context.TODO(), conn, pod)
+		Expect(err).NotTo(HaveOccurred())
+
+		container := pod.Spec.Containers[0]
+		Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(2000)))
+		Expect(cpuMilliValue(container.Resources.Limits[corev1.ResourceCPU])).To(Equal(int64(3000)))
+	})
+
+	It("does not pass spark.driver.cores to the server", func() {
+		cores := int32(2)
+		conn.Spec.Server.Cores = &cores
+		pod := &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: conn.Namespace,
+			},
+		}
+
+		err := reconciler.mutateServerPod(context.TODO(), conn, pod)
+		Expect(err).NotTo(HaveOccurred())
+
+		Expect(pod.Spec.Containers[0].Args).NotTo(ContainElement(ContainSubstring(common.SparkDriverCores)))
+	})
+
 	It("does not mutate the server pod template", func() {
 		coreRequest := resource.MustParse("500m")
 		conn.Spec.Server.CoreRequest = &coreRequest

internal/webhook/sparkconnect_validator.go

@@ -431,7 +431,7 @@ func validateCPURequestLELimit(request, limit *resource.Quantity) error {
 }
 
 func effectiveServerCPUResources(sc *v1alpha1.SparkConnect) (request, limit *resource.Quantity) {
-	request = sc.Spec.Server.CoreRequest
+	request = util.CPURequestOrCores(sc.Spec.Server.CoreRequest, sc.Spec.Server.Cores)
 	limit = sc.Spec.Server.CoreLimit
 
 	if template := sc.Spec.Server.Template; template != nil {