Block eventing cr deletion

api/v1alpha1/eventing_types.go

@@ -31,11 +31,13 @@ const (
 	StateReady      string = "Ready"
 	StateError      string = "Error"
 	StateProcessing string = "Processing"
+	StateWarning    string = "Warning"
 
 	ConditionNATSAvailable            ConditionType = "NATSAvailable"
 	ConditionPublisherProxyReady      ConditionType = "PublisherProxyReady"
 	ConditionWebhookReady             ConditionType = "WebhookReady"
 	ConditionSubscriptionManagerReady ConditionType = "SubscriptionManagerReady"
+	ConditionDeleted                  ConditionType = "Deleted"
 
 	// common reasons
 	ConditionReasonProcessing ConditionReason = "Processing"
@@ -51,7 +53,7 @@ const (
 	ConditionReasonForbidden                  ConditionReason = "Forbidden"
 	ConditionReasonWebhookFailed              ConditionReason = "WebhookFailed"
 	ConditionReasonWebhookReady               ConditionReason = "Ready"
-	ConditionReasonDeletedFailed              ConditionReason = "DeletionFailed"
+	ConditionReasonDeletionError              ConditionReason = "DeletionError"
 
 	// message for conditions
 	ConditionPublisherProxyReadyMessage        = "Publisher proxy is deployed"
@@ -66,7 +68,6 @@ const (
 	ConditionReasonEventMeshSubManagerReady      ConditionReason = "EventMeshSubscriptionManagerReady"
 	ConditionReasonEventMeshSubManagerFailed     ConditionReason = "EventMeshSubscriptionManagerFailed"
 	ConditionReasonEventMeshSubManagerStopFailed ConditionReason = "EventMeshSubscriptionManagerStopFailed"
-	ConditionReasonSubscriptionManagerProcessing ConditionReason = "SubscriptionManagerProcessing"
 )
 
 // NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.

api/v1alpha1/status.go

@@ -56,6 +56,18 @@ func (sm *EventingStatus) UpdateConditionSubscriptionManagerReady(status metav1.
 	meta.SetStatusCondition(&sm.Conditions, condition)
 }
 
+func (es *EventingStatus) UpdateConditionDeletion(status metav1.ConditionStatus, reason ConditionReason,
+	message string) {
+	condition := metav1.Condition{
+		Type:               string(ConditionDeleted),
+		Status:             status,
+		LastTransitionTime: metav1.Now(),
+		Reason:             string(reason),
+		Message:            message,
+	}
+	meta.SetStatusCondition(&es.Conditions, condition)
+}
+
 func (es *EventingStatus) SetSubscriptionManagerReadyConditionToTrue() {
 	es.UpdateConditionSubscriptionManagerReady(metav1.ConditionTrue, ConditionReasonEventMeshSubManagerReady,
 		ConditionSubscriptionManagerReadyMessage)
@@ -67,6 +79,10 @@ func (es *EventingStatus) SetStateReady() {
 	es.UpdateConditionPublisherProxyReady(metav1.ConditionTrue, ConditionReasonDeployed, ConditionPublisherProxyReadyMessage)
 }
 
+func (ns *EventingStatus) SetStateWarning() {
+	ns.State = StateWarning
+}
+
 func (es *EventingStatus) SetNATSAvailableConditionToTrue() {
 	es.UpdateConditionNATSAvailable(metav1.ConditionTrue, ConditionReasonNATSAvailable, ConditionNATSAvailableMessage)
 }

internal/controller/eventing/controller.go

@@ -18,8 +18,8 @@ package eventing
 
 import (
 	"context"
+	"errors"
 	"fmt"
-
 	eventingv1alpha1 "github.com/kyma-project/eventing-manager/api/v1alpha1"
 	"github.com/kyma-project/eventing-manager/options"
 	"github.com/kyma-project/eventing-manager/pkg/env"
@@ -60,6 +60,8 @@ const (
 	NamespacePrefix                       = "/"
 	EventMeshPublishEndpointForSubscriber = "/sap/ems/v1"
 	EventMeshPublishEndpointForPublisher  = "/sap/ems/v1/events"
+
+	SubscriptionExistsErrMessage = "cannot delete the eventing module as subscription exists"
 )
 
 // Reconciler reconciles an Eventing object
@@ -265,6 +267,18 @@ func (r *Reconciler) handleEventingDeletion(ctx context.Context, eventing *event
 		return ctrl.Result{}, nil
 	}
 
+	// check if subscription resources exist
+	exists, err := r.eventingManager.SubscriptionExists(ctx)
+	if err != nil {
+		eventing.Status.SetStateError()
+		return ctrl.Result{}, r.syncStatusWithDeletionErr(ctx, eventing, err, log)
+	}
+	if exists {
+		eventing.Status.SetStateWarning()
+		return ctrl.Result{Requeue: true}, r.syncStatusWithDeletionErr(ctx, eventing,
+			errors.New(SubscriptionExistsErrMessage), log)
+	}
+
 	log.Info("handling Eventing deletion...")
 	if eventing.Spec.Backend.Type == eventingv1alpha1.NatsBackendType {
 		if err := r.stopNATSSubManager(true, log); err != nil {
@@ -284,7 +298,7 @@ func (r *Reconciler) handleEventingDeletion(ctx context.Context, eventing *event
 	// delete cluster-scoped resources, such as clusterrole and clusterrolebinding.
 	if err := r.deleteClusterScopedResources(ctx, eventing); err != nil {
 		return ctrl.Result{}, r.syncStatusWithPublisherProxyErrWithReason(ctx,
-			eventingv1alpha1.ConditionReasonDeletedFailed, eventing, err, log)
+			eventingv1alpha1.ConditionReasonDeletionError, eventing, err, log)
 	}
 	eventing.Status.SetPublisherProxyConditionToFalse(
 		eventingv1alpha1.ConditionReasonDeleted,

internal/controller/eventing/integrationtests/controller/integration_test.go

@@ -7,7 +7,6 @@ import (
 	eventingv1alpha1 "github.com/kyma-project/eventing-manager/api/v1alpha1"
 	eventingcontroller "github.com/kyma-project/eventing-manager/internal/controller/eventing"
 	"github.com/kyma-project/eventing-manager/pkg/eventing"
-	ecsubmanagermocks "github.com/kyma-project/eventing-manager/pkg/subscriptionmanager/mocks/ec"
 	"github.com/kyma-project/eventing-manager/test/matchers"
 	"github.com/kyma-project/eventing-manager/test/utils"
 	testutils "github.com/kyma-project/eventing-manager/test/utils/integration"
@@ -16,6 +15,7 @@ import (
 	gomegatypes "github.com/onsi/gomega/types"
 	"github.com/stretchr/testify/require"
 
+	eventinv1alpha2 "github.com/kyma-project/kyma/components/eventing-controller/api/v1alpha2"
 	natstestutils "github.com/kyma-project/nats-manager/testutils"
 	v1 "k8s.io/api/apps/v1"
 )
@@ -560,9 +560,10 @@ func Test_CreateEventingCR_EventMesh(t *testing.T) {
 func Test_DeleteEventingCR(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {
-		name                    string
-		givenEventing           *eventingv1alpha1.Eventing
-		subscriptionManagerMock *ecsubmanagermocks.Manager
+		name              string
+		givenEventing     *eventingv1alpha1.Eventing
+		givenSubscription *eventinv1alpha2.Subscription
+		wantMatches       gomegatypes.GomegaMatcher
 	}{
 		{
 			name: "Delete Eventing CR should delete the owned resources",
@@ -580,12 +581,42 @@ func Test_DeleteEventingCR(t *testing.T) {
 				utils.WithEventingEventTypePrefix("test-prefix"),
 			),
 		},
+		{
+			name: "Delete should be blocked as subscription exists for NATS",
+			givenEventing: utils.NewEventingCR(
+				utils.WithEventingCRMinimal(),
+				utils.WithEventingStreamData("Memory", "1M", 1, 1),
+				utils.WithEventingPublisherData(1, 1, "199m", "99Mi", "399m", "199Mi"),
+			),
+			givenSubscription: utils.NewSubscription("test-nats-subscription", "test-nats-namespace"),
+			wantMatches: gomega.And(
+				matchers.HaveStatusWarning(),
+				matchers.HaveDeletionErrorCondition(eventingcontroller.SubscriptionExistsErrMessage),
+				matchers.HaveFinalizer(),
+			),
+		},
+		{
+			name: "Delete should be blocked as subscription exist for EventMesh",
+			givenEventing: utils.NewEventingCR(
+				utils.WithEventMeshBackend("test-secret-name"),
+				utils.WithEventingPublisherData(1, 1, "199m", "99Mi", "399m", "199Mi"),
+				utils.WithEventingEventTypePrefix("test-prefix"),
+			),
+			givenSubscription: utils.NewSubscription("test-eventmesh-subscription", "test-eventmesh-namespace"),
+			wantMatches: gomega.And(
+				matchers.HaveStatusWarning(),
+				matchers.HaveDeletionErrorCondition(eventingcontroller.SubscriptionExistsErrMessage),
+				matchers.HaveFinalizer(),
+			),
+		},
 	}
 
 	for _, tc := range testCases {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
+			g := gomega.NewWithT(t)
+
 			// given
 			eventingcontroller.IsDeploymentReady = func(deployment *v1.Deployment) bool {
 				return true
@@ -615,6 +646,10 @@ func Test_DeleteEventingCR(t *testing.T) {
 			testEnvironment.EnsureK8sResourceCreated(t, tc.givenEventing)
 
 			defer func() {
+				if tc.givenSubscription != nil {
+					testEnvironment.EnsureSubscriptionResourceDeletion(t, tc.givenSubscription.Name, tc.givenSubscription.Namespace)
+				}
+
 				if !*testEnvironment.EnvTestInstance.UseExistingCluster {
 					testEnvironment.EnsureDeploymentDeletion(t, eventing.GetPublisherDeploymentName(*tc.givenEventing), givenNamespace)
 				}
@@ -627,30 +662,43 @@ func Test_DeleteEventingCR(t *testing.T) {
 			testEnvironment.EnsureDeploymentExists(t, eventing.GetPublisherDeploymentName(*tc.givenEventing), givenNamespace)
 			testEnvironment.EnsureHPAExists(t, eventing.GetPublisherDeploymentName(*tc.givenEventing), givenNamespace)
 
-			testEnvironment.EnsureEventingResourceDeletion(t, tc.givenEventing.Name, givenNamespace)
+			if tc.givenSubscription != nil {
+				// create subscriptions if given.
+				testEnvironment.EnsureNamespaceCreation(t, tc.givenSubscription.Namespace)
+				testEnvironment.EnsureK8sResourceCreated(t, tc.givenSubscription)
+				testEnvironment.EnsureSubscriptionExists(t, tc.givenSubscription.Name, tc.givenSubscription.Namespace)
 
-			// then
-			if *testEnvironment.EnvTestInstance.UseExistingCluster {
-				testEnvironment.EnsureDeploymentNotFound(t, eventing.GetPublisherDeploymentName(*tc.givenEventing), givenNamespace)
-				testEnvironment.EnsureHPANotFound(t, eventing.GetPublisherDeploymentName(*tc.givenEventing), givenNamespace)
-				testEnvironment.EnsureK8sServiceNotFound(t,
-					eventing.GetPublisherPublishServiceName(*tc.givenEventing), givenNamespace)
-				testEnvironment.EnsureK8sServiceNotFound(t,
-					eventing.GetPublisherMetricsServiceName(*tc.givenEventing), givenNamespace)
-				testEnvironment.EnsureK8sServiceNotFound(t,
-					eventing.GetPublisherHealthServiceName(*tc.givenEventing), givenNamespace)
-				testEnvironment.EnsureK8sServiceAccountNotFound(t,
-					eventing.GetPublisherServiceAccountName(*tc.givenEventing), givenNamespace)
+				//then
+				// givenSubscription existence means deletion of Eventing CR should be blocked.
+				testEnvironment.EnsureK8sResourceDeleted(t, tc.givenEventing)
+				testEnvironment.GetEventingAssert(g, tc.givenEventing).Should(tc.wantMatches)
 			} else {
-				// check if the owner reference is set.
-				// if owner reference is set then these resources would be garbage collected in real k8s cluster.
-				testEnvironment.EnsureEPPK8sResourcesHaveOwnerReference(t, *tc.givenEventing)
-				// ensure clusterrole and clusterrolebindings are deleted.
+				// then
+				// givenSubscription is nil means deletion of Eventing CR should be successful.
+				testEnvironment.EnsureEventingResourceDeletion(t, tc.givenEventing.Name, givenNamespace)
+
+				// then
+				if *testEnvironment.EnvTestInstance.UseExistingCluster {
+					testEnvironment.EnsureDeploymentNotFound(t, eventing.GetPublisherDeploymentName(*tc.givenEventing), givenNamespace)
+					testEnvironment.EnsureHPANotFound(t, eventing.GetPublisherDeploymentName(*tc.givenEventing), givenNamespace)
+					testEnvironment.EnsureK8sServiceNotFound(t,
+						eventing.GetPublisherPublishServiceName(*tc.givenEventing), givenNamespace)
+					testEnvironment.EnsureK8sServiceNotFound(t,
+						eventing.GetPublisherMetricsServiceName(*tc.givenEventing), givenNamespace)
+					testEnvironment.EnsureK8sServiceNotFound(t,
+						eventing.GetPublisherHealthServiceName(*tc.givenEventing), givenNamespace)
+					testEnvironment.EnsureK8sServiceAccountNotFound(t,
+						eventing.GetPublisherServiceAccountName(*tc.givenEventing), givenNamespace)
+				} else {
+					// check if the owner reference is set.
+					// if owner reference is set then these resources would be garbage collected in real k8s cluster.
+					testEnvironment.EnsureEPPK8sResourcesHaveOwnerReference(t, *tc.givenEventing)
+				}
+				testEnvironment.EnsureK8sClusterRoleNotFound(t,
+					eventing.GetPublisherClusterRoleName(*tc.givenEventing), givenNamespace)
+				testEnvironment.EnsureK8sClusterRoleBindingNotFound(t,
+					eventing.GetPublisherClusterRoleBindingName(*tc.givenEventing), givenNamespace)
 			}
-			testEnvironment.EnsureK8sClusterRoleNotFound(t,
-				eventing.GetPublisherClusterRoleName(*tc.givenEventing), givenNamespace)
-			testEnvironment.EnsureK8sClusterRoleBindingNotFound(t,
-				eventing.GetPublisherClusterRoleBindingName(*tc.givenEventing), givenNamespace)
 		})
 	}
 }

internal/controller/eventing/status.go

@@ -93,6 +93,14 @@ func (r *Reconciler) syncStatusWithWebhookErr(ctx context.Context,
 	return errors.Join(err, r.syncEventingStatus(ctx, eventing, log))
 }
 
+func (r *Reconciler) syncStatusWithDeletionErr(ctx context.Context,
+	eventing *eventingv1alpha1.Eventing, err error, log *zap.SugaredLogger) error {
+	eventing.Status.UpdateConditionDeletion(metav1.ConditionFalse,
+		eventingv1alpha1.ConditionReasonDeletionError, err.Error())
+
+	return errors.Join(err, r.syncEventingStatus(ctx, eventing, log))
+}
+
 // syncEventingStatus syncs Eventing status.
 func (r *Reconciler) syncEventingStatus(ctx context.Context,
 	eventing *eventingv1alpha1.Eventing, log *zap.SugaredLogger) error {

pkg/eventing/manager.go

@@ -39,6 +39,7 @@ type Manager interface {
 	DeployPublisherProxyResources(context.Context, *v1alpha1.Eventing, *appsv1.Deployment) error
 	GetBackendConfig() *env.BackendConfig
 	SetBackendConfig(env.BackendConfig)
+	SubscriptionExists(ctx context.Context) (bool, error)
 }
 
 type EventingManager struct {
@@ -215,6 +216,17 @@ func (em EventingManager) DeployPublisherProxyResources(
 	return nil
 }
 
+func (em *EventingManager) SubscriptionExists(ctx context.Context) (bool, error) {
+	subscriptionList, err := em.kubeClient.GetSubscriptions(ctx)
+	if err != nil {
+		return false, err
+	}
+	if len(subscriptionList.Items) > 0 {
+		return true, nil
+	}
+	return false, nil
+}
+
 func convertECBackendType(backendType v1alpha1.BackendType) (ecv1alpha1.BackendType, error) {
 	switch backendType {
 	case v1alpha1.EventMeshBackendType: