Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #1057 (2020-11-10 00:06:00)

Default Logic: Add SparkConf check
Only default Driver/Executor resources if not set both in Driver/Executor Spec and in SparkConf.

Ref: https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/issues/1055 for details 


Patch:
diff --git a/docs/api-docs.md b/docs/api-docs.md
index 3581b25..9f2faf6 100644
--- a/docs/api-docs.md
+++ b/docs/api-docs.md
@@ -719,7 +719,8 @@ string
 </td>
 <td>
 <em>(Optional)</em>
-<p>PriorityClassName stands for the name of k8s PriorityClass resource, it&rsquo;s being used in Volcano batch scheduler.</p>
+<p>PriorityClassName stands for the name of k8s PriorityClass resource, it&rsquo;s being used for job scheduling and 
+preemption, whether with the Volcano batch scheduler or the Kubernetes default scheduler.</p>
 </td>
 </tr>
 <tr>
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta1/defaults.go b/pkg/apis/sparkoperator.k8s.io/v1beta1/defaults.go
index 63fba60..43eb5d0 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta1/defaults.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta1/defaults.go
@@ -43,31 +43,31 @@ func SetSparkApplicationDefaults(app *SparkApplication) {
 		}
 	}
 
-	setDriverSpecDefaults(app.Spec.Driver)
-	setExecutorSpecDefaults(app.Spec.Executor)
+	setDriverSpecDefaults(&app.Spec.Driver, app.Spec.SparkConf)
+	setExecutorSpecDefaults(&app.Spec.Executor, app.Spec.SparkConf)
 }
 
-func setDriverSpecDefaults(spec DriverSpec) {
-	if spec.Cores == nil {
+func setDriverSpecDefaults(spec *DriverSpec, sparkConf map[string]string) {
+	if _, exists := sparkConf["spark.driver.cores"]; !exists && spec.Cores == nil {
 		spec.Cores = new(float32)
 		*spec.Cores = 1
 	}
-	if spec.Memory == nil {
+	if _, exists := sparkConf["spark.driver.memory"]; !exists && spec.Memory == nil {
 		spec.Memory = new(string)
 		*spec.Memory = "1g"
 	}
 }
 
-func setExecutorSpecDefaults(spec ExecutorSpec) {
-	if spec.Cores == nil {
+func setExecutorSpecDefaults(spec *ExecutorSpec, sparkConf map[string]string) {
+	if _, exists := sparkConf["spark.executor.cores"]; !exists && spec.Cores == nil {
 		spec.Cores = new(float32)
 		*spec.Cores = 1
 	}
-	if spec.Memory == nil {
+	if _, exists := sparkConf["spark.executor.memory"]; !exists && spec.Memory == nil {
 		spec.Memory = new(string)
 		*spec.Memory = "1g"
 	}
-	if spec.Instances == nil {
+	if _, exists := sparkConf["spark.executor.instances"]; !exists && spec.Instances == nil {
 		spec.Instances = new(int32)
 		*spec.Instances = 1
 	}
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults.go
index 3dae561..f722a36 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults.go
@@ -43,31 +43,32 @@ func SetSparkApplicationDefaults(app *SparkApplication) {
 		}
 	}
 
-	setDriverSpecDefaults(&app.Spec.Driver)
-	setExecutorSpecDefaults(&app.Spec.Executor)
+	setDriverSpecDefaults(&app.Spec.Driver, app.Spec.SparkConf)
+	setExecutorSpecDefaults(&app.Spec.Executor, app.Spec.SparkConf)
 }
 
-func setDriverSpecDefaults(spec *DriverSpec) {
-	if spec.Cores == nil {
+func setDriverSpecDefaults(spec *DriverSpec, sparkConf map[string]string) {
+
+	if _, exists := sparkConf["spark.driver.cores"]; !exists && spec.Cores == nil {
 		spec.Cores = new(int32)
 		*spec.Cores = 1
 	}
-	if spec.Memory == nil {
+	if _, exists := sparkConf["spark.driver.memory"]; !exists && spec.Memory == nil {
 		spec.Memory = new(string)
 		*spec.Memory = "1g"
 	}
 }
 
-func setExecutorSpecDefaults(spec *ExecutorSpec) {
-	if spec.Cores == nil {
+func setExecutorSpecDefaults(spec *ExecutorSpec, sparkConf map[string]string) {
+	if _, exists := sparkConf["spark.executor.cores"]; !exists && spec.Cores == nil {
 		spec.Cores = new(int32)
 		*spec.Cores = 1
 	}
-	if spec.Memory == nil {
+	if _, exists := sparkConf["spark.executor.memory"]; !exists && spec.Memory == nil {
 		spec.Memory = new(string)
 		*spec.Memory = "1g"
 	}
-	if spec.Instances == nil {
+	if _, exists := sparkConf["spark.executor.instances"]; !exists && spec.Instances == nil {
 		spec.Instances = new(int32)
 		*spec.Instances = 1
 	}
diff --git a/pkg/webhook/patch.go b/pkg/webhook/patch.go
index 10c49df..b4f5d1f 100644
--- a/pkg/webhook/patch.go
+++ b/pkg/webhook/patch.go
@@ -68,6 +68,11 @@ func patchSparkPod(pod *corev1.Pod, app *v1beta2.SparkApplication) []patchOperat
 		patchOps = append(patchOps, *op)
 	}
 
+	op = addPriorityClassName(pod, app)
+	if op != nil {
+		patchOps = append(patchOps, *op)
+	}
+
 	if pod.Spec.Affinity == nil {
 		op := addAffinity(pod, app)
 		if op != nil {
@@ -518,6 +523,19 @@ func addSchedulerName(pod *corev1.Pod, app *v1beta2.SparkApplication) *patchOper
 	return &patchOperation{Op: "add", Path: "/spec/schedulerName", Value: *schedulerName}
 }
 
+func addPriorityClassName(pod *corev1.Pod, app *v1beta2.SparkApplication) *patchOperation {
+	var priorityClassName *string
+
+	if app.Spec.BatchSchedulerOptions != nil {
+		priorityClassName = app.Spec.BatchSchedulerOptions.PriorityClassName
+	}
+
+	if priorityClassName == nil || *priorityClassName == "" {
+		return nil
+	}
+	return &patchOperation{Op: "add", Path: "/spec/priorityClassName", Value: *priorityClassName}
+}
+
 func addToleration(pod *corev1.Pod, toleration corev1.Toleration, first bool) patchOperation {
 	path := "/spec/tolerations"
 	var value interface{}


Test patch:
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults_test.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults_test.go
index fa38aaa..093e4ef 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults_test.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/defaults_test.go
@@ -121,6 +121,8 @@ func TestSetSparkApplicationDefaultsOnFailureRestartPolicyShouldSetDefaultValueF
 }
 
 func TestSetSparkApplicationDefaultslDriverSpecDefaults(t *testing.T) {
+
+	//Case1: Driver config not set.
 	app := &SparkApplication{
 		Spec: SparkApplicationSpec{},
 	}
@@ -138,9 +140,24 @@ func TestSetSparkApplicationDefaultslDriverSpecDefaults(t *testing.T) {
 	} else {
 		assert.Equal(t, "1g", *app.Spec.Driver.Memory)
 	}
+
+	//Case2: Driver config set via SparkConf.
+	app = &SparkApplication{
+		Spec: SparkApplicationSpec{
+			SparkConf: map[string]string{
+				"spark.driver.memory": "200M",
+				"spark.driver.cores":  "1",
+			},
+		},
+	}
+	SetSparkApplicationDefaults(app)
+
+	assert.Nil(t, app.Spec.Driver.Cores)
+	assert.Nil(t, app.Spec.Driver.Memory)
 }
 
 func TestSetSparkApplicationDefaultslExecutorSpecDefaults(t *testing.T) {
+	//Case1: Executor config not set.
 	app := &SparkApplication{
 		Spec: SparkApplicationSpec{},
 	}
@@ -164,4 +181,22 @@ func TestSetSparkApplicationDefaultslExecutorSpecDefaults(t *testing.T) {
 	} else {
 		assert.Equal(t, int32(1), *app.Spec.Executor.Instances)
 	}
+
+	//Case2: Executor config set via SparkConf.
+	app = &SparkApplication{
+		Spec: SparkApplicationSpec{
+			SparkConf: map[string]string{
+				"spark.executor.cores":     "2",
+				"spark.executor.memory":    "500M",
+				"spark.executor.instances": "3",
+			},
+		},
+	}
+
+	SetSparkApplicationDefaults(app)
+
+	assert.Nil(t, app.Spec.Executor.Cores)
+	assert.Nil(t, app.Spec.Executor.Memory)
+	assert.Nil(t, app.Spec.Executor.Instances)
+
 }
diff --git a/pkg/webhook/patch_test.go b/pkg/webhook/patch_test.go
index 74f859d..027356f 100644
--- a/pkg/webhook/patch_test.go
+++ b/pkg/webhook/patch_test.go
@@ -772,6 +772,78 @@ func TestPatchSparkPod_SchedulerName(t *testing.T) {
 	assert.Equal(t, defaultScheduler, modifiedExecutorPod.Spec.SchedulerName)
 }
 
+func TestPatchSparkPod_PriorityClassName(t *testing.T) {
+	var priorityClassName = "critical"
+
+	app := &v1beta2.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-test-patch-priorityclassname",
+			UID:  "spark-test-1",
+		},
+		Spec: v1beta2.SparkApplicationSpec{
+			BatchSchedulerOptions: &v1beta2.BatchSchedulerConfiguration{
+				PriorityClassName: &priorityClassName,
+			},
+			Driver: v1beta2.DriverSpec{
+				SparkPodSpec: v1beta2.SparkPodSpec{},
+			},
+			Executor: v1beta2.ExecutorSpec{
+				SparkPodSpec: v1beta2.SparkPodSpec{},
+			},
+		},
+	}
+
+	driverPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-driver",
+			Labels: map[string]string{
+				config.SparkRoleLabel:               config.SparkDriverRole,
+				config.LaunchedBySparkOperatorLabel: "true",
+			},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:  config.SparkDriverContainerName,
+					Image: "spark-driver:latest",
+				},
+			},
+		},
+	}
+
+	modifiedDriverPod, err := getModifiedPod(driverPod, app)
+	if err != nil {
+		t.Fatal(err)
+	}
+	//Driver priorityClassName should be populated when specified
+	assert.Equal(t, priorityClassName, modifiedDriverPod.Spec.PriorityClassName)
+
+	executorPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-executor",
+			Labels: map[string]string{
+				config.SparkRoleLabel:               config.SparkExecutorRole,
+				config.LaunchedBySparkOperatorLabel: "true",
+			},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:  config.SparkExecutorContainerName,
+					Image: "spark-executor:latest",
+				},
+			},
+		},
+	}
+
+	modifiedExecutorPod, err := getModifiedPod(executorPod, app)
+	if err != nil {
+		t.Fatal(err)
+	}
+	//Executor priorityClassName should also be populated when specified in SparkApplicationSpec
+	assert.Equal(t, priorityClassName, modifiedExecutorPod.Spec.PriorityClassName)
+}
+
 func TestPatchSparkPod_Sidecars(t *testing.T) {
 	app := &v1beta2.SparkApplication{
 		ObjectMeta: metav1.ObjectMeta{