Add getGeneration in the interface

Signed-off-by: Jian Qiu <jqiu@redhat.com>

Author: qiujian16

pkg/cloudevents/clients/store/informer.go

@@ -37,17 +37,17 @@ func NewAgentInformerWatcherStore[T generic.ResourceObject]() *AgentInformerWatc
 
 func (s *AgentInformerWatcherStore[T]) Add(resource runtime.Object) error {
 	s.Watcher.Receive(watch.Event{Type: watch.Added, Object: resource})
-	return nil
+	return s.Store.Add(resource)
 }
 
 func (s *AgentInformerWatcherStore[T]) Update(resource runtime.Object) error {
 	s.Watcher.Receive(watch.Event{Type: watch.Modified, Object: resource})
-	return nil
+	return s.Store.Update(resource)
 }
 
 func (s *AgentInformerWatcherStore[T]) Delete(resource runtime.Object) error {
 	s.Watcher.Receive(watch.Event{Type: watch.Deleted, Object: resource})
-	return nil
+	return s.Store.Delete(resource)
 }
 
 func (s *AgentInformerWatcherStore[T]) HandleReceivedResource(action types.ResourceAction, resource T) error {

pkg/cloudevents/clients/work/agent/client/manifestwork.go

@@ -125,7 +125,8 @@ func (c *ManifestWorkAgentClient) Watch(ctx context.Context, opts metav1.ListOpt
 }
 
 func (c *ManifestWorkAgentClient) Patch(ctx context.Context, name string, pt kubetypes.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *workv1.ManifestWork, err error) {
-	klog.V(4).Infof("patching manifestwork %s/%s", c.namespace, name)
+	logger := klog.FromContext(ctx)
+	logger.V(4).Info("patching manifestwork")
 
 	// avoid race conditions among the agent's go routines
 	c.Lock()
@@ -175,8 +176,7 @@ func (c *ManifestWorkAgentClient) Patch(ctx context.Context, name string, pt kub
 		Action:              types.UpdateRequestAction,
 	}
 
-	// we first compare local resource version
-	if returnErr = versoinCompare(patchedWork, lastWork); returnErr != nil {
+	if returnErr = versionCompare(patchedWork, lastWork); returnErr != nil {
 		return nil, returnErr
 	}
 
@@ -232,7 +232,8 @@ func (c *ManifestWorkAgentClient) Patch(ctx context.Context, name string, pt kub
 		returnErr = errors.NewNotFound(common.ManifestWorkGR, name)
 		return nil, returnErr
 	}
-	if returnErr = versoinCompare(patchedWork, latestWork); returnErr != nil {
+
+	if returnErr = versionCompare(patchedWork, latestWork); returnErr != nil {
 		return nil, returnErr
 	}
 	if err := c.watcherStore.Update(newWork); err != nil {
@@ -242,7 +243,7 @@ func (c *ManifestWorkAgentClient) Patch(ctx context.Context, name string, pt kub
 	return newWork, nil
 }
 
-func versoinCompare(new, old *workv1.ManifestWork) *errors.StatusError {
+func versionCompare(new, old *workv1.ManifestWork) *errors.StatusError {
 	// If resource version is empty or 0, skip the comparison (e.g., when publishing status updates)
 	if new.GetResourceVersion() == "" || new.GetResourceVersion() == "0" {
 		return nil
@@ -256,6 +257,7 @@ func versoinCompare(new, old *workv1.ManifestWork) *errors.StatusError {
 	if err != nil {
 		return errors.NewInternalError(err)
 	}
+
 	// ensure the resource version of the work is not outdated
 	if newResourceVersion < lastResourceVersion {
 		// It's safe to return a conflict error here, even if the status update event

pkg/cloudevents/clients/work/agent/client/manifestwork_test.go

@@ -111,7 +111,7 @@ func TestVersionCompare(t *testing.T) {
 				},
 			}
 
-			err := versoinCompare(newWork, oldWork)
+			err := versionCompare(newWork, oldWork)
 
 			if tt.expectError {
 				if err == nil {

pkg/cloudevents/clients/work/agent/codec/manifestbundle.go

@@ -3,8 +3,6 @@ package codec
 import (
 	"encoding/json"
 	"fmt"
-	"strconv"
-
 	"github.com/bwmarrin/snowflake"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cloudeventstypes "github.com/cloudevents/sdk-go/v2/types"
@@ -50,16 +48,6 @@ func (c *ManifestBundleCodec) Encode(source string, eventType types.CloudEventsT
 		return nil, fmt.Errorf("unsupported cloudevents data type %s", eventType.CloudEventsDataType)
 	}
 
-	var resourceVersion int64
-	// If resource version is empty, use 0 as default (e.g., when publishing status updates)
-	if work.ResourceVersion != "" {
-		var err error
-		resourceVersion, err = strconv.ParseInt(work.ResourceVersion, 10, 64)
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse the resourceversion of the work %s, %v", work.UID, err)
-		}
-	}
-
 	originalSource, ok := work.Labels[common.CloudEventsOriginalSourceLabelKey]
 	if !ok {
 		return nil, fmt.Errorf("failed to find originalsource from the work %s", work.UID)
@@ -68,7 +56,7 @@ func (c *ManifestBundleCodec) Encode(source string, eventType types.CloudEventsT
 	evt := types.NewEventBuilder(source, eventType).
 		WithResourceID(string(work.UID)).
 		WithStatusUpdateSequenceID(sequenceGenerator.Generate().String()).
-		WithResourceVersion(resourceVersion).
+		WithResourceVersion(work.Generation).
 		WithClusterName(work.Namespace).
 		WithOriginalSource(originalSource).
 		NewEvent()
@@ -144,7 +132,7 @@ func (c *ManifestBundleCodec) Decode(evt *cloudevents.Event) (*workv1.ManifestWo
 		metaObj.Name = resourceID
 	}
 	metaObj.Namespace = clusterName
-	// This is explicitly set to empy since it will be managed by local client.
+	// This is explicitly set to empty since it will be managed by local client.
 	metaObj.ResourceVersion = ""
 	// The resourceVersion in cloudevent actually sematically equals to generation, since it increments when
 	// spec changes

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

@@ -111,7 +111,7 @@ func (v *versioner) increment(name string) int64 {
 	defer v.lock.Unlock()
 
 	if _, ok := v.versions[name]; !ok {
-		v.versions[name] = 0
+		v.versions[name] = 1
 	} else {
 		v.versions[name] = v.versions[name] + 1
 	}
@@ -184,9 +184,7 @@ func (s *AgentInformerWatcherStore) HandleReceivedResource(action types.Resource
 
 		updatedWork := work.DeepCopy()
 
-		// restore the fields that are maintained by local agent
-		updatedWork.Labels = lastWork.Labels
-		updatedWork.Annotations = lastWork.Annotations
+		// restore the fields that are maintained by local agent.
 		updatedWork.Finalizers = lastWork.Finalizers
 		updatedWork.Status = lastWork.Status
 
@@ -201,9 +199,8 @@ func (s *AgentInformerWatcherStore) HandleReceivedResource(action types.Resource
 			return nil
 		}
 
+		// we should only update the deletionTimestamp or the local work
 		updatedWork := lastWork.DeepCopy()
-		updatedWork.Generation = work.Generation
-		updatedWork.ResourceVersion = work.ResourceVersion
 		updatedWork.DeletionTimestamp = work.DeletionTimestamp
 		return s.Update(updatedWork)
 	default:

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

@@ -20,23 +20,23 @@ func TestVersioner(t *testing.T) {
 	t.Run("increment version for new resource", func(t *testing.T) {
 		v := newVersioner()
 		version := v.increment("test-work")
-		if version != 0 {
-			t.Errorf("expected version 0 for new resource, got %d", version)
+		if version != 1 {
+			t.Errorf("expected version 1 for new resource, got %d", version)
 		}
 	})
 
 	t.Run("increment version for existing resource", func(t *testing.T) {
 		v := newVersioner()
 		v.increment("test-work")
 		version := v.increment("test-work")
-		if version != 1 {
-			t.Errorf("expected version 1 for existing resource, got %d", version)
+		if version != 2 {
+			t.Errorf("expected version 2 for existing resource, got %d", version)
 		}
 	})
 
 	t.Run("increment multiple times", func(t *testing.T) {
 		v := newVersioner()
-		for i := 0; i < 5; i++ {
+		for i := 1; i <= 5; i++ {
 			version := v.increment("test-work")
 			if version != int64(i) {
 				t.Errorf("expected version %d, got %d", i, version)
@@ -49,8 +49,8 @@ func TestVersioner(t *testing.T) {
 		v.increment("test-work")
 		v.delete("test-work")
 		version := v.increment("test-work")
-		if version != 0 {
-			t.Errorf("expected version 0 after delete, got %d", version)
+		if version != 1 {
+			t.Errorf("expected version 1 after delete, got %d", version)
 		}
 	})
 
@@ -70,8 +70,8 @@ func TestVersioner(t *testing.T) {
 		}
 
 		finalVersion := v.increment("test-work")
-		if finalVersion != 10 {
-			t.Errorf("expected version 10 after 10 concurrent increments, got %d", finalVersion)
+		if finalVersion != 11 {
+			t.Errorf("expected version 11 after 10 concurrent increments, got %d", finalVersion)
 		}
 	})
 }
@@ -118,8 +118,8 @@ func TestAgentInformerWatcherStore_Add(t *testing.T) {
 		t.Fatalf("unexpected error adding work: %v", err)
 	}
 
-	if work.ResourceVersion != "0" {
-		t.Errorf("expected resource version 0, got %s", work.ResourceVersion)
+	if work.ResourceVersion != "1" {
+		t.Errorf("expected resource version 1, got %s", work.ResourceVersion)
 	}
 
 	// Add another work
@@ -135,8 +135,8 @@ func TestAgentInformerWatcherStore_Add(t *testing.T) {
 		t.Fatalf("unexpected error adding work: %v", err)
 	}
 
-	if work2.ResourceVersion != "1" {
-		t.Errorf("expected resource version 1 for second add, got %s", work2.ResourceVersion)
+	if work2.ResourceVersion != "2" {
+		t.Errorf("expected resource version 2 for second add, got %s", work2.ResourceVersion)
 	}
 }
 
@@ -198,8 +198,8 @@ func TestAgentInformerWatcherStore_Update(t *testing.T) {
 		t.Fatalf("unexpected error updating work: %v", err)
 	}
 
-	if updatedWork.ResourceVersion != "1" {
-		t.Errorf("expected resource version 1 after update, got %s", updatedWork.ResourceVersion)
+	if updatedWork.ResourceVersion != "2" {
+		t.Errorf("expected resource version 2 after update, got %s", updatedWork.ResourceVersion)
 	}
 
 	// Update again
@@ -208,8 +208,8 @@ func TestAgentInformerWatcherStore_Update(t *testing.T) {
 		t.Fatalf("unexpected error updating work: %v", err)
 	}
 
-	if updatedWork.ResourceVersion != "2" {
-		t.Errorf("expected resource version 2 after second update, got %s", updatedWork.ResourceVersion)
+	if updatedWork.ResourceVersion != "3" {
+		t.Errorf("expected resource version 3 after second update, got %s", updatedWork.ResourceVersion)
 	}
 }
 
@@ -267,7 +267,7 @@ func TestAgentInformerWatcherStore_Delete(t *testing.T) {
 		t.Fatalf("unexpected error deleting work: %v", err)
 	}
 
-	// Add the same work again, version should be reset to 0
+	// Add the same work again, version should be reset to 1
 	newWork := &workv1.ManifestWork{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-work",
@@ -280,8 +280,8 @@ func TestAgentInformerWatcherStore_Delete(t *testing.T) {
 		t.Fatalf("unexpected error adding work after delete: %v", err)
 	}
 
-	if newWork.ResourceVersion != "0" {
-		t.Errorf("expected resource version 0 after delete and re-add, got %s", newWork.ResourceVersion)
+	if newWork.ResourceVersion != "1" {
+		t.Errorf("expected resource version 1 after delete and re-add, got %s", newWork.ResourceVersion)
 	}
 }
 
@@ -323,7 +323,7 @@ func TestAgentInformerWatcherStore_ResourceVersionIncrement(t *testing.T) {
 	}
 
 	// Verify resource versions increment properly
-	expectedVersions := []string{"0", "1", "2", "3"}
+	expectedVersions := []string{"1", "2", "3", "4"}
 
 	// Add
 	err = store.Add(work)

pkg/cloudevents/generic/clients/agentclient.go

@@ -3,7 +3,6 @@ package clients
 import (
 	"context"
 	"fmt"
-	"strconv"
 	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -84,19 +83,10 @@ func (c *CloudEventAgentClient[T]) Resync(ctx context.Context, source string) er
 
 	resources := &payload.ResourceVersionList{Versions: make([]payload.ResourceVersion, len(objs))}
 	for i, obj := range objs {
-		var resourceVersion int64
-		// If resource version is empty, use 0 as default
-		if obj.GetResourceVersion() != "" {
-			var err error
-			resourceVersion, err = strconv.ParseInt(obj.GetResourceVersion(), 10, 64)
-			if err != nil {
-				return err
-			}
-		}
-
 		resources.Versions[i] = payload.ResourceVersion{
-			ResourceID:      string(obj.GetUID()),
-			ResourceVersion: resourceVersion,
+			ResourceID: string(obj.GetUID()),
+			// this should be set as generation, since the resource version of the object is local version.
+			ResourceVersion: obj.GetGeneration(),
 		}
 	}
 
@@ -304,27 +294,11 @@ func (c *CloudEventAgentClient[T]) specAction(
 
 	// if both the current and the last object have the resource version "0" or empty, then object
 	// is considered as modified, the message broker guarantees the order of the messages
-	if (obj.GetResourceVersion() == "0" || obj.GetResourceVersion() == "") &&
-		(lastObj.GetResourceVersion() == "0" || lastObj.GetResourceVersion() == "") {
+	if lastObj.GetGeneration() == 0 && obj.GetGeneration() == 0 {
 		return types.Modified, nil
 	}
 
-	// If resource version is empty, treat as modified (cannot compare versions)
-	if obj.GetResourceVersion() == "" || lastObj.GetResourceVersion() == "" {
-		return types.Modified, nil
-	}
-
-	resourceVersion, err := strconv.ParseInt(obj.GetResourceVersion(), 10, 64)
-	if err != nil {
-		return evt, err
-	}
-
-	lastResourceVersion, err := strconv.ParseInt(lastObj.GetResourceVersion(), 10, 64)
-	if err != nil {
-		return evt, err
-	}
-
-	if resourceVersion <= lastResourceVersion {
+	if obj.GetGeneration() <= lastObj.GetGeneration() {
 		return evt, nil
 	}
 

pkg/cloudevents/generic/clients/agentclient_test.go

@@ -122,6 +122,7 @@ func TestAgentPublish(t *testing.T) {
 			clusterName: "cluster1",
 			resources: &generictesting.MockResource{
 				UID:             kubetypes.UID("1234"),
+				Generation:      2,
 				ResourceVersion: "2",
 				Status:          "test-status",
 				Namespace:       "cluster1",
@@ -409,6 +410,7 @@ func TestReceiveResourceSpec(t *testing.T) {
 					eventType,
 					&generictesting.MockResource{
 						UID:             kubetypes.UID("test1"),
+						Generation:      1,
 						ResourceVersion: "1",
 						Namespace:       "cluster1",
 					})
@@ -435,6 +437,7 @@ func TestReceiveResourceSpec(t *testing.T) {
 					eventType,
 					&generictesting.MockResource{
 						UID:             kubetypes.UID("test1"),
+						Generation:      2,
 						ResourceVersion: "2",
 						Namespace:       "cluster1",
 					})
@@ -505,14 +508,15 @@ func TestReceiveResourceSpec(t *testing.T) {
 					eventType,
 					&generictesting.MockResource{
 						UID:             kubetypes.UID("test1"),
+						Generation:      2,
 						ResourceVersion: "2",
 						Namespace:       "cluster1",
 					})
 				return *evt
 			}(),
 			resources: []*generictesting.MockResource{
-				{UID: kubetypes.UID("test1"), ResourceVersion: "2", Namespace: "cluster1"},
-				{UID: kubetypes.UID("test2"), ResourceVersion: "1", Namespace: "cluster1"},
+				{UID: kubetypes.UID("test1"), Generation: 2, ResourceVersion: "2", Namespace: "cluster1"},
+				{UID: kubetypes.UID("test2"), Generation: 1, ResourceVersion: "1", Namespace: "cluster1"},
 			},
 			validate: func(event types.ResourceAction, resource *generictesting.MockResource) {
 				if len(event) != 0 {

pkg/cloudevents/generic/clients/sourceclient_test.go

@@ -105,6 +105,7 @@ func TestSourcePublish(t *testing.T) {
 			resources: &generictesting.MockResource{
 				UID:             kubetypes.UID("1234"),
 				ResourceVersion: "2",
+				Generation:      2,
 				Spec:            "test-spec",
 			},
 			eventType: types.CloudEventsType{

pkg/cloudevents/generic/interface.go

@@ -25,8 +25,12 @@ type ResourceObject interface {
 	// GetResourceVersion returns the resource version of this object. The resource version is a required int64 sequence
 	// number property that must be incremented by the source whenever this resource changes.
 	// The source should guarantee its incremental nature.
+	// Deprecated: use GetGeneration() instead.
 	GetResourceVersion() string
 
+	// GetGeneration returns the generation number of this object to reflect the spec change of the resource.
+	GetGeneration() int64
+
 	// GetDeletionTimestamp returns the deletion timestamp of this object. The deletiontimestamp is an optional
 	// timestamp property representing the resource is deleting from the source, the agent needs to clean up the
 	// resource from its cluster.