[ISSUE-524][operator] Upgrading rss could also be deleted (#531)

### What changes were proposed in this pull request?
1. soft the constraint that upgrading rss object cannot be deleted

### Why are the changes needed?
more flexibility.
This fixes #524 

### Does this PR introduce _any_ user-facing change?
For RSS cluster admin, they can delete a upgrading rss cluster.

### How was this patch tested?
Manually verified and an added UT

deploy/kubernetes/operator/pkg/controller/controller/rss.go

@@ -354,7 +354,8 @@ func (r *rssController) processRss(namespace, name string) (bool, error) {
 func (r *rssController) processDeleting(rss *unifflev1alpha1.RemoteShuffleService) (bool, error) {
 	klog.V(4).Infof("process rss (%v) to be deleted in %v phase",
 		utils.UniqueName(rss), rss.Status.Phase)
-	if rss.Status.Phase == unifflev1alpha1.RSSRunning || rss.Status.Phase == unifflev1alpha1.RSSPending {
+	if rss.Status.Phase == unifflev1alpha1.RSSRunning || rss.Status.Phase == unifflev1alpha1.RSSPending ||
+		rss.Status.Phase == unifflev1alpha1.RSSUpgrading {
 		return false, r.updateRssStatus(rss, &unifflev1alpha1.RemoteShuffleServiceStatus{
 			Phase: unifflev1alpha1.RSSTerminating,
 		})

deploy/kubernetes/operator/pkg/controller/controller/rss_test.go

@@ -27,6 +27,7 @@ import (
 	. "github.com/onsi/gomega"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
@@ -235,6 +236,37 @@ var _ = Describe("RssController", func() {
 			Expect(err).ToNot(HaveOccurred())
 			Expect(sts).ToNot(BeNil())
 			Expect(*sts.Spec.Replicas).To(Equal(int32(3)))
+
+			// since we are in the env test, the rss object may never transmit upgrading to running.
+			By("Ensure rss object is still upgrading")
+			err = wait.Poll(time.Second, time.Second*5, func() (bool, error) {
+				curRss, getErr := testRssClient.UniffleV1alpha1().RemoteShuffleServices(testNamespace).
+					Get(context.TODO(), testRssName, metav1.GetOptions{})
+				if getErr != nil {
+					return false, getErr
+				}
+				if curRss.Status.Phase != unifflev1alpha1.RSSUpgrading {
+					return false, nil
+				}
+				return true, nil
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Delete the upgrading rss object")
+			err = testRssClient.UniffleV1alpha1().RemoteShuffleServices(corev1.NamespaceDefault).
+				Delete(context.TODO(), testRssName, metav1.DeleteOptions{})
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Waiting the rss object being delete")
+			err = wait.Poll(time.Second, time.Second*5, func() (done bool, err error) {
+				_, getErr := testRssClient.UniffleV1alpha1().RemoteShuffleServices(testNamespace).
+					Get(context.TODO(), testRssName, metav1.GetOptions{})
+				if getErr != nil && errors.IsNotFound(getErr) {
+					return true, nil
+				}
+				return false, nil
+			})
+			Expect(err).ToNot(HaveOccurred())
 		})
 	})
 })

deploy/kubernetes/operator/pkg/webhook/inspector/rss.go

@@ -42,6 +42,8 @@ func (i *inspector) validateRSS(ar *admissionv1.AdmissionReview) *admissionv1.Ad
 			return util.AdmissionReviewFailed(ar, err)
 		}
 		// for security purposes, we forbid updating rss objects when they are in upgrading phase.
+		// generally speaking, we should also deny updating when rss is terminating. However, it would introduce more
+		// complexity and controller's current terminating logic can tolerate the rss object update.
 		if oldRSS.Status.Phase == unifflev1alpha1.RSSUpgrading {
 			message := "can not update upgrading rss object: " + utils.UniqueName(oldRSS)
 			return util.AdmissionReviewForbidden(ar, message)