Fix integration tests

Signed-off-by: Jian Qiu <[email protected]>

Author: qiujian16

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

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"
@@ -52,27 +53,34 @@ func (c *ManifestWorkSourceClient) SetNamespace(namespace string) {
 }
 
 func (c *ManifestWorkSourceClient) Create(ctx context.Context, manifestWork *workv1.ManifestWork, opts metav1.CreateOptions) (*workv1.ManifestWork, error) {
+	var returnErr *errors.StatusError
+
+	defer func() {
+		if returnErr != nil {
+			metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
+		} else {
+			metrics.IncreaseWorkProcessedCounter("create", metav1.StatusSuccess)
+		}
+	}()
+
 	if manifestWork.Namespace != "" && manifestWork.Namespace != c.namespace {
-		returnErr := errors.NewInvalid(common.ManifestWorkGK, manifestWork.Name, field.ErrorList{
+		returnErr = errors.NewInvalid(common.ManifestWorkGK, manifestWork.Name, field.ErrorList{
 			field.Invalid(
 				field.NewPath("metadata").Child("namespace"),
 				manifestWork.Namespace,
 				fmt.Sprintf("does not match the namespace %s", c.namespace),
 			),
 		})
-		metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
 		return nil, returnErr
 	}
 
 	_, exists, err := c.watcherStore.Get(c.namespace, manifestWork.Name)
 	if err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 	if exists {
-		returnErr := errors.NewAlreadyExists(common.ManifestWorkGR, manifestWork.Name)
-		metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewAlreadyExists(common.ManifestWorkGR, manifestWork.Name)
 		return nil, returnErr
 	}
 
@@ -87,34 +95,36 @@ 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 {
+		returnErr = errors.NewInternalError(err)
+		return nil, returnErr
+	}
+	newWork.Generation = generation
+	newWork.ResourceVersion = rv
 
 	if err := utils.EncodeManifests(newWork); err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 
 	if errs := utils.ValidateWork(newWork); len(errs) != 0 {
-		returnErr := errors.NewInvalid(common.ManifestWorkGK, manifestWork.Name, errs)
-		metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInvalid(common.ManifestWorkGK, manifestWork.Name, errs)
 		return nil, returnErr
 	}
 
 	if err := c.cloudEventsClient.Publish(ctx, eventType, newWork); err != nil {
-		returnErr := cloudeventserrors.ToStatusError(common.ManifestWorkGR, manifestWork.Name, err)
-		metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
+		returnErr = cloudeventserrors.ToStatusError(common.ManifestWorkGR, manifestWork.Name, err)
 		return nil, returnErr
 	}
 
 	// add the new work to the local cache.
 	if err := c.watcherStore.Add(newWork); err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("create", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 
-	metrics.IncreaseWorkProcessedCounter("create", metav1.StatusSuccess)
 	return newWork.DeepCopy(), nil
 }
 
@@ -238,29 +248,34 @@ func (c *ManifestWorkSourceClient) Patch(ctx context.Context, name string, pt ku
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("patching manifestwork", "manifestWorkName", name)
 
+	var returnErr *errors.StatusError
+	defer func() {
+		if returnErr != nil {
+			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		} else {
+			metrics.IncreaseWorkProcessedCounter("patch", metav1.StatusSuccess)
+		}
+	}()
+
 	if len(subresources) != 0 {
 		msg := fmt.Sprintf("unsupported to update subresources %v", subresources)
-		returnErr := errors.NewGenericServerResponse(http.StatusMethodNotAllowed, "patch", common.ManifestWorkGR, name, msg, 0, false)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewGenericServerResponse(http.StatusMethodNotAllowed, "patch", common.ManifestWorkGR, name, msg, 0, false)
 		return nil, returnErr
 	}
 
 	lastWork, exists, err := c.watcherStore.Get(c.namespace, name)
 	if err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 	if !exists {
-		returnErr := errors.NewNotFound(common.ManifestWorkGR, name)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewNotFound(common.ManifestWorkGR, name)
 		return nil, returnErr
 	}
 
 	patchedWork, err := utils.Patch(pt, lastWork, data)
 	if err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 
@@ -273,28 +288,30 @@ 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 {
+		returnErr = errors.NewInternalError(err)
+		return nil, returnErr
+	}
+	newWork.Generation = generation
+	newWork.ResourceVersion = rv
 
 	if errs := utils.ValidateWork(newWork); len(errs) != 0 {
-		returnErr := errors.NewInvalid(common.ManifestWorkGK, name, errs)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInvalid(common.ManifestWorkGK, name, errs)
 		return nil, returnErr
 	}
 
 	if err := c.cloudEventsClient.Publish(ctx, eventType, newWork); err != nil {
-		returnErr := cloudeventserrors.ToStatusError(common.ManifestWorkGR, name, err)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = cloudeventserrors.ToStatusError(common.ManifestWorkGR, name, err)
 		return nil, returnErr
 	}
 
 	// modify the updated work in the local cache.
 	if err := c.watcherStore.Update(newWork); err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 
-	metrics.IncreaseWorkProcessedCounter("patch", metav1.StatusSuccess)
 	return newWork.DeepCopy(), nil
 }
 
@@ -303,15 +320,29 @@ 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, 64)
+		if err != nil {
+			return "", 0, errors.NewInternalError(err)
+		}
 	}
 
-	if work.ResourceVersion != "" {
-		return work.ResourceVersion
+	if generation == 0 {
+		generation = work.Generation
+	}
+
+	if len(resourceVersion) == 0 {
+		if len(work.ResourceVersion) != 0 {
+			resourceVersion = work.ResourceVersion
+		} else {
+			resourceVersion = "0"
+		}
 	}
 
-	return "0"
+	return resourceVersion, generation, nil
 }

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",

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 greater than its local generation, ignore",
+			"localGeneration", lastWork.Generation, "remoteGeneration", work.Generation)
 		return nil
 	}
 

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 = work.Generation
 		return s.Update(updatedWork)
 	default:
 		return fmt.Errorf("unsupported resource action %s", action)