add pubsub protocol support for cloudevents transport.

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

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.3.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.32.0
+	google.golang.org/api v0.255.0
+	google.golang.org/grpc v1.76.0
+	google.golang.org/protobuf v1.36.10
 	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.17.0 // indirect
+	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
+	cloud.google.com/go/compute/metadata v0.9.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.37.0 // indirect
+	go.opentelemetry.io/otel/metric v1.37.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.37.0 // indirect
+	go.opentelemetry.io/otel/trace v1.37.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/crypto v0.43.0 // indirect
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
-	golang.org/x/net v0.43.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
+	golang.org/x/net v0.46.0 // indirect
+	golang.org/x/sync v0.17.0 // indirect
+	golang.org/x/sys v0.37.0 // indirect
+	golang.org/x/term v0.36.0 // indirect
+	golang.org/x/text v0.30.0 // indirect
+	golang.org/x/time v0.14.0 // 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-20251029180050-ab9386a59fda // 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

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

pkg/cloudevents/constants/constants.go

@@ -7,6 +7,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/pubsub"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
@@ -21,6 +22,7 @@ type ConfigLoader struct {
 // Available configuration types:
 //   - mqtt
 //   - grpc
+//   - pubsub
 func NewConfigLoader(configType, configPath string) *ConfigLoader {
 	return &ConfigLoader{
 		configType: configType,
@@ -45,6 +47,13 @@ func (l *ConfigLoader) LoadConfig() (string, any, error) {
 		}
 
 		return grpcOptions.Dialer.URL, grpcOptions, nil
+	case constants.ConfigTypePubSub:
+		pubsubOptions, err := pubsub.BuildPubSubOptionsFromFlags(l.configPath)
+		if err != nil {
+			return "", nil, err
+		}
+
+		return "", pubsubOptions, nil
 	}
 
 	return "", nil, fmt.Errorf("unsupported config type %s", l.configType)
@@ -58,6 +67,8 @@ func BuildCloudEventsSourceOptions(config any,
 		return mqtt.NewSourceOptions(config, clientId, sourceId), nil
 	case *grpc.GRPCOptions:
 		return grpc.NewSourceOptions(config, sourceId, dataType), nil
+	case *pubsub.PubSubOptions:
+		return pubsub.NewSourceOptions(config, sourceId), nil
 	default:
 		return nil, fmt.Errorf("unsupported client configuration type %T", config)
 	}
@@ -71,6 +82,8 @@ func BuildCloudEventsAgentOptions(config any,
 		return mqtt.NewAgentOptions(config, clusterName, clientId), nil
 	case *grpc.GRPCOptions:
 		return grpc.NewAgentOptions(config, clusterName, clientId, dataType), nil
+	case *pubsub.PubSubOptions:
+		return pubsub.NewAgentOptions(config, clusterName, clientId), nil
 	default:
 		return nil, fmt.Errorf("unsupported client configuration type %T", config)
 	}

pkg/cloudevents/generic/options/builder/optionsbuilder.go

@@ -0,0 +1,112 @@
+package pubsub
+
+import (
+	"context"
+	"fmt"
+
+	"cloud.google.com/go/pubsub/v2"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"google.golang.org/api/option"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"k8s.io/klog/v2"
+
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
+)
+
+type pubsubAgentTransport struct {
+	PubSubOptions
+	clusterName string
+	client      *pubsub.Client
+	publisher   *pubsub.Publisher
+	subscriber  *pubsub.Subscriber
+	errorChan   chan error
+}
+
+// NewAgentOptions creates a new CloudEventsAgentOptions for Pub/Sub.
+func NewAgentOptions(pubsubOptions *PubSubOptions,
+	clusterName, agentID string) *options.CloudEventsAgentOptions {
+	return &options.CloudEventsAgentOptions{
+		CloudEventsTransport: &pubsubAgentTransport{
+			PubSubOptions: *pubsubOptions,
+			clusterName:   clusterName,
+			errorChan:     make(chan error),
+		},
+		AgentID:     agentID,
+		ClusterName: clusterName,
+	}
+}
+
+func (o *pubsubAgentTransport) Connect(ctx context.Context) error {
+	options := []option.ClientOption{}
+	if o.CredentialsFile != "" {
+		options = append(options, option.WithCredentialsFile(o.CredentialsFile))
+	}
+	if o.Endpoint != "" {
+		options = append(options, option.WithEndpoint(o.Endpoint))
+		if o.CredentialsFile == "" {
+			pubsubConn, err := grpc.NewClient(o.Endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
+			if err != nil {
+				return err
+			}
+			options = append(options, option.WithGRPCConn(pubsubConn))
+		}
+	}
+	client, err := pubsub.NewClient(ctx, o.ProjectID, options...)
+	if err != nil {
+		return err
+	}
+
+	o.client = client
+	o.publisher = client.Publisher(o.Topics.AgentEvents)
+	o.subscriber = client.Subscriber(o.Subscriptions.SourceEvents)
+
+	return nil
+}
+
+func (o *pubsubAgentTransport) Send(ctx context.Context, evt cloudevents.Event) error {
+	msg, err := Encode(evt)
+	if err != nil {
+		return err
+	}
+
+	// Publish the message
+	result := o.publisher.Publish(ctx, msg)
+
+	// Block until the result is returned and a server-generated
+	// ID is returned for the published message.
+	id, err := result.Get(ctx)
+	if err != nil {
+		return fmt.Errorf("pubsub: result.Get: %w", err)
+	}
+	klog.V(4).Infof("Published a message; msg ID: %v\n", id)
+	return nil
+}
+
+func (o *pubsubAgentTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
+	err := o.subscriber.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
+		evt, err := Decode(msg)
+		if err != nil {
+			klog.Warningf("failed to decode Pub/Sub message: %v", err)
+		} else {
+			// Handle the event
+			fn(evt)
+		}
+		// TODO: decide when to Ack
+		msg.Ack()
+	})
+
+	if err != nil {
+		return fmt.Errorf("pubsub: subscriber.Receive: %w", err)
+	}
+
+	return nil
+}
+
+func (o *pubsubAgentTransport) Close(ctx context.Context) error {
+	return o.client.Close()
+}
+
+func (o *pubsubAgentTransport) ErrorChan() <-chan error {
+	return o.errorChan
+}