add pubsub support for cloudevents transport.

Signed-off-by: morvencao <[email protected]>

rh-pre-commit.version: 2.3.2
rh-pre-commit.check-secrets: ENABLED

Author: morvencao

go.mod

@@ -3,6 +3,7 @@ module open-cluster-management.io/sdk-go
 go 1.24.0
 
 require (
+	cloud.google.com/go/pubsub/v2 v2.0.0
 	github.com/bwmarrin/snowflake v0.3.0
 	github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-20250922144431-372892d7c84d
 	github.com/cloudevents/sdk-go/v2 v2.16.2
@@ -24,9 +25,10 @@ require (
 	github.com/prometheus/client_model v0.6.1
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.11.1
-	golang.org/x/oauth2 v0.27.0
-	google.golang.org/grpc v1.68.1
-	google.golang.org/protobuf v1.36.5
+	golang.org/x/oauth2 v0.30.0
+	google.golang.org/api v0.247.0
+	google.golang.org/grpc v1.74.2
+	google.golang.org/protobuf v1.36.7
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.33.2
 	k8s.io/apimachinery v0.33.2
@@ -41,21 +43,30 @@ require (
 )
 
 require (
-	cel.dev/expr v0.23.1 // indirect
-	cloud.google.com/go/compute/metadata v0.5.0 // indirect
+	cel.dev/expr v0.24.0 // indirect
+	cloud.google.com/go v0.121.6 // indirect
+	cloud.google.com/go/auth v0.16.4 // indirect
+	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
+	cloud.google.com/go/compute/metadata v0.8.0 // indirect
+	cloud.google.com/go/iam v1.5.2 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.23.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/gnostic-models v0.6.9 // indirect
+	github.com/google/s2a-go v0.1.9 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.6 // indirect
+	github.com/googleapis/gax-go/v2 v2.15.0 // indirect
 	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
@@ -71,20 +82,30 @@ require (
 	github.com/prometheus/common v0.62.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/rs/xid v1.4.0 // indirect
-	github.com/stoewer/go-strcase v1.3.0 // indirect
+	github.com/stoewer/go-strcase v1.3.1 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
-	go.opentelemetry.io/otel v1.33.0 // indirect
-	go.opentelemetry.io/otel/trace v1.33.0 // indirect
+	go.einride.tech/aip v0.73.0 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect
+	go.opentelemetry.io/otel v1.36.0 // indirect
+	go.opentelemetry.io/otel/metric v1.36.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.36.0 // indirect
+	go.opentelemetry.io/otel/trace v1.36.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/crypto v0.41.0 // indirect
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/net v0.43.0 // indirect
+	golang.org/x/sync v0.16.0 // indirect
 	golang.org/x/sys v0.35.0 // indirect
 	golang.org/x/term v0.34.0 // indirect
 	golang.org/x/text v0.28.0 // indirect
 	golang.org/x/time v0.12.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20241209162323-e6fa225c2576 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576 // indirect
+	google.golang.org/genproto v0.0.0-20250603155806-513f23925822 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250818200422-3122310a409c // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250811230008-5f3141c8851a // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect

go.sum

@@ -6,7 +6,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cloudevents/sdk-go/v2/protocol/gochan"
 	jsonpatch "github.com/evanphx/json-patch/v5"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -64,11 +63,14 @@ func TestPatch(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
 			watcherStore := store.NewAgentInformerWatcherStore[*addonapiv1alpha1.ManagedClusterAddOn]()
 
-			ceClientOpt := fake.NewAgentOptions(gochan.New(), nil, c.clusterName, c.clusterName+"agent")
+			ceClientOpt := fake.NewAgentOptions(fake.NewEventChan(), c.clusterName, c.clusterName+"agent")
 			ceClient, err := clients.NewCloudEventAgentClient(
-				context.Background(),
+				ctx,
 				ceClientOpt,
 				store.NewAgentWatcherStoreLister(watcherStore),
 				statushash.StatusHash,
@@ -89,7 +91,7 @@ func TestPatch(t *testing.T) {
 			watcherStore.SetInformer(informer)
 
 			if _, err = addonClientSet.AddonV1alpha1().ManagedClusterAddOns(c.clusterName).Patch(
-				context.Background(),
+				ctx,
 				c.addon.Name,
 				types.MergePatchType,
 				c.patch,

pkg/cloudevents/clients/addon/client_test.go

@@ -6,7 +6,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cloudevents/sdk-go/v2/protocol/gochan"
 	jsonpatch "github.com/evanphx/json-patch/v5"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -57,7 +56,7 @@ func TestCreate(t *testing.T) {
 			defer cancel()
 
 			watcherStore := store.NewAgentInformerWatcherStore[*clusterv1.ManagedCluster]()
-			ceClientOpt := fake.NewAgentOptions(gochan.New(), nil, "cluster1", "cluster1-agent")
+			ceClientOpt := fake.NewAgentOptions(fake.NewEventChan(), "cluster1", "cluster1-agent")
 			ceClient, err := clients.NewCloudEventAgentClient(
 				ctx,
 				ceClientOpt,
@@ -116,7 +115,7 @@ func TestPatch(t *testing.T) {
 			defer cancel()
 
 			watcherStore := store.NewAgentInformerWatcherStore[*clusterv1.ManagedCluster]()
-			ceClientOpt := fake.NewAgentOptions(gochan.New(), nil, "cluster1", "test-agent")
+			ceClientOpt := fake.NewAgentOptions(fake.NewEventChan(), "cluster1", "test-agent")
 			ceClient, err := clients.NewCloudEventAgentClient(
 				ctx,
 				ceClientOpt,

pkg/cloudevents/clients/cluster/client_test.go

@@ -7,8 +7,6 @@ import (
 
 	v1 "open-cluster-management.io/api/cluster/v1"
 
-	"github.com/cloudevents/sdk-go/v2/protocol/gochan"
-
 	certificatev1 "k8s.io/api/certificates/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -61,7 +59,7 @@ func TestCreate(t *testing.T) {
 			defer cancel()
 
 			watcherStore := store.NewAgentInformerWatcherStore[*certificatev1.CertificateSigningRequest]()
-			ceClientOpt := fake.NewAgentOptions(gochan.New(), nil, "cluster1", "cluster1-agent")
+			ceClientOpt := fake.NewAgentOptions(fake.NewEventChan(), "cluster1", "cluster1-agent")
 			ceClient, err := clients.NewCloudEventAgentClient(
 				ctx,
 				ceClientOpt,

pkg/cloudevents/clients/csr/client_test.go

@@ -5,8 +5,6 @@ import (
 	"encoding/json"
 	"testing"
 
-	"github.com/cloudevents/sdk-go/v2/protocol/gochan"
-
 	eventsv1 "k8s.io/api/events/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -34,9 +32,12 @@ func TestCreate(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
 			ceClient, err := clients.NewCloudEventAgentClient(
-				context.Background(),
-				fake.NewAgentOptions(gochan.New(), nil, c.clusterName, c.clusterName+"agent"),
+				ctx,
+				fake.NewAgentOptions(fake.NewEventChan(), c.clusterName, c.clusterName+"agent"),
 				nil,
 				statushash.StatusHash,
 				NewEventCodec())
@@ -89,9 +90,12 @@ func TestPatch(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
 			ceClient, err := clients.NewCloudEventAgentClient(
-				context.Background(),
-				fake.NewAgentOptions(gochan.New(), nil, c.clusterName, c.clusterName+"agent"),
+				ctx,
+				fake.NewAgentOptions(fake.NewEventChan(), c.clusterName, c.clusterName+"agent"),
 				nil,
 				statushash.StatusHash,
 				NewEventCodec())
@@ -101,7 +105,7 @@ func TestPatch(t *testing.T) {
 
 			evtClient := NewEventClient(ceClient).WithNamespace(c.clusterName)
 
-			if _, err := evtClient.Patch(context.Background(),
+			if _, err := evtClient.Patch(ctx,
 				c.event.Name,
 				types.StrategicMergePatchType,
 				c.patch,

pkg/cloudevents/clients/event/client_test.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/cloudevents/sdk-go/v2/protocol/gochan"
-
 	coordv1 "k8s.io/api/coordination/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -35,10 +33,13 @@ func TestUpdate(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
 			leaseWatchStore := store.NewSimpleStore[*coordv1.Lease]()
 			ceClient, err := clients.NewCloudEventAgentClient(
-				context.Background(),
-				fake.NewAgentOptions(gochan.New(), nil, c.clusterName, c.clusterName+"agent"),
+				ctx,
+				fake.NewAgentOptions(fake.NewEventChan(), c.clusterName, c.clusterName+"agent"),
 				store.NewAgentWatcherStoreLister(leaseWatchStore),
 				statushash.StatusHash,
 				NewLeaseCodec())
@@ -81,6 +82,9 @@ func TestGet(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
 			leaseWatchStore := store.NewSimpleStore[*coordv1.Lease]()
 			for _, lease := range c.leases {
 				if err := leaseWatchStore.Add(lease); err != nil {
@@ -89,8 +93,8 @@ func TestGet(t *testing.T) {
 			}
 
 			ceClient, err := clients.NewCloudEventAgentClient(
-				context.Background(),
-				fake.NewAgentOptions(gochan.New(), nil, "cluster1", "cluster1-agent"),
+				ctx,
+				fake.NewAgentOptions(fake.NewEventChan(), "cluster1", "cluster1-agent"),
 				store.NewAgentWatcherStoreLister(leaseWatchStore),
 				statushash.StatusHash,
 				NewLeaseCodec())

pkg/cloudevents/clients/lease/client_test.go

@@ -111,7 +111,7 @@ func (o *GenericClientOptions[T]) AgentClient(ctx context.Context) (generic.Clou
 		o.watcherStore = store.NewAgentInformerWatcherStore[T]()
 	}
 
-	options, err := builder.BuildCloudEventsAgentOptions(o.config, o.clusterName, o.clientID)
+	options, err := builder.BuildCloudEventsAgentOptions(o.config, o.clusterName, o.clientID, o.codec.EventDataType())
 	if err != nil {
 		return nil, err
 	}
@@ -182,7 +182,7 @@ func (o *GenericClientOptions[T]) SourceClient(ctx context.Context) (generic.Clo
 		return nil, fmt.Errorf("a watcher store is required")
 	}
 
-	options, err := builder.BuildCloudEventsSourceOptions(o.config, o.clientID, o.sourceID)
+	options, err := builder.BuildCloudEventsSourceOptions(o.config, o.clientID, o.sourceID, o.codec.EventDataType())
 	if err != nil {
 		return nil, err
 	}

pkg/cloudevents/clients/options/generic.go

@@ -1,6 +1,7 @@
 package constants
 
 const (
-	ConfigTypeMQTT = "mqtt"
-	ConfigTypeGRPC = "grpc"
+	ConfigTypeMQTT   = "mqtt"
+	ConfigTypeGRPC   = "grpc"
+	ConfigTypePubSub = "pubsub"
 )

pkg/cloudevents/constants/constants.go

@@ -48,10 +48,9 @@ func NewCloudEventAgentClient[T generic.ResourceObject](
 ) (generic.CloudEventsClient[T], error) {
 	baseClient := &baseClient{
 		clientID:               agentOptions.AgentID,
-		cloudEventsOptions:     agentOptions.CloudEventsOptions,
+		transport:              agentOptions.CloudEventsTransport,
 		cloudEventsRateLimiter: utils.NewRateLimiter(agentOptions.EventRateLimit),
 		reconnectedChan:        make(chan struct{}),
-		dataType:               codec.EventDataType(),
 	}
 
 	if err := baseClient.connect(ctx); err != nil {