Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #1506 (2022-04-02 06:29:37)

Implement --logs and --delete flags on 'sparkctl create' and a timeout on 'sparkctl log' to wait a pod startup
This PR added the following features:

- New flags on `sparkctl create`
  - `--logs`: watch SparkApplication logs until finished
  - `--delete`: delete the SparkApplication if already exists
- Timeout on `sparkctl log` to wait the pod startup and produce a log


Patch:
diff --git a/charts/spark-operator-chart/Chart.yaml b/charts/spark-operator-chart/Chart.yaml
index ab6ed61..53bbadd 100644
--- a/charts/spark-operator-chart/Chart.yaml
+++ b/charts/spark-operator-chart/Chart.yaml
@@ -1,8 +1,8 @@
 apiVersion: v2
 name: spark-operator
 description: A Helm chart for Spark on Kubernetes operator
-version: 1.1.19
-appVersion: v1beta2-1.3.3-3.1.1
+version: 1.1.21
+appVersion: v1beta2-1.3.5-3.1.1
 keywords:
   - spark
 home: https://github.com/GoogleCloudPlatform/spark-on-k8s-operator
diff --git a/docs/user-guide.md b/docs/user-guide.md
index ef93d6e..fc47058 100644
--- a/docs/user-guide.md
+++ b/docs/user-guide.md
@@ -128,9 +128,9 @@ There are two ways to add Spark configuration: setting individual Spark configur
 ```yaml
 spec:
   sparkConf:
-    "spark.ui.port": "4045"
-    "spark.eventLog.enabled": "true"
-    "spark.eventLog.dir": "hdfs://hdfs-namenode-1:8020/spark/spark-events"
+    spark.ui.port: "4045"
+    spark.eventLog.enabled: "true"
+    spark.eventLog.dir": "hdfs://hdfs-namenode-1:8020/spark/spark-events"
 ```
 
 ### Specifying Hadoop Configuration
diff --git a/main.go b/main.go
index b2ddbba..080af47 100644
--- a/main.go
+++ b/main.go
@@ -73,6 +73,7 @@ var (
 	metricsPort                    = flag.String("metrics-port", "10254", "Port for the metrics endpoint.")
 	metricsEndpoint                = flag.String("metrics-endpoint", "/metrics", "Metrics endpoint.")
 	metricsPrefix                  = flag.String("metrics-prefix", "", "Prefix for the metrics.")
+	ingressClassName               = flag.String("ingress-class-name", "", "Set ingressClassName for ingress resources created.")
 	metricsLabels                  util.ArrayFlags
 	metricsJobStartLatencyBuckets  util.HistogramBuckets = util.DefaultJobStartLatencyBuckets
 )
@@ -184,7 +185,7 @@ func main() {
 	}
 
 	applicationController := sparkapplication.NewController(
-		crClient, kubeClient, crInformerFactory, podInformerFactory, metricConfig, *namespace, *ingressURLFormat, batchSchedulerMgr, *enableUIService)
+		crClient, kubeClient, crInformerFactory, podInformerFactory, metricConfig, *namespace, *ingressURLFormat, *ingressClassName, batchSchedulerMgr, *enableUIService)
 	scheduledApplicationController := scheduledsparkapplication.NewController(
 		crClient, kubeClient, apiExtensionsClient, crInformerFactory, clock.RealClock{})
 
diff --git a/manifest/spark-operator-with-webhook-install/spark-operator-with-webhook.yaml b/manifest/spark-operator-with-webhook-install/spark-operator-with-webhook.yaml
new file mode 100644
index 0000000..5e4f318
--- /dev/null
+++ b/manifest/spark-operator-with-webhook-install/spark-operator-with-webhook.yaml
@@ -0,0 +1,96 @@
+#
+# Copyright 2017 Google LLC
+#
+# Licensed 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
+#
+#     https://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. 
+#
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: sparkoperator
+  namespace: spark-operator
+  labels:
+    app.kubernetes.io/name: sparkoperator
+    app.kubernetes.io/version: v2.4.0-v1beta1
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: sparkoperator
+      app.kubernetes.io/version: v2.4.0-v1beta1
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: sparkoperator
+        app.kubernetes.io/version: v2.4.0-v1beta1
+      initializers:
+        pending: []
+    spec:
+      serviceAccountName: sparkoperator
+      volumes:
+        - name: webhook-certs
+          secret:
+            secretName: spark-webhook-certs
+      containers:
+        - name: sparkoperator
+          image: gcr.io/spark-operator/spark-operator:v2.4.0-v1beta1-latest
+          imagePullPolicy: Always
+          volumeMounts:
+            - name: webhook-certs
+              mountPath: /etc/webhook-certs
+          ports:
+            - containerPort: 8080
+          args:
+            - -logtostderr
+            - -enable-webhook=true
+            - -v=2
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: sparkoperator-init
+  namespace: spark-operator
+  labels:
+    app.kubernetes.io/name: sparkoperator
+    app.kubernetes.io/version: v2.4.0-v1beta1
+spec:
+  backoffLimit: 3
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: sparkoperator
+        app.kubernetes.io/version: v2.4.0-v1beta1
+    spec:
+      serviceAccountName: sparkoperator
+      restartPolicy: Never
+      containers:
+        - name: main
+          image: gcr.io/spark-operator/spark-operator:v2.4.0-v1beta1-latest
+          imagePullPolicy: IfNotPresent
+          command: ["/usr/bin/gencerts.sh", "-p"]
+---
+kind: Service
+apiVersion: v1
+metadata:
+  name: spark-webhook
+  namespace: spark-operator
+spec:
+  ports:
+    - port: 443
+      targetPort: 8080
+      name: webhook
+  selector:
+    app.kubernetes.io/name: sparkoperator
+    app.kubernetes.io/version: v2.4.0-v1beta1
diff --git a/pkg/controller/sparkapplication/controller.go b/pkg/controller/sparkapplication/controller.go
index 81c3634..4740fd3 100644
--- a/pkg/controller/sparkapplication/controller.go
+++ b/pkg/controller/sparkapplication/controller.go
@@ -76,6 +76,7 @@ type Controller struct {
 	applicationLister crdlisters.SparkApplicationLister
 	podLister         v1.PodLister
 	ingressURLFormat  string
+	ingressClassName  string
 	batchSchedulerMgr *batchscheduler.SchedulerManager
 	enableUIService   bool
 }
@@ -89,6 +90,7 @@ func NewController(
 	metricsConfig *util.MetricConfig,
 	namespace string,
 	ingressURLFormat string,
+	ingressClassName string,
 	batchSchedulerMgr *batchscheduler.SchedulerManager,
 	enableUIService bool) *Controller {
 	crdscheme.AddToScheme(scheme.Scheme)
@@ -100,7 +102,7 @@ func NewController(
 	})
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{Component: "spark-operator"})
 
-	return newSparkApplicationController(crdClient, kubeClient, crdInformerFactory, podInformerFactory, recorder, metricsConfig, ingressURLFormat, batchSchedulerMgr, enableUIService)
+	return newSparkApplicationController(crdClient, kubeClient, crdInformerFactory, podInformerFactory, recorder, metricsConfig, ingressURLFormat, ingressClassName, batchSchedulerMgr, enableUIService)
 }
 
 func newSparkApplicationController(
@@ -111,6 +113,7 @@ func newSparkApplicationController(
 	eventRecorder record.EventRecorder,
 	metricsConfig *util.MetricConfig,
 	ingressURLFormat string,
+	ingressClassName string,
 	batchSchedulerMgr *batchscheduler.SchedulerManager,
 	enableUIService bool) *Controller {
 	queue := workqueue.NewNamedRateLimitingQueue(&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(queueTokenRefillRate), queueTokenBucketSize)},
@@ -122,6 +125,7 @@ func newSparkApplicationController(
 		recorder:          eventRecorder,
 		queue:             queue,
 		ingressURLFormat:  ingressURLFormat,
+		ingressClassName:  ingressClassName,
 		batchSchedulerMgr: batchSchedulerMgr,
 		enableUIService:   enableUIService,
 	}
@@ -575,7 +579,15 @@ func (c *Controller) syncSparkApplication(key string) error {
 			appCopy.Status.AppState.State = v1beta2.FailedState
 			c.recordSparkApplicationEvent(appCopy)
 		} else if isNextRetryDue(appCopy.Spec.RestartPolicy.OnSubmissionFailureRetryInterval, appCopy.Status.SubmissionAttempts, appCopy.Status.LastSubmissionAttemptTime) {
-			appCopy = c.submitSparkApplication(appCopy)
+			if c.validateSparkResourceDeletion(appCopy) {
+				c.submitSparkApplication(appCopy)
+			} else {
+				if err := c.deleteSparkResources(appCopy); err != nil {
+					glog.Errorf("failed to delete resources associated with SparkApplication %s/%s: %v",
+						appCopy.Namespace, appCopy.Name, err)
+					return err
+				}
+			}
 		}
 	case v1beta2.InvalidatingState:
 		// Invalidate the current run and enqueue the SparkApplication for re-execution.
@@ -664,14 +676,16 @@ func (c *Controller) submitSparkApplication(app *v1beta2.SparkApplication) *v1be
 		}
 	}
 
+	driverInfo := v1beta2.DriverInfo{}
+
 	if c.enableUIService {
 		service, err := createSparkUIService(app, c.kubeClient)
 		if err != nil {
 			glog.Errorf("failed to create UI service for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
 		} else {
-			app.Status.DriverInfo.WebUIServiceName = service.serviceName
-			app.Status.DriverInfo.WebUIPort = service.servicePort
-			app.Status.DriverInfo.WebUIAddress = fmt.Sprintf("%s:%d", service.serviceIP, app.Status.DriverInfo.WebUIPort)
+			driverInfo.WebUIServiceName = service.serviceName
+			driverInfo.WebUIPort = service.servicePort
+			driverInfo.WebUIAddress = fmt.Sprintf("%s:%d", service.serviceIP, app.Status.DriverInfo.WebUIPort)
 			// Create UI Ingress if ingress-format is set.
 			if c.ingressURLFormat != "" {
 				// We are going to want to use an ingress url.
@@ -687,12 +701,12 @@ func (c *Controller) submitSparkApplication(app *v1beta2.SparkApplication) *v1be
 						app.Spec.SparkConf["spark.ui.proxyBase"] = ingressURL.Path
 						app.Spec.SparkConf["spark.ui.proxyRedirectUri"] = "/"
 					}
-					ingress, err := createSparkUIIngress(app, *service, ingressURL, c.kubeClient)
+					ingress, err := createSparkUIIngress(app, *service, ingressURL, c.ingressClassName, c.kubeClient)
 					if err != nil {
 						glog.Errorf("failed to create UI Ingress for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
 					} else {
-						app.Status.DriverInfo.WebUIIngressAddress = ingress.ingressURL.String()
-						app.Status.DriverInfo.WebUIIngressName = ingress.ingressName
+						driverInfo.WebUIIngressAddress = ingress.ingressURL.String()
+						driverInfo.WebUIIngressName = ingress.ingressName
 					}
 				}
 			}
@@ -700,6 +714,7 @@ func (c *Controller) submitSparkApplication(app *v1beta2.SparkApplication) *v1be
 	}
 
 	driverPodName := getDriverPodName(app)
+	driverInfo.PodName = driverPodName
 	submissionID := uuid.New().String()
 	submissionCmdArgs, err := buildSubmissionCommandArgs(app, driverPodName, submissionID)
 	if err != nil {
@@ -741,9 +756,7 @@ func (c *Controller) submitSparkApplication(app *v1beta2.SparkApplication) *v1be
 		AppState: v1beta2.ApplicationState{
 			State: v1beta2.SubmittedState,
 		},
-		DriverInfo: v1beta2.DriverInfo{
-			PodName: driverPodName,
-		},
+		DriverInfo:                driverInfo,
 		SubmissionAttempts:        app.Status.SubmissionAttempts + 1,
 		ExecutionAttempts:         app.Status.ExecutionAttempts + 1,
 		LastSubmissionAttemptTime: metav1.Now(),
diff --git a/pkg/controller/sparkapplication/sparkui.go b/pkg/controller/sparkapplication/sparkui.go
index 2a0d09f..c3c856a 100644
--- a/pkg/controller/sparkapplication/sparkui.go
+++ b/pkg/controller/sparkapplication/sparkui.go
@@ -75,21 +75,22 @@ type SparkService struct {
 
 // SparkIngress encapsulates information about the driver UI ingress.
 type SparkIngress struct {
-	ingressName string
-	ingressURL  *url.URL
-	annotations map[string]string
-	ingressTLS  []networkingv1.IngressTLS
+	ingressName      string
+	ingressURL       *url.URL
+	ingressClassName string
+	annotations      map[string]string
+	ingressTLS       []networkingv1.IngressTLS
 }
 
-func createSparkUIIngress(app *v1beta2.SparkApplication, service SparkService, ingressURL *url.URL, kubeClient clientset.Interface) (*SparkIngress, error) {
+func createSparkUIIngress(app *v1beta2.SparkApplication, service SparkService, ingressURL *url.URL, ingressClassName string, kubeClient clientset.Interface) (*SparkIngress, error) {
 	if util.IngressCapabilities.Has("networking.k8s.io/v1") {
-		return createSparkUIIngress_v1(app, service, ingressURL, kubeClient)
+		return createSparkUIIngress_v1(app, service, ingressURL, ingressClassName, kubeClient)
 	} else {
 		return createSparkUIIngress_legacy(app, service, ingressURL, kubeClient)
 	}
 }
 
-func createSparkUIIngress_v1(app *v1beta2.SparkApplication, service SparkService, ingressURL *url.URL, kubeClient clientset.Interface) (*SparkIngress, error) {
+func createSparkUIIngress_v1(app *v1beta2.SparkApplication, service SparkService, ingressURL *url.URL, ingressClassName string, kubeClient clientset.Interface) (*SparkIngress, error) {
 	ingressResourceAnnotations := getIngressResourceAnnotations(app)
 	ingressTlsHosts := getIngressTlsHosts(app)
 
@@ -145,16 +146,21 @@ func createSparkUIIngress_v1(app *v1beta2.SparkApplication, service SparkService
 	if len(ingressTlsHosts) != 0 {
 		ingress.Spec.TLS = ingressTlsHosts
 	}
+	if len(ingressClassName) != 0 {
+		ingress.Spec.IngressClassName = &ingressClassName
+	}
+
 	glog.Infof("Creating an Ingress %s for the Spark UI for application %s", ingress.Name, app.Name)
 	_, err := kubeClient.NetworkingV1().Ingresses(ingress.Namespace).Create(context.TODO(), &ingress, metav1.CreateOptions{})
 	if err != nil {
 		return nil, err
 	}
 	return &SparkIngress{
-		ingressName: ingress.Name,
-		ingressURL:  ingressURL,
-		annotations: ingress.Annotations,
-		ingressTLS:  ingressTlsHosts,
+		ingressName:      ingress.Name,
+		ingressURL:       ingressURL,
+		ingressClassName: ingressClassName,
+		annotations:      ingress.Annotations,
+		ingressTLS:       ingressTlsHosts,
 	}, nil
 }
 
diff --git a/sparkctl/cmd/create.go b/sparkctl/cmd/create.go
index 318829a..21b56aa 100644
--- a/sparkctl/cmd/create.go
+++ b/sparkctl/cmd/create.go
@@ -41,6 +41,8 @@ import (
 
 const bufferSize = 1024
 
+var DeleteIfExists bool
+var LogsEnabled bool
 var RootPath string
 var UploadToPath string
 var UploadToEndpoint string
@@ -90,6 +92,10 @@ var createCmd = &cobra.Command{
 }
 
 func init() {
+	createCmd.Flags().BoolVarP(&DeleteIfExists, "delete", "d", false,
+		"delete the SparkApplication if already exists")
+	createCmd.Flags().BoolVarP(&LogsEnabled, "logs", "l", false,
+		"watch the SparkApplication logs")
 	createCmd.Flags().StringVarP(&UploadToPath, "upload-to", "u", "",
 		"the name of the bucket where local application dependencies are to be uploaded")
 	createCmd.Flags().StringVarP(&RootPath, "upload-prefix", "p", "",
@@ -151,6 +157,10 @@ func createFromScheduledSparkApplication(name string, kubeClient clientset.Inter
 }
 
 func createSparkApplication(app *v1beta2.SparkApplication, kubeClient clientset.Interface, crdClient crdclientset.Interface) error {
+	if DeleteIfExists {
+		deleteSparkApplication(app.Name, crdClient)
+	}
+
 	v1beta2.SetSparkApplicationDefaults(app)
 	if err := validateSpec(app.Spec); err != nil {
 		return err
@@ -177,6 +187,10 @@ func createSparkApplication(app *v1beta2.SparkApplication, kubeClient clientset.
 
 	fmt.Printf("SparkApplication \"%s\" created\n", app.Name)
 
+	if LogsEnabled {
+		doLog(app.Name, true, kubeClient, crdClient)
+	}
+
 	return nil
 }
 
diff --git a/sparkctl/cmd/delete.go b/sparkctl/cmd/delete.go
index bd70301..293fc34 100644
--- a/sparkctl/cmd/delete.go
+++ b/sparkctl/cmd/delete.go
@@ -51,8 +51,7 @@ var deleteCmd = &cobra.Command{
 }
 
 func doDelete(name string, crdClientset crdclientset.Interface) error {
-	err := crdClientset.SparkoperatorV1beta2().SparkApplications(Namespace).Delete(context.TODO(), name, metav1.DeleteOptions{})
-	if err != nil {
+	if err := deleteSparkApplication(name, crdClientset); err != nil {
 		return err
 	}
 
@@ -60,3 +59,7 @@ func doDelete(name string, crdClientset crdclientset.Interface) error {
 
 	return nil
 }
+
+func deleteSparkApplication(name string, crdClientset crdclientset.Interface) error {
+	return crdClientset.SparkoperatorV1beta2().SparkApplications(Namespace).Delete(context.TODO(), name, metav1.DeleteOptions{})
+}
diff --git a/sparkctl/cmd/log.go b/sparkctl/cmd/log.go
index 8d61f42..f7ce655 100644
--- a/sparkctl/cmd/log.go
+++ b/sparkctl/cmd/log.go
@@ -21,7 +21,7 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"strings"
+	"time"
 
 	"github.com/spf13/cobra"
 
@@ -57,7 +57,7 @@ var logCommand = &cobra.Command{
 			return
 		}
 
-		if err := doLog(args[0], kubeClientset, crdClientset); err != nil {
+		if err := doLog(args[0], FollowLogs, kubeClientset, crdClientset); err != nil {
 			fmt.Fprintf(os.Stderr, "failed to get driver logs of SparkApplication %s: %v\n", args[0], err)
 		}
 	},
@@ -69,36 +69,76 @@ func init() {
 	logCommand.Flags().BoolVarP(&FollowLogs, "follow", "f", false, "whether to stream the logs")
 }
 
-func doLog(name string, kubeClientset clientset.Interface, crdClientset crdclientset.Interface) error {
-	app, err := crdClientset.SparkoperatorV1beta2().SparkApplications(Namespace).Get(context.TODO(), name, metav1.GetOptions{})
-	if err != nil {
-		return fmt.Errorf("failed to get SparkApplication %s: %v", name, err)
-	}
+func doLog(
+	name string,
+	followLogs bool,
+	kubeClient clientset.Interface,
+	crdClient crdclientset.Interface) error {
 
+	timeout := 30 * time.Second
+
+	podNameChannel := getPodNameChannel(name, crdClient)
 	var podName string
-	if ExecutorId < 0 {
-		podName = app.Status.DriverInfo.PodName
-	} else {
-		podName = strings.NewReplacer("driver", fmt.Sprintf("exec-%d", ExecutorId)).
-			Replace(app.Status.DriverInfo.PodName)
+
+	select {
+	case podName = <-podNameChannel:
+	case <-time.After(timeout):
+		return fmt.Errorf("not found pod name")
 	}
 
-	if podName == "" {
-		return fmt.Errorf("unable to fetch logs as the name of the target pod is empty")
+	waitLogsChannel := waitForLogsFromPodChannel(podName, kubeClient, crdClient)
+
+	select {
+	case <-waitLogsChannel:
+	case <-time.After(timeout):
+		return fmt.Errorf("timeout to fetch logs from pod \"%s\"", podName)
 	}
 
-	out := os.Stdout
-	if FollowLogs {
-		if err := streamLogs(out, kubeClientset, podName); err != nil {
-			return err
-		}
+	if followLogs {
+		return streamLogs(os.Stdout, kubeClient, podName)
 	} else {
-		if err := printLogs(out, kubeClientset, podName); err != nil {
-			return err
-		}
+		return printLogs(os.Stdout, kubeClient, podName)
 	}
+}
 
-	return nil
+func getPodNameChannel(
+	sparkApplicationName string,
+	crdClient crdclientset.Interface) chan string {
+
+	channel := make(chan string, 1)
+	go func() {
+		for true {
+			app, _ := crdClient.SparkoperatorV1beta2().SparkApplications(Namespace).Get(
+				context.TODO(),
+				sparkApplicationName,
+				metav1.GetOptions{})
+
+			if app.Status.DriverInfo.PodName != "" {
+				channel <- app.Status.DriverInfo.PodName
+				break
+			}
+		}
+	}()
+	return channel
+}
+
+func waitForLogsFromPodChannel(
+	podName string,
+	kubeClient clientset.Interface,
+	crdClient crdclientset.Interface) chan bool {
+
+	channel := make(chan bool, 1)
+	go func() {
+		for true {
+			_, err := kubeClient.CoreV1().Pods(Namespace).GetLogs(podName, &apiv1.PodLogOptions{}).Do(context.TODO()).Raw()
+
+			if err == nil {
+				channel <- true
+				break
+			}
+		}
+	}()
+	return channel
 }
 
 // printLogs is a one time operation that prints the fetched logs of the given pod.


Test patch:
diff --git a/pkg/controller/sparkapplication/controller_test.go b/pkg/controller/sparkapplication/controller_test.go
index 3bcb60c..1dcb8e5 100644
--- a/pkg/controller/sparkapplication/controller_test.go
+++ b/pkg/controller/sparkapplication/controller_test.go
@@ -68,7 +68,7 @@ func newFakeController(app *v1beta2.SparkApplication, pods ...*apiv1.Pod) (*Cont
 
 	podInformerFactory := informers.NewSharedInformerFactory(kubeClient, 0*time.Second)
 	controller := newSparkApplicationController(crdClient, kubeClient, informerFactory, podInformerFactory, recorder,
-		&util.MetricConfig{}, "", nil, true)
+		&util.MetricConfig{}, "", "", nil, true)
 
 	informer := informerFactory.Sparkoperator().V1beta2().SparkApplications().Informer()
 	if app != nil {
@@ -1619,6 +1619,52 @@ func TestIngressWithSubpathAffectsSparkConfiguration(t *testing.T) {
 	}
 }
 
+func TestIngressWithClassName(t *testing.T) {
+	os.Setenv(kubernetesServiceHostEnvVar, "localhost")
+	os.Setenv(kubernetesServicePortEnvVar, "443")
+
+	appName := "ingressaffectssparkconfig"
+
+	app := &v1beta2.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      appName,
+			Namespace: "test",
+		},
+		Spec: v1beta2.SparkApplicationSpec{
+			RestartPolicy: v1beta2.RestartPolicy{
+				Type: v1beta2.Never,
+			},
+			TimeToLiveSeconds: int64ptr(1),
+		},
+		Status: v1beta2.SparkApplicationStatus{},
+	}
+
+	ctrl, _ := newFakeController(app)
+	ctrl.ingressURLFormat = "{{$appNamespace}}.{{$appName}}.example.com"
+	ctrl.ingressClassName = "nginx"
+	ctrl.enableUIService = true
+	_, err := ctrl.crdClient.SparkoperatorV1beta2().SparkApplications(app.Namespace).Create(context.TODO(), app, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = ctrl.syncSparkApplication(fmt.Sprintf("%s/%s", app.Namespace, app.Name))
+	assert.Nil(t, err)
+	_, err = ctrl.crdClient.SparkoperatorV1beta2().SparkApplications(app.Namespace).Get(context.TODO(), app.Name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ingresses, err := ctrl.kubeClient.NetworkingV1().Ingresses(app.Namespace).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ingresses == nil || ingresses.Items == nil || len(ingresses.Items) != 1 {
+		t.Fatal("The ingress does not exist, has no items, or wrong amount of items")
+	}
+	if ingresses.Items[0].Spec.IngressClassName == nil || *ingresses.Items[0].Spec.IngressClassName != "nginx" {
+		t.Fatal("The ingressClassName does not exists, or wrong value is set")
+	}
+}
+
 func stringptr(s string) *string {
 	return &s
 }
diff --git a/pkg/controller/sparkapplication/sparkui_test.go b/pkg/controller/sparkapplication/sparkui_test.go
index f6b8128..acec9b7 100644
--- a/pkg/controller/sparkapplication/sparkui_test.go
+++ b/pkg/controller/sparkapplication/sparkui_test.go
@@ -330,7 +330,7 @@ func TestCreateSparkUIIngress(t *testing.T) {
 		expectError     bool
 	}
 
-	testFn := func(test testcase, t *testing.T, ingressURLFormat string) {
+	testFn := func(test testcase, t *testing.T, ingressURLFormat string, ingressClassName string) {
 		fakeClient := fake.NewSimpleClientset()
 		sparkService, err := createSparkUIService(test.app, fakeClient)
 		if err != nil {
@@ -340,7 +340,7 @@ func TestCreateSparkUIIngress(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		sparkIngress, err := createSparkUIIngress(test.app, *sparkService, ingressURL, fakeClient)
+		sparkIngress, err := createSparkUIIngress(test.app, *sparkService, ingressURL, ingressClassName, fakeClient)
 		if err != nil {
 			if test.expectError {
 				return
@@ -492,6 +492,7 @@ func TestCreateSparkUIIngress(t *testing.T) {
 			},
 		},
 	}
+
 	testcases := []testcase{
 		{
 			name: "simple ingress object",
@@ -550,7 +551,7 @@ func TestCreateSparkUIIngress(t *testing.T) {
 	}
 
 	for _, test := range testcases {
-		testFn(test, t, "{{$appName}}.ingress.clusterName.com")
+		testFn(test, t, "{{$appName}}.ingress.clusterName.com", "")
 	}
 
 	testcases = []testcase{
@@ -569,7 +570,23 @@ func TestCreateSparkUIIngress(t *testing.T) {
 	}
 
 	for _, test := range testcases {
-		testFn(test, t, "ingress.clusterName.com/{{$appNamespace}}/{{$appName}}")
+		testFn(test, t, "ingress.clusterName.com/{{$appNamespace}}/{{$appName}}", "")
+	}
+
+	testcases = []testcase{
+		{
+			name: "simple ingress object with ingressClassName set",
+			app:  app1,
+			expectedIngress: SparkIngress{
+				ingressName:      fmt.Sprintf("%s-ui-ingress", app1.GetName()),
+				ingressURL:       parseURLAndAssertError(app1.GetName()+".ingress.clusterName.com", t),
+				ingressClassName: "nginx",
+			},
+			expectError: false,
+		},
+	}
+	for _, test := range testcases {
+		testFn(test, t, "{{$appName}}.ingress.clusterName.com", "nginx")
 	}
 }
 
diff --git a/test/e2e/framework/framework.go b/test/e2e/framework/framework.go
index ffcdcb7..9480ee9 100644
--- a/test/e2e/framework/framework.go
+++ b/test/e2e/framework/framework.go
@@ -39,6 +39,7 @@ type Framework struct {
 	SparkApplicationClient crdclientset.Interface
 	MasterHost             string
 	Namespace              *v1.Namespace
+	SparkTestNamespace     *v1.Namespace
 	OperatorPod            *v1.Pod
 	DefaultTimeout         time.Duration
 }
@@ -64,6 +65,11 @@ func New(ns, sparkNs, kubeconfig, opImage string, opImagePullPolicy string) (*Fr
 		fmt.Println(nil, err, namespace)
 	}
 
+	sparkTestNamespace, err := CreateNamespace(cli, sparkNs)
+	if err != nil {
+		fmt.Println(nil, err, sparkNs)
+	}
+
 	saClient, err := crdclientset.NewForConfig(config)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create SparkApplication client")
@@ -74,6 +80,7 @@ func New(ns, sparkNs, kubeconfig, opImage string, opImagePullPolicy string) (*Fr
 		KubeClient:             cli,
 		SparkApplicationClient: saClient,
 		Namespace:              namespace,
+		SparkTestNamespace:     sparkTestNamespace,
 		DefaultTimeout:         time.Minute,
 	}
 
@@ -94,31 +101,31 @@ func (f *Framework) Setup(sparkNs, opImage string, opImagePullPolicy string) err
 }
 
 func (f *Framework) setupOperator(sparkNs, opImage string, opImagePullPolicy string) error {
-	if _, err := CreateServiceAccount(f.KubeClient, f.Namespace.Name, "../../manifest/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if _, err := CreateServiceAccount(f.KubeClient, f.Namespace.Name, "../../manifest/spark-operator-install/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to create operator service account")
 	}
 
-	if err := CreateClusterRole(f.KubeClient, "../../manifest/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if err := CreateClusterRole(f.KubeClient, "../../manifest/spark-operator-install/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to create cluster role")
 	}
 
-	if _, err := CreateClusterRoleBinding(f.KubeClient, "../../manifest/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if _, err := CreateClusterRoleBinding(f.KubeClient, "../../manifest/spark-operator-install/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to create cluster role binding")
 	}
 
-	if _, err := CreateServiceAccount(f.KubeClient, sparkNs, "../../manifest/spark-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if _, err := CreateServiceAccount(f.KubeClient, f.SparkTestNamespace.Name, "../../manifest/spark-application-rbac/spark-application-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to create Spark service account")
 	}
 
-	if err := CreateRole(f.KubeClient, sparkNs, "../../manifest/spark-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if err := CreateRole(f.KubeClient, f.SparkTestNamespace.Name, "../../manifest/spark-application-rbac/spark-application-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to create role")
 	}
 
-	if _, err := CreateRoleBinding(f.KubeClient, sparkNs, "../../manifest/spark-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if _, err := CreateRoleBinding(f.KubeClient, f.SparkTestNamespace.Name, "../../manifest/spark-application-rbac/spark-application-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to create role binding")
 	}
 
-	job, err := MakeJob("../../manifest/spark-operator-with-webhook.yaml")
+	job, err := MakeJob("../../manifest/spark-operator-with-webhook-install/spark-operator-webhook.yaml")
 	if err != nil {
 		return err
 	}
@@ -146,11 +153,11 @@ func (f *Framework) setupOperator(sparkNs, opImage string, opImagePullPolicy str
 		return errors.Wrap(err, "failed to delete the init job")
 	}
 
-	if _, err := CreateService(f.KubeClient, f.Namespace.Name, "../../manifest/spark-operator-with-webhook.yaml"); err != nil {
+	if _, err := CreateService(f.KubeClient, f.Namespace.Name, "../../manifest/spark-operator-with-webhook-install/spark-operator-webhook.yaml"); err != nil {
 		return errors.Wrap(err, "failed to create webhook service")
 	}
 
-	deploy, err := MakeDeployment("../../manifest/spark-operator-with-webhook.yaml")
+	deploy, err := MakeDeployment("../../manifest/spark-operator-with-webhook-install/spark-operator-with-webhook.yaml")
 	if err != nil {
 		return err
 	}
@@ -185,11 +192,11 @@ func (f *Framework) setupOperator(sparkNs, opImage string, opImagePullPolicy str
 
 // Teardown tears down a previously initialized test environment.
 func (f *Framework) Teardown() error {
-	if err := DeleteClusterRole(f.KubeClient, "../../manifest/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if err := DeleteClusterRole(f.KubeClient, "../../manifest/spark-operator-install/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to delete operator cluster role")
 	}
 
-	if err := DeleteClusterRoleBinding(f.KubeClient, "../../manifest/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
+	if err := DeleteClusterRoleBinding(f.KubeClient, "../../manifest/spark-operator-install/spark-operator-rbac.yaml"); err != nil && !apierrors.IsAlreadyExists(err) {
 		return errors.Wrap(err, "failed to delete operator cluster role binding")
 	}
 
@@ -201,5 +208,9 @@ func (f *Framework) Teardown() error {
 		return err
 	}
 
+	if err := DeleteNamespace(f.KubeClient, f.SparkTestNamespace.Name); err != nil {
+		return err
+	}
+
 	return nil
 }
diff --git a/test/e2e/main_test.go b/test/e2e/main_test.go
index 1ab3d99..12b08b7 100644
--- a/test/e2e/main_test.go
+++ b/test/e2e/main_test.go
@@ -57,7 +57,7 @@ func TestMain(m *testing.M) {
 	opImage := flag.String("operator-image", "", "operator image, e.g. image:tag")
 	opImagePullPolicy := flag.String("operator-image-pullPolicy", "IfNotPresent", "pull policy, e.g. Always")
 	ns := flag.String("namespace", "spark-operator", "e2e test namespace")
-	sparkTestNamespace := flag.String("spark-test-namespace", "default", "e2e test spark-test-namespace")
+	sparkTestNamespace := flag.String("spark", "spark", "e2e test spark-test-namespace")
 	sparkTestImage := flag.String("spark-test-image", "", "spark test image, e.g. image:tag")
 	sparkTestServiceAccount := flag.String("spark-test-service-account", "spark", "e2e test spark test service account")
 	flag.Parse()