using mw finalizer instead of resource finalizer (#150)

Signed-off-by: Wei Liu <liuweixa@redhat.com>

Author: skeeey

pkg/cloudevents/clients/common/common.go

@@ -30,8 +30,6 @@ const CloudEventsOriginalSourceLabelKey = "cloudevents.open-cluster-management.i
 // ResourceDeleted represents a resource is deleted.
 const ResourceDeleted = "Deleted"
 
-const ResourceFinalizer = "cloudevents.open-cluster-management.io/resource-cleanup"
-
 var ManagedClusterGK = schema.GroupKind{Group: clusterv1.GroupName, Kind: "ManagedCluster"}
 var ManagedClusterGR = schema.GroupResource{Group: clusterv1.GroupName, Resource: "managedclusters"}
 

pkg/cloudevents/clients/utils/utils.go

@@ -21,7 +21,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/common"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
 )
 
@@ -225,23 +224,6 @@ func UID(sourceID, groupResource, namespace, name string) string {
 	return uuid.NewSHA1(uuid.NameSpaceOID, []byte(id)).String()
 }
 
-// EnsureResourceFinalizer ensures the resource finalizer in the given finalizers
-func EnsureResourceFinalizer(finalizers []string) []string {
-	has := false
-	for _, f := range finalizers {
-		if f == common.ResourceFinalizer {
-			has = true
-			break
-		}
-	}
-
-	if !has {
-		finalizers = append(finalizers, common.ResourceFinalizer)
-	}
-
-	return finalizers
-}
-
 func IsStatusPatch(subresources []string) bool {
 	if len(subresources) == 0 {
 		return false

pkg/cloudevents/clients/utils/utils_test.go

@@ -2,7 +2,6 @@ package utils
 
 import (
 	"encoding/json"
-	"reflect"
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -453,41 +452,3 @@ func TestCompareSnowflakeSequenceIDs(t *testing.T) {
 		})
 	}
 }
-
-func TestEnsureResourceFinalizer(t *testing.T) {
-	tests := []struct {
-		name       string
-		input      []string
-		wantOutput []string
-	}{
-		{
-			name:       "empty finalizers",
-			input:      []string{},
-			wantOutput: []string{common.ResourceFinalizer},
-		},
-		{
-			name:       "finalizer already exists",
-			input:      []string{"other-finalizer", common.ResourceFinalizer},
-			wantOutput: []string{"other-finalizer", common.ResourceFinalizer},
-		},
-		{
-			name:       "finalizer not present",
-			input:      []string{"finalizer1", "finalizer2"},
-			wantOutput: []string{"finalizer1", "finalizer2", common.ResourceFinalizer},
-		},
-		{
-			name:       "nil input",
-			input:      nil,
-			wantOutput: []string{common.ResourceFinalizer},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := EnsureResourceFinalizer(tt.input)
-			if !reflect.DeepEqual(got, tt.wantOutput) {
-				t.Errorf("EnsureFinalizers() = %v, want %v", got, tt.wantOutput)
-			}
-		})
-	}
-}

pkg/cloudevents/clients/utils/work.go

@@ -48,3 +48,20 @@ func EncodeManifests(work *workv1.ManifestWork) error {
 
 	return nil
 }
+
+// EnsureManifestWorkFinalizer ensures the manifestwork finalizer in the given finalizers
+func EnsureManifestWorkFinalizer(finalizers []string) []string {
+	has := false
+	for _, f := range finalizers {
+		if f == workv1.ManifestWorkFinalizer {
+			has = true
+			break
+		}
+	}
+
+	if !has {
+		finalizers = append(finalizers, workv1.ManifestWorkFinalizer)
+	}
+
+	return finalizers
+}

pkg/cloudevents/clients/utils/work_test.go

@@ -2,6 +2,7 @@ package utils
 
 import (
 	"encoding/json"
+	"reflect"
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
@@ -64,6 +65,44 @@ func TestEncodeManifests(t *testing.T) {
 	}
 }
 
+func TestEnsureManifestWorkFinalizer(t *testing.T) {
+	tests := []struct {
+		name       string
+		input      []string
+		wantOutput []string
+	}{
+		{
+			name:       "empty finalizers",
+			input:      []string{},
+			wantOutput: []string{workv1.ManifestWorkFinalizer},
+		},
+		{
+			name:       "finalizer already exists",
+			input:      []string{"other-finalizer", workv1.ManifestWorkFinalizer},
+			wantOutput: []string{"other-finalizer", workv1.ManifestWorkFinalizer},
+		},
+		{
+			name:       "finalizer not present",
+			input:      []string{"finalizer1", "finalizer2"},
+			wantOutput: []string{"finalizer1", "finalizer2", workv1.ManifestWorkFinalizer},
+		},
+		{
+			name:       "nil input",
+			input:      nil,
+			wantOutput: []string{workv1.ManifestWorkFinalizer},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := EnsureManifestWorkFinalizer(tt.input)
+			if !reflect.DeepEqual(got, tt.wantOutput) {
+				t.Errorf("EnsureManifestWorkFinalizer() = %v, want %v", got, tt.wantOutput)
+			}
+		})
+	}
+}
+
 func configMap() *corev1.ConfigMap {
 	return &corev1.ConfigMap{
 		TypeMeta: metav1.TypeMeta{

pkg/cloudevents/clients/work/store/base.go

@@ -155,8 +155,8 @@ func (b *workProcessor) handleWork(work *workv1.ManifestWork) error {
 		return nil
 	}
 
-	// the work has been handled by agent, we ensure a finalizer on the work
-	updatedWork.Finalizers = utils.EnsureResourceFinalizer(updatedWork.Finalizers)
+	// the work has been handled by agent, we ensure the manifestwork finalizer on the work
+	updatedWork.Finalizers = utils.EnsureManifestWorkFinalizer(updatedWork.Finalizers)
 	updatedWork.Annotations[common.CloudEventsSequenceIDAnnotationKey] = sequenceID
 	updatedWork.Status = work.Status
 	// update the work with status in the local cache.

pkg/cloudevents/generic/options/grpc/protocol/heartbeat_integration_test.go

@@ -74,13 +74,17 @@ func setupMockServerWithHeartbeat(t *testing.T, heartbeatInterval time.Duration,
 	}
 	pbv1.RegisterCloudEventServiceServer(s, service)
 
+	serverReady := make(chan struct{})
 	go func() {
+		close(serverReady)
 		if err := s.Serve(lis); err != nil {
 			t.Logf("Server exited with error: %v", err)
 		}
 	}()
 
-	time.Sleep(50 * time.Millisecond)
+	// Wait for server to be ready
+	<-serverReady
+	time.Sleep(10 * time.Millisecond) // Small grace period for gRPC initialization
 
 	bufDialer := func(context.Context, string) (net.Conn, error) {
 		return lis.Dial()
@@ -117,32 +121,32 @@ func TestProtocol_HeartbeatIntegration(t *testing.T) {
 			heartbeatInterval:        100 * time.Millisecond,
 			serverHealthinessTimeout: ptr.To(500 * time.Millisecond),
 			expectHealthCheckError:   false,
-			testDuration:             800 * time.Millisecond,
+			testDuration:             1500 * time.Millisecond, // Increased for stability
 			stopServerAfterEvents:    0,
 		},
 		{
 			name:                     "health check timeout when no heartbeats",
 			heartbeatInterval:        0, // No heartbeats
 			serverHealthinessTimeout: ptr.To(200 * time.Millisecond),
 			expectHealthCheckError:   true,
-			testDuration:             500 * time.Millisecond,
+			testDuration:             1000 * time.Millisecond, // Increased to allow Subscribe + timeout
 			stopServerAfterEvents:    0,
 		},
 		{
 			name:                     "health check disabled",
 			heartbeatInterval:        0,   // No heartbeats
 			serverHealthinessTimeout: nil, // Disabled
 			expectHealthCheckError:   false,
-			testDuration:             300 * time.Millisecond,
+			testDuration:             500 * time.Millisecond, // Increased for stability
 			stopServerAfterEvents:    0,
 		},
 		{
 			name:                     "heartbeat stops mid-stream",
 			heartbeatInterval:        50 * time.Millisecond,
 			serverHealthinessTimeout: ptr.To(200 * time.Millisecond),
 			expectHealthCheckError:   true,
-			testDuration:             600 * time.Millisecond,
-			stopServerAfterEvents:    3, // Stop after 3 heartbeats
+			testDuration:             1000 * time.Millisecond, // Increased to allow Subscribe + timeout
+			stopServerAfterEvents:    3,                       // Stop after 3 heartbeats
 		},
 	}
 
@@ -169,7 +173,10 @@ func TestProtocol_HeartbeatIntegration(t *testing.T) {
 			ctx, cancel := context.WithTimeout(context.Background(), tt.testDuration)
 			defer cancel()
 
+			// Use a channel to ensure OpenInbound has started before checking for errors
+			started := make(chan struct{})
 			go func() {
+				close(started)
 				if err := p.OpenInbound(ctx); err != nil {
 					select {
 					case p.reconnectErrorChan <- err:
@@ -178,6 +185,10 @@ func TestProtocol_HeartbeatIntegration(t *testing.T) {
 				}
 			}()
 
+			// Wait for goroutine to start and give Subscribe call time to establish
+			<-started
+			time.Sleep(100 * time.Millisecond)
+
 			if tt.expectHealthCheckError {
 				select {
 				case err := <-reconnectErrorChan:
@@ -218,7 +229,7 @@ func TestProtocol_StartEventsReceiver_HeartbeatFiltering(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond)
 	defer cancel()
 
 	var receivedEvents atomic.Int32
@@ -232,7 +243,10 @@ func TestProtocol_StartEventsReceiver_HeartbeatFiltering(t *testing.T) {
 		}
 	}()
 
+	// Use a channel to ensure OpenInbound has started before checking for errors
+	started := make(chan struct{})
 	go func() {
+		close(started)
 		if err := p.OpenInbound(ctx); err != nil {
 			select {
 			case p.reconnectErrorChan <- err:
@@ -241,6 +255,10 @@ func TestProtocol_StartEventsReceiver_HeartbeatFiltering(t *testing.T) {
 		}
 	}()
 
+	// Wait for goroutine to start and give Subscribe call time to establish
+	<-started
+	time.Sleep(100 * time.Millisecond)
+
 	<-ctx.Done()
 
 	// Should receive 0 events since only heartbeats are sent (which are filtered out)
@@ -313,7 +331,7 @@ func TestProtocol_OpenInbound_ValidationErrors(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+			ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
 			defer cancel()
 
 			err = p.OpenInbound(ctx)