Fix integration tests

qiujian16 · qiujian16 · commit e7ee2c458d38 · 2025-11-11T18:45:34.000+08:00
Signed-off-by: Jian Qiu &lt;jqiu@redhat.com&gt;
diff --git a/pkg/cloudevents/clients/store/informer.go b/pkg/cloudevents/clients/store/informer.go
@@ -9,8 +9,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/klog/v2"
-
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/utils"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
@@ -52,8 +50,6 @@ func (s *AgentInformerWatcherStore[T]) Delete(resource runtime.Object) error {
 }
 
 func (s *AgentInformerWatcherStore[T]) HandleReceivedResource(ctx context.Context, action types.ResourceAction, resource T) error {
-	logger := klog.FromContext(ctx)
-
 	switch action {
 	case types.Added:
 		newObj, err := utils.ToRuntimeObject(resource)
@@ -63,24 +59,22 @@ func (s *AgentInformerWatcherStore[T]) HandleReceivedResource(ctx context.Contex
 
 		return s.Add(newObj)
 	case types.Modified:
-		newObj, err := meta.Accessor(resource)
+		accessor, err := meta.Accessor(resource)
 		if err != nil {
 			return err
 		}
 
-		lastObj, exists, err := s.Get(newObj.GetNamespace(), newObj.GetName())
+		lastObj, exists, err := s.Get(accessor.GetNamespace(), accessor.GetName())
 		if err != nil {
 			return err
 		}
 		if !exists {
-			return fmt.Errorf("the resource %s/%s does not exist", newObj.GetNamespace(), newObj.GetName())
+			return fmt.Errorf("the resource %s/%s does not exist", accessor.GetNamespace(), accessor.GetName())
 		}
 
-		// prevent the resource from being updated if it is deleting
+		// if resource is deleting, keep the deletiong timestamp
 		if !lastObj.GetDeletionTimestamp().IsZero() {
-			logger.Info("the resource is deleting, ignore the update",
-				"resourceNamespace", newObj.GetNamespace(), "resourceName", newObj.GetName())
-			return nil
+			accessor.SetDeletionTimestamp(lastObj.GetDeletionTimestamp())
 		}
 
 		updated, err := utils.ToRuntimeObject(resource)
@@ -99,10 +93,6 @@ func (s *AgentInformerWatcherStore[T]) HandleReceivedResource(ctx context.Contex
 			return nil
 		}
 
-		if len(newObj.GetFinalizers()) != 0 {
-			return nil
-		}
-
 		last, exists, err := s.Get(newObj.GetNamespace(), newObj.GetName())
 		if err != nil {
 			return err
@@ -116,6 +106,19 @@ func (s *AgentInformerWatcherStore[T]) HandleReceivedResource(ctx context.Contex
 			return err
 		}
 
+		// trigger an update event if the object is deleting.
+		// Only need to update generation/finalizer/deletionTimeStamp of the object.
+		if len(newObj.GetFinalizers()) != 0 {
+			accessor, err := meta.Accessor(deletingObj)
+			if err != nil {
+				return err
+			}
+			accessor.SetDeletionTimestamp(newObj.GetDeletionTimestamp())
+			accessor.SetFinalizers(newObj.GetFinalizers())
+			accessor.SetGeneration(newObj.GetGeneration())
+			return s.Update(deletingObj)
+		}
+
 		return s.Delete(deletingObj)
 	default:
 		return fmt.Errorf("unsupported resource action %s", action)
diff --git a/pkg/cloudevents/clients/work/source/client/manifestwork.go b/pkg/cloudevents/clients/work/source/client/manifestwork.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strconv"
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -87,7 +88,13 @@ func (c *ManifestWorkSourceClient) Create(ctx context.Context, manifestWork *wor
 	newWork := manifestWork.DeepCopy()
 	newWork.UID = kubetypes.UID(utils.UID(c.sourceID, common.ManifestWorkGR.String(), c.namespace, newWork.Name))
 	newWork.Namespace = c.namespace
-	newWork.ResourceVersion = getWorkResourceVersion(manifestWork)
+
+	rv, generation, err := getWorkResourceVersion(manifestWork)
+	if err != nil {
+		return nil, errors.NewInternalError(err)
+	}
+	newWork.Generation = generation
+	newWork.ResourceVersion = rv
 
 	if err := utils.EncodeManifests(newWork); err != nil {
 		returnErr := errors.NewInternalError(err)
@@ -273,7 +280,12 @@ func (c *ManifestWorkSourceClient) Patch(ctx context.Context, name string, pt ku
 	}
 
 	newWork := patchedWork.DeepCopy()
-	newWork.ResourceVersion = getWorkResourceVersion(patchedWork)
+	rv, generation, err := getWorkResourceVersion(patchedWork)
+	if err != nil {
+		return nil, errors.NewInternalError(err)
+	}
+	newWork.Generation = generation
+	newWork.ResourceVersion = rv
 
 	if errs := utils.ValidateWork(newWork); len(errs) != 0 {
 		returnErr := errors.NewInvalid(common.ManifestWorkGK, name, errs)
@@ -303,15 +315,27 @@ func (c *ManifestWorkSourceClient) Patch(ctx context.Context, name string, pt ku
 // firstly, if no annotation is set, we will get the the resource version from work itself,
 // if the wok does not have it, "0" will be returned, which means the version of the work
 // will not be maintained on source, the message broker guarantees the work update order.
-func getWorkResourceVersion(work *workv1.ManifestWork) string {
+func getWorkResourceVersion(work *workv1.ManifestWork) (string, int64, error) {
+	var generation int64
+	var err error
+
 	resourceVersion, ok := work.Annotations[common.CloudEventsResourceVersionAnnotationKey]
 	if ok {
-		return resourceVersion
+		generation, err = strconv.ParseInt(resourceVersion, 10, 16)
+		if err != nil {
+			return "", 0, errors.NewInternalError(err)
+		}
+	}
+
+	if generation == 0 {
+		generation = work.Generation
 	}
 
-	if work.ResourceVersion != "" {
-		return work.ResourceVersion
+	if len(resourceVersion) == 0 && len(work.ResourceVersion) != 0 {
+		resourceVersion = work.ResourceVersion
+	} else {
+		resourceVersion = "0"
 	}
 
-	return "0"
+	return resourceVersion, generation, nil
 }
diff --git a/pkg/cloudevents/clients/work/source/codec/manifestbundle_test.go b/pkg/cloudevents/clients/work/source/codec/manifestbundle_test.go
@@ -204,8 +204,8 @@ func TestManifestBundleDecode(t *testing.T) {
 			}(),
 			expectedWork: &workv1.ManifestWork{
 				ObjectMeta: metav1.ObjectMeta{
-					UID:             kubetypes.UID("test"),
-					ResourceVersion: "13",
+					UID:        kubetypes.UID("test"),
+					Generation: 13,
 					Annotations: map[string]string{
 						"cloudevents.open-cluster-management.io/sequenceid": "1834773391719010304",
 					},
@@ -225,13 +225,13 @@ func TestManifestBundleDecode(t *testing.T) {
 			name: "decode a manifestbundle status cloudevent with meta and spec",
 			event: func() *cloudevents.Event {
 				metaJson, err := json.Marshal(metav1.ObjectMeta{
-					UID:             kubetypes.UID("test"),
-					ResourceVersion: "13",
-					Name:            "work1",
-					Namespace:       "cluster1",
-					Labels:          map[string]string{"test1": "test1"},
-					Annotations:     map[string]string{"test2": "test2"},
-					Finalizers:      []string{"test"},
+					UID:         kubetypes.UID("test"),
+					Generation:  13,
+					Name:        "work1",
+					Namespace:   "cluster1",
+					Labels:      map[string]string{"test1": "test1"},
+					Annotations: map[string]string{"test2": "test2"},
+					Finalizers:  []string{"test"},
 				})
 				if err != nil {
 					t.Fatal(err)
@@ -290,11 +290,11 @@ func TestManifestBundleDecode(t *testing.T) {
 			}(),
 			expectedWork: &workv1.ManifestWork{
 				ObjectMeta: metav1.ObjectMeta{
-					UID:             kubetypes.UID("test"),
-					ResourceVersion: "13",
-					Name:            "work1",
-					Namespace:       "cluster1",
-					Labels:          map[string]string{"test1": "test1"},
+					UID:        kubetypes.UID("test"),
+					Generation: 13,
+					Name:       "work1",
+					Namespace:  "cluster1",
+					Labels:     map[string]string{"test1": "test1"},
 					Annotations: map[string]string{
 						"cloudevents.open-cluster-management.io/sequenceid": "1834773391719010304",
 						"test2": "test2",
diff --git a/pkg/cloudevents/clients/work/store/base.go b/pkg/cloudevents/clients/work/store/base.go
@@ -3,7 +3,6 @@ package store
 import (
 	"context"
 	"fmt"
-	"strconv"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/equality"
@@ -116,22 +115,10 @@ func (b *workProcessor) handleWork(ctx context.Context, work *workv1.ManifestWor
 		return b.store.Delete(updatedWork)
 	}
 
-	lastResourceVersion, err := strconv.Atoi(lastWork.ResourceVersion)
-	if err != nil {
-		logger.Error(err, "invalid resource version for work")
-		return nil
-	}
-
-	resourceVersion, err := strconv.Atoi(work.ResourceVersion)
-	if err != nil {
-		logger.Error(err, "invalid resource version for work")
-		return nil
-	}
-
 	// the current work's version is maintained on source and the agent's work is newer than source, ignore
-	if lastResourceVersion != 0 && resourceVersion > lastResourceVersion {
-		logger.Info("the work resource version is great than its generation, ignore",
-			"agentResourceVersion", resourceVersion, "sourceResourceVersion", lastResourceVersion)
+	if lastWork.Generation != 0 && work.Generation > lastWork.Generation {
+		logger.Info("the work generation is great than its local generation, ignore",
+			"agentGeneration", lastWork.Generation, "sourceGeneration", work.Generation)
 		return nil
 	}
 
diff --git a/pkg/cloudevents/clients/work/store/informer.go b/pkg/cloudevents/clients/work/store/informer.go
@@ -187,7 +187,6 @@ func (s *AgentInformerWatcherStore) HandleReceivedResource(ctx context.Context,
 		// restore the fields that are maintained by local agent.
 		updatedWork.Finalizers = lastWork.Finalizers
 		updatedWork.Status = lastWork.Status
-
 		return s.Update(updatedWork)
 	case types.Deleted:
 		// the manifestwork is deleting on the source, we just update its deletion timestamp.
@@ -199,9 +198,12 @@ func (s *AgentInformerWatcherStore) HandleReceivedResource(ctx context.Context,
 			return nil
 		}
 
-		// we should only update the deletionTimestamp or the local work
+		// update the deletionTimeStamp and generation of last work.
+		// generation needs to be updated because it is possible that generation still change after
+		// the object is in deleting state.
 		updatedWork := lastWork.DeepCopy()
 		updatedWork.DeletionTimestamp = work.DeletionTimestamp
+		updatedWork.Generation = lastWork.Generation
 		return s.Update(updatedWork)
 	default:
 		return fmt.Errorf("unsupported resource action %s", action)
diff --git a/pkg/cloudevents/generic/clients/agentclient.go b/pkg/cloudevents/generic/clients/agentclient.go
@@ -298,13 +298,13 @@ func (c *CloudEventAgentClient[T]) specAction(
 		return types.Deleted, nil
 	}
 
-	// if both the current and the last object have the resource version "0" or empty, then object
+	// if both the current and the last object have the generation "0" or empty, then object
 	// is considered as modified, the message broker guarantees the order of the messages
 	if lastObj.GetGeneration() == 0 && obj.GetGeneration() == 0 {
 		return types.Modified, nil
 	}
 
-	if obj.GetGeneration() <= lastObj.GetGeneration() {
+	if obj.GetGeneration() < lastObj.GetGeneration() {
 		return evt, nil
 	}
 
diff --git a/pkg/cloudevents/generic/clients/agentclient_test.go b/pkg/cloudevents/generic/clients/agentclient_test.go
@@ -515,8 +515,8 @@ func TestReceiveResourceSpec(t *testing.T) {
 				return *evt
 			}(),
 			resources: []*generictesting.MockResource{
-				{UID: kubetypes.UID("test1"), Generation: 2, ResourceVersion: "2", Namespace: "cluster1"},
-				{UID: kubetypes.UID("test2"), Generation: 1, ResourceVersion: "1", Namespace: "cluster1"},
+				{UID: kubetypes.UID("test1"), Generation: 3, ResourceVersion: "1", Namespace: "cluster1"},
+				{UID: kubetypes.UID("test2"), Generation: 3, ResourceVersion: "1", Namespace: "cluster1"},
 			},
 			validate: func(event types.ResourceAction, resource *generictesting.MockResource) {
 				if len(event) != 0 {
diff --git a/pkg/cloudevents/generic/clients/baseclient.go b/pkg/cloudevents/generic/clients/baseclient.go
@@ -143,7 +143,7 @@ func (c *baseClient) publish(ctx context.Context, evt cloudevents.Event) error {
 	}
 
 	logger.V(2).Info("Sending event", "event", evt.Context)
-	logger.V(5).Info("Sending event", "event", func() any { return evt.String() })
+	logger.V(5).Info("Sending event", "event", evt.String())
 	if err := c.transport.Send(ctx, evt); err != nil {
 		return err
 	}
diff --git a/test/integration-test.mk b/test/integration-test.mk
@@ -31,7 +31,7 @@ integration: test-cloudevents-integration test-basecontroller-integration test-s
 
 test-cloudevents-integration:
 	go test -c ./test/integration/cloudevents
-	./cloudevents.test -ginkgo.slowSpecThreshold=15 -ginkgo.v -ginkgo.failFast
+	./cloudevents.test -ginkgo.slowSpecThreshold=15 -ginkgo.v -ginkgo.failFast -v=5
 .PHONY: test-cloudevents-integration
 
 test-basecontroller-integration: ensure-kubebuilder-tools
diff --git a/test/integration/cloudevents/cloudevents_resync_test.go b/test/integration/cloudevents/cloudevents_resync_test.go
@@ -6,15 +6,13 @@ import (
 	"fmt"
 	jsonpatch "github.com/evanphx/json-patch/v5"
 	"k8s.io/apimachinery/pkg/types"
-	workinformers "open-cluster-management.io/api/client/work/informers/externalversions"
 	"time"
 
 	"github.com/onsi/ginkgo"
 	"github.com/onsi/gomega"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/rand"
 
 	workv1 "open-cluster-management.io/api/work/v1"
@@ -82,29 +80,20 @@ var _ = ginkgo.Describe("CloudEvents Clients Test - RESYNC", func() {
 			gomega.Expect(err).ToNot(gomega.HaveOccurred())
 
 			ginkgo.By("start a work agent to resync the resources from agent")
-			clientHolder, _, err := agent.StartWorkAgent(ctx, clusterName, mqttOptions, codec.NewManifestBundleCodec())
-
-			factory := workinformers.NewSharedInformerFactoryWithOptions(
-				clientHolder.WorkInterface(),
-				5*time.Minute,
-				workinformers.WithNamespace(clusterName),
-			)
-			informer := factory.Work().V1().ManifestWorks()
-			go informer.Informer().Run(ctx.Done())
+			clientHolder, watchStore, err := agent.StartWorkAgent(ctx, clusterName, mqttOptions, codec.NewManifestBundleCodec())
 
 			gomega.Expect(err).ToNot(gomega.HaveOccurred())
-			lister := informer.Lister().ManifestWorks(clusterName)
 			agentWorkClient := clientHolder.ManifestWorks(clusterName)
 
 			ginkgo.By("ensure the resources is synced on the agent")
 			gomega.Eventually(func() error {
-				list, err := lister.List(labels.Everything())
+				list, err := watchStore.List(clusterName, metav1.ListOptions{})
 				if err != nil {
 					return err
 				}
 
 				// ensure there is only one work was synced on the cluster1
-				if len(list) != 1 {
+				if len(list.Items) != 1 {
 					return fmt.Errorf("unexpected work list %v", list)
 				}
 
@@ -131,6 +120,9 @@ var _ = ginkgo.Describe("CloudEvents Clients Test - RESYNC", func() {
 					return err
 				}
 				patchData, err := jsonpatch.CreateMergePatch(workData, newWorkData)
+				if err != nil {
+					return err
+				}
 
 				// only update the status on the agent local part
 				if _, err := agentWorkClient.Patch(context.Background(), workName, types.MergePatchType, patchData, metav1.PatchOptions{}); err != nil {
diff --git a/test/integration/cloudevents/manifestworkclients_reync_test.go b/test/integration/cloudevents/manifestworkclients_reync_test.go
@@ -58,7 +58,7 @@ var _ = ginkgo.Describe("ManifestWork Clients Test - Resync", func() {
 			work2UID := utils.UID(sourceID, common.ManifestWorkGR.String(), clusterName, work2Name)
 			work2 := util.NewManifestWorkWithStatus(clusterName, work2Name)
 			work2.UID = apitypes.UID(work2UID)
-			work2.ResourceVersion = "1"
+			work2.Generation = 1
 			work2.Labels = map[string]string{common.CloudEventsOriginalSourceLabelKey: sourceID}
 			work2.Annotations = map[string]string{common.CloudEventsDataTypeAnnotationKey: payload.ManifestBundleEventDataType.String()}
 			gomega.Expect(watchStore.Add(work2)).ToNot(gomega.HaveOccurred())
diff --git a/test/integration/cloudevents/util/work.go b/test/integration/cloudevents/util/work.go
@@ -178,8 +178,9 @@ func UpdateWorkStatus(ctx context.Context, workClient workv1client.ManifestWorkI
 func NewManifestWork(namespace, name string, withVersion bool) *workv1.ManifestWork {
 	work := &workv1.ManifestWork{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
+			Name:       name,
+			Namespace:  namespace,
+			Generation: 1,
 		},
 		Spec: workv1.ManifestWorkSpec{
 			Workload: workv1.ManifestsTemplate{

Original file line number	Diff line number	Diff line change
`@@ -298,13 +298,13 @@ func (c *CloudEventAgentClient[T]) specAction(`
`298`	`298`	`return types.Deleted, nil`
`299`	`299`	`}`
`300`	`300`
`301`		`- // if both the current and the last object have the resource version "0" or empty, then object`
	`301`	`+ // if both the current and the last object have the generation "0" or empty, then object`
`302`	`302`	`// is considered as modified, the message broker guarantees the order of the messages`
`303`	`303`	`if lastObj.GetGeneration() == 0 && obj.GetGeneration() == 0 {`
`304`	`304`	`return types.Modified, nil`
`305`	`305`	`}`
`306`	`306`
`307`		`- if obj.GetGeneration() <= lastObj.GetGeneration() {`
	`307`	`+ if obj.GetGeneration() < lastObj.GetGeneration() {`
`308`	`308`	`return evt, nil`
`309`	`309`	`}`
`310`	`310`
Original file line number	Diff line number	Diff line change
`@@ -143,7 +143,7 @@ func (c *baseClient) publish(ctx context.Context, evt cloudevents.Event) error {`
`143`	`143`	`}`
`144`	`144`
`145`	`145`	`logger.V(2).Info("Sending event", "event", evt.Context)`
`146`		`- logger.V(5).Info("Sending event", "event", func() any { return evt.String() })`
	`146`	`+ logger.V(5).Info("Sending event", "event", evt.String())`
`147`	`147`	`if err := c.transport.Send(ctx, evt); err != nil {`
`148`	`148`	`return err`
`149`	`149`	`}`