Add getGeneration in the interface

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

Author: qiujian16

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

@@ -256,6 +256,8 @@ func versoinCompare(new, old *workv1.ManifestWork) *errors.StatusError {
 	if err != nil {
 		return errors.NewInternalError(err)
 	}
+
+	klog.Infof("last resource version is %d, resource version is %d", lastResourceVersion, newResourceVersion)
 	// 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/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()

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
 	}
@@ -143,6 +143,7 @@ func (s *AgentInformerWatcherStore) Add(resource runtime.Object) error {
 		return err
 	}
 	accessor.SetResourceVersion(strconv.FormatInt(s.versions.increment(accessor.GetName()), 10))
+	klog.Infof("inc resource version to %s because of adding", accessor.GetResourceVersion())
 	return s.AgentInformerWatcherStore.Add(resource)
 }
 
@@ -152,6 +153,7 @@ func (s *AgentInformerWatcherStore) Update(resource runtime.Object) error {
 		return err
 	}
 	accessor.SetResourceVersion(strconv.FormatInt(s.versions.increment(accessor.GetName()), 10))
+	klog.Infof("inc resource version to %s because of updating", accessor.GetResourceVersion())
 	return s.AgentInformerWatcherStore.Update(resource)
 }
 

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"
@@ -85,19 +84,9 @@ 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,
+			ResourceVersion: obj.GetGeneration(),
 		}
 	}
 
@@ -305,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

@@ -116,6 +116,7 @@ func TestAgentPublish(t *testing.T) {
 			clusterName: "cluster1",
 			resources: &generictesting.MockResource{
 				UID:             kubetypes.UID("1234"),
+				Generation:      2,
 				ResourceVersion: "2",
 				Status:          "test-status",
 				Namespace:       "cluster1",
@@ -396,6 +397,7 @@ func TestReceiveResourceSpec(t *testing.T) {
 					eventType,
 					&generictesting.MockResource{
 						UID:             kubetypes.UID("test1"),
+						Generation:      1,
 						ResourceVersion: "1",
 						Namespace:       "cluster1",
 					})
@@ -422,6 +424,7 @@ func TestReceiveResourceSpec(t *testing.T) {
 					eventType,
 					&generictesting.MockResource{
 						UID:             kubetypes.UID("test1"),
+						Generation:      2,
 						ResourceVersion: "2",
 						Namespace:       "cluster1",
 					})
@@ -492,14 +495,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

@@ -99,6 +99,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.

pkg/cloudevents/generic/testing/resource.go

@@ -2,6 +2,7 @@ package testing
 
 import (
 	"fmt"
+	"strconv"
 	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -22,6 +23,7 @@ var MockEventDataType = types.CloudEventsDataType{
 type MockResource struct {
 	UID               kubetypes.UID `json:"uid"`
 	ResourceVersion   string        `json:"resourceVersion"`
+	Generation        int64         `json:"generation"`
 	DeletionTimestamp *metav1.Time  `json:"deletionTimestamp,omitempty"`
 	Namespace         string
 	Spec              string `json:"spec"`
@@ -36,6 +38,10 @@ func (r *MockResource) GetResourceVersion() string {
 	return r.ResourceVersion
 }
 
+func (r *MockResource) GetGeneration() int64 {
+	return r.Generation
+}
+
 func (r *MockResource) GetDeletionTimestamp() *metav1.Time {
 	return r.DeletionTimestamp
 }
@@ -75,7 +81,7 @@ func (c *MockResourceCodec) Encode(source string, eventType types.CloudEventsTyp
 	evt.SetType(eventType.String())
 	evt.SetTime(time.Now())
 	evt.SetExtension("resourceid", string(obj.UID))
-	evt.SetExtension("resourceversion", obj.ResourceVersion)
+	evt.SetExtension("resourceversion", strconv.FormatInt(obj.Generation, 10))
 	evt.SetExtension("clustername", obj.Namespace)
 	if obj.GetDeletionTimestamp() != nil {
 		evt.SetExtension("deletiontimestamp", obj.DeletionTimestamp.Time)
@@ -97,9 +103,15 @@ func (c *MockResourceCodec) Decode(evt *cloudevents.Event) (*MockResource, error
 		return nil, fmt.Errorf("failed to get resource version: %v", err)
 	}
 
+	generation, err := strconv.ParseInt(fmt.Sprintf("%s", resourceVersion), 10, 64)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse resource generation: %v", err)
+	}
+
 	res := &MockResource{
 		UID:             kubetypes.UID(fmt.Sprintf("%s", resourceID)),
 		ResourceVersion: fmt.Sprintf("%s", resourceVersion),
+		Generation:      generation,
 		Status:          string(evt.Data()),
 	}
 

test/integration/cloudevents/manifestworkclients_watch_test.go

@@ -185,10 +185,18 @@ var _ = ginkgo.Describe("ManifestWork Clients Test - Watch Only", func() {
 				gomega.Eventually(func() error {
 					workClient := agentClient.ManifestWorks(clusterName)
 
+					work, err := workClient.Get(ctx, workName, metav1.GetOptions{})
+					gomega.Expect(err).ToNot(gomega.HaveOccurred())
+					ginkgo.By(fmt.Sprintf("resourceVersion is %s", work.ResourceVersion))
+
 					if err := util.AddWorkFinalizer(ctx, workClient, workName); err != nil {
 						return err
 					}
 
+					work, err = workClient.Get(ctx, workName, metav1.GetOptions{})
+					gomega.Expect(err).ToNot(gomega.HaveOccurred())
+					ginkgo.By(fmt.Sprintf("resourceVersion is %s after patch", work.ResourceVersion))
+
 					return util.UpdateWorkStatus(ctx, workClient, workName, util.WorkCreatedCondition)
 				}, 10*time.Second, 1*time.Second).Should(gomega.Succeed())