add proxy user support (kubeflow#1022)

Signed-off-by: Abhilash Pallerlamudi <[email protected]>

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>

manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml

@@ -3613,6 +3613,8 @@ spec:
                     additionalProperties:
                       type: string
                     type: object
+                  proxyUser:
+                    type: string
                   pythonVersion:
                     enum:
                     - "2"

manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml

@@ -3599,6 +3599,8 @@ spec:
                 additionalProperties:
                   type: string
                 type: object
+              proxyUser:
+                type: string
               pythonVersion:
                 enum:
                 - "2"

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

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))

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])
+}