Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #1504 (2022-03-31 15:26:01)

Added missing manifest yaml, point the manifest to the right direction

google-cla[bot]: Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

For more information, open the [CLA check for this pull request](https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/pull/1504/checks?check_run_id=5773772333).

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/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..7538fee 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.
@@ -687,7 +699,7 @@ 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 {
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
 }
 


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