Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #1022 (2020-09-16 19:59:00)

add proxy user support
This PR adds proxy user support for impersonation.

Signed-off-by: Abhilash Pallerlamudi <stp.abhi@gmail.com>

jinxingwang: Can you please try to add a test in the submission_test.go?
jinxingwang: Please link the spark-submit doc about --proxy-user, thanks.
stpabhi: @jinxingwang  Added test case. Unfortunately, I couldn't find the ```--proxy-user``` arg in spark documentation. I hope the spark-submit code reference help is sufficient. Check [here](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala#L529).
stpabhi: @jinxingwang can it be merged?

Patch:
diff --git a/docs/api-docs.md b/docs/api-docs.md
index a0ff575..ee6eb4a 100644
--- a/docs/api-docs.md
+++ b/docs/api-docs.md
@@ -259,6 +259,19 @@ DeployMode
 </tr>
 <tr>
 <td>
+<code>proxyUser</code></br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ProxyUser specifies the user to impersonate when submitting the application.
+It maps to the command-line flag &ldquo;&ndash;proxy-user&rdquo; in spark-submit.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>image</code></br>
 <em>
 string
@@ -709,6 +722,21 @@ string
 <p>PriorityClassName stands for the name of k8s PriorityClass resource, it&rsquo;s being used in Volcano batch scheduler.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>resources</code></br>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.13/#resourcelist-v1-core">
+Kubernetes core/v1.ResourceList
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Resources stands for the resource list custom request for. Usually it is used to define the lower-bound limit.
+If specified, volcano scheduler will consider it as the resources requested.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="sparkoperator.k8s.io/v1beta2.ConcurrencyPolicy">ConcurrencyPolicy
@@ -1840,6 +1868,19 @@ DeployMode
 </tr>
 <tr>
 <td>
+<code>proxyUser</code></br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ProxyUser specifies the user to impersonate when submitting the application.
+It maps to the command-line flag &ldquo;&ndash;proxy-user&rdquo; in spark-submit.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>image</code></br>
 <em>
 string
@@ -2746,5 +2787,5 @@ map[string]string
 <hr/>
 <p><em>
 Generated with <code>gen-crd-api-reference-docs</code>
-on git commit <code>555c27a</code>.
+on git commit <code>bc7bbd0</code>.
 </em></p>
diff --git a/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml b/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml
index f1b68cd..23b6d4c 100644
--- a/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml
+++ b/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml
@@ -3613,6 +3613,8 @@ spec:
                     additionalProperties:
                       type: string
                     type: object
+                  proxyUser:
+                    type: string
                   pythonVersion:
                     enum:
                     - "2"
diff --git a/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml b/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml
index 04765f5..e61ac11 100644
--- a/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml
+++ b/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml
@@ -3599,6 +3599,8 @@ spec:
                 additionalProperties:
                   type: string
                 type: object
+              proxyUser:
+                type: string
               pythonVersion:
                 enum:
                 - "2"
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
index 032e215..30b2637 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
@@ -187,6 +187,10 @@ type SparkApplicationSpec struct {
 	// Mode is the deployment mode of the Spark application.
 	// +kubebuilder:validation:Enum={cluster,client}
 	Mode DeployMode `json:"mode,omitempty"`
+	// ProxyUser specifies the user to impersonate when submitting the application.
+	// It maps to the command-line flag "--proxy-user" in spark-submit.
+	// +optional
+	ProxyUser *string `json:"proxyUser,omitempty"`
 	// Image is the container image for the driver, executor, and init-container. Any custom container images for the
 	// driver, executor, or init-container takes precedence over this.
 	// +optional
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go
index c0e014d..eb3d8c8 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go
@@ -593,6 +593,11 @@ func (in *SparkApplicationList) DeepCopyObject() runtime.Object {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *SparkApplicationSpec) DeepCopyInto(out *SparkApplicationSpec) {
 	*out = *in
+	if in.ProxyUser != nil {
+		in, out := &in.ProxyUser, &out.ProxyUser
+		*out = new(string)
+		**out = **in
+	}
 	if in.Image != nil {
 		in, out := &in.Image, &out.Image
 		*out = new(string)
diff --git a/pkg/controller/sparkapplication/submission.go b/pkg/controller/sparkapplication/submission.go
index cebf3b0..5531b43 100644
--- a/pkg/controller/sparkapplication/submission.go
+++ b/pkg/controller/sparkapplication/submission.go
@@ -96,6 +96,12 @@ func buildSubmissionCommandArgs(app *v1beta2.SparkApplication, driverPodName str
 
 	args = append(args, "--master", masterURL)
 	args = append(args, "--deploy-mode", string(app.Spec.Mode))
+
+	// Add proxy user
+	if app.Spec.ProxyUser != nil {
+		args = append(args, "--proxy-user", *app.Spec.ProxyUser)
+	}
+
 	args = append(args, "--conf", fmt.Sprintf("%s=%s", config.SparkAppNamespaceKey, app.Namespace))
 	args = append(args, "--conf", fmt.Sprintf("%s=%s", config.SparkAppNameKey, app.Name))
 	args = append(args, "--conf", fmt.Sprintf("%s=%s", config.SparkDriverPodNameKey, driverPodName))


Test patch:
diff --git a/pkg/controller/sparkapplication/submission_test.go b/pkg/controller/sparkapplication/submission_test.go
index e00015b..9f32bbb 100644
--- a/pkg/controller/sparkapplication/submission_test.go
+++ b/pkg/controller/sparkapplication/submission_test.go
@@ -18,6 +18,7 @@ package sparkapplication
 
 import (
 	"fmt"
+	"os"
 	"reflect"
 	"sort"
 	"strconv"
@@ -543,3 +544,42 @@ func TestDynamicAllocationOptions(t *testing.T) {
 	assert.Equal(t, fmt.Sprintf("%s=10", config.SparkDynamicAllocationMaxExecutors), options[4])
 	assert.Equal(t, fmt.Sprintf("%s=6000000", config.SparkDynamicAllocationShuffleTrackingTimeout), options[5])
 }
+
+func TestProxyUserArg(t *testing.T) {
+	const (
+		host = "localhost"
+		port = "6443"
+	)
+
+	if err := os.Setenv(kubernetesServiceHostEnvVar, host); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Setenv(kubernetesServicePortEnvVar, port); err != nil {
+		t.Fatal(err)
+	}
+
+	app := &v1beta2.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-test",
+			UID:  "spark-test-1",
+		},
+		Spec: v1beta2.SparkApplicationSpec{
+			Mode:      v1beta2.ClusterMode,
+			ProxyUser: stringptr("foo"),
+		},
+	}
+
+	submissionID := uuid.New().String()
+	driverPodName := getDriverPodName(app)
+	args, err := buildSubmissionCommandArgs(app, driverPodName, submissionID)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Equal(t, "--master", args[0])
+	assert.Equal(t, fmt.Sprintf("k8s://https://%s:%s", host, port), args[1])
+	assert.Equal(t, "--deploy-mode", args[2])
+	assert.Equal(t, string(v1beta2.ClusterMode), args[3])
+	assert.Equal(t, "--proxy-user", args[4])
+	assert.Equal(t, "foo", args[5])
+}