address comments.

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

Author: morvencao

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

@@ -106,6 +106,11 @@ func TestBuildCloudEventsSourceOptions(t *testing.T) {
 					AgentEvents:    "projects/test-project/subscriptions/agentevents-source1",
 					AgentBroadcast: "projects/test-project/subscriptions/agentbroadcast-source1",
 				},
+				KeepaliveSettings: &pubsub.KeepaliveSettings{
+					Time:                5 * time.Minute,
+					Timeout:             20 * time.Second,
+					PermitWithoutStream: false,
+				},
 			},
 		},
 	}
@@ -167,6 +172,11 @@ func TestBuildCloudEventsAgentOptions(t *testing.T) {
 					SourceEvents:    "projects/test-project/subscriptions/sourceevents-cluster1",
 					SourceBroadcast: "projects/test-project/subscriptions/sourcebroadcast-cluster1",
 				},
+				KeepaliveSettings: &pubsub.KeepaliveSettings{
+					Time:                5 * time.Minute,
+					Timeout:             20 * time.Second,
+					PermitWithoutStream: false,
+				},
 			},
 		},
 	}

pkg/cloudevents/generic/options/pubsub/agentoptions.go

@@ -9,6 +9,7 @@ import (
 	"google.golang.org/api/option"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/keepalive"
 	"k8s.io/klog/v2"
 
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
@@ -30,8 +31,7 @@ type pubsubAgentTransport struct {
 	subscriber *pubsub.Subscriber
 	// Subscriber for resync broadcasts
 	resyncSubscriber *pubsub.Subscriber
-	// TODO: handle error channel
-	errorChan chan error
+	errorChan        chan error
 }
 
 // NewAgentOptions creates a new CloudEventsAgentOptions for Pub/Sub.
@@ -49,23 +49,28 @@ func NewAgentOptions(pubsubOptions *PubSubOptions,
 }
 
 func (o *pubsubAgentTransport) Connect(ctx context.Context) error {
-	options := []option.ClientOption{}
+	clientOptions := []option.ClientOption{}
 	if o.CredentialsFile != "" {
-		options = append(options, option.WithCredentialsFile(o.CredentialsFile))
+		clientOptions = append(clientOptions, option.WithCredentialsFile(o.CredentialsFile))
 	}
 	if o.Endpoint != "" {
-		options = append(options, option.WithEndpoint(o.Endpoint))
+		clientOptions = append(clientOptions, option.WithEndpoint(o.Endpoint))
 		if o.CredentialsFile == "" {
 			// use the insecure connection for local development/testing when no credentials are provided
 			pubsubConn, err := grpc.NewClient(o.Endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
 			if err != nil {
 				return err
 			}
-			options = append(options, option.WithGRPCConn(pubsubConn))
+			clientOptions = append(clientOptions, option.WithGRPCConn(pubsubConn))
 		}
 	}
 
-	client, err := pubsub.NewClient(ctx, o.ProjectID, options...)
+	if o.KeepaliveSettings != nil {
+		// config keepalive parameters for pubsub client
+		clientOptions = append(clientOptions, option.WithGRPCDialOption(grpc.WithKeepaliveParams(toGRPCKeepaliveParamater(o.KeepaliveSettings))))
+	}
+
+	client, err := pubsub.NewClient(ctx, o.ProjectID, clientOptions...)
 	if err != nil {
 		return err
 	}
@@ -77,6 +82,13 @@ func (o *pubsubAgentTransport) Connect(ctx context.Context) error {
 	o.subscriber = client.Subscriber(o.Subscriptions.SourceEvents)
 	o.resyncSubscriber = client.Subscriber(o.Subscriptions.SourceBroadcast)
 
+	// configure receive settings if provided
+	if o.ReceiveSettings != nil {
+		receiveSettings := toPubSubReceiveSettings(o.ReceiveSettings)
+		o.subscriber.ReceiveSettings = receiveSettings
+		o.resyncSubscriber.ReceiveSettings = receiveSettings
+	}
+
 	return nil
 }
 
@@ -88,7 +100,7 @@ func (o *pubsubAgentTransport) Send(ctx context.Context, evt cloudevents.Event)
 
 	eventType, err := types.ParseCloudEventsType(evt.Context.GetType())
 	if err != nil {
-		return fmt.Errorf("unsupported event type %s, %v", eventType, err)
+		return fmt.Errorf("unsupported event type %s, %v", evt.Context.GetType(), err)
 	}
 
 	// determine publisher based on event type
@@ -105,10 +117,6 @@ func (o *pubsubAgentTransport) Send(ctx context.Context, evt cloudevents.Event)
 }
 
 func (o *pubsubAgentTransport) Receive(ctx context.Context, fn options.ReceiveHandlerFn) error {
-	// create error channels for both subscribers
-	subscriberErrChan := make(chan error, 1)
-	resyncSubscriberErrChan := make(chan error, 1)
-
 	// start the subscriber for spec updates
 	go func() {
 		err := o.subscriber.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
@@ -123,7 +131,15 @@ func (o *pubsubAgentTransport) Receive(ctx context.Context, fn options.ReceiveHa
 			msg.Ack()
 		})
 		if err != nil {
-			subscriberErrChan <- fmt.Errorf("subscriber is interrupted by error: %w", err)
+			// When keepalive is enabled, the Pub/Sub client's Receive call automatically retries on retryable errors.
+			// See: https://github.com/googleapis/google-cloud-go/blob/b8e70aa0056a3e126bc36cb7bf242d987f32c0bd/pubsub/service.go#L51
+			// If Receive returns an error, it’s usually due to a non-retryable issue or service outage.
+			// In such cases, we send the error to the error channel to signal the source/agent client to reconnect later.
+			select {
+			case o.errorChan <- fmt.Errorf("subscriber is interrupted by error: %w", err):
+			default:
+				klog.Warningf("error channel full, dropping subscriber error: %v", err)
+			}
 		}
 	}()
 
@@ -141,19 +157,21 @@ func (o *pubsubAgentTransport) Receive(ctx context.Context, fn options.ReceiveHa
 			msg.Ack()
 		})
 		if err != nil {
-			resyncSubscriberErrChan <- fmt.Errorf("resync subscriber is interrupted by error: %w", err)
+			// When keepalive is enabled, the Pub/Sub client's Receive call automatically retries on retryable errors.
+			// See: https://github.com/googleapis/google-cloud-go/blob/b8e70aa0056a3e126bc36cb7bf242d987f32c0bd/pubsub/service.go#L51
+			// If Receive returns an error, it’s usually due to a non-retryable issue or service outage.
+			// In such cases, we send the error to the error channel to signal the source/agent client to reconnect later.
+			select {
+			case o.errorChan <- fmt.Errorf("resync subscriber is interrupted by error: %w", err):
+			default:
+				klog.Warningf("error channel full, dropping resync subscriber error: %v", err)
+			}
 		}
 	}()
 
-	// wait for either subscriber to error or context cancellation
-	select {
-	case err := <-subscriberErrChan:
-		return err
-	case err := <-resyncSubscriberErrChan:
-		return err
-	case <-ctx.Done():
-		return ctx.Err()
-	}
+	// wait for context cancellation or timeout
+	<-ctx.Done()
+	return ctx.Err()
 }
 
 func (o *pubsubAgentTransport) Close(ctx context.Context) error {
@@ -163,3 +181,44 @@ func (o *pubsubAgentTransport) Close(ctx context.Context) error {
 func (o *pubsubAgentTransport) ErrorChan() <-chan error {
 	return o.errorChan
 }
+
+// toGRPCKeepaliveParamater converts our KeepaliveSettings to GRPC ClientParameters.
+func toGRPCKeepaliveParamater(settings *KeepaliveSettings) keepalive.ClientParameters {
+	keepaliveParamater := keepalive.ClientParameters{
+		PermitWithoutStream: settings.PermitWithoutStream,
+	}
+	if settings.Time > 0 {
+		keepaliveParamater.Time = settings.Time
+	}
+	if settings.Timeout > 0 {
+		keepaliveParamater.Timeout = settings.Timeout
+	}
+
+	return keepaliveParamater
+}
+
+// toPubSubReceiveSettings converts our ReceiveSettings to Pub/Sub ReceiveSettings.
+func toPubSubReceiveSettings(settings *ReceiveSettings) pubsub.ReceiveSettings {
+	receiveSettings := pubsub.ReceiveSettings{}
+
+	if settings.MaxExtension > 0 {
+		receiveSettings.MaxExtension = settings.MaxExtension
+	}
+	if settings.MaxDurationPerAckExtension > 0 {
+		receiveSettings.MaxDurationPerAckExtension = settings.MaxDurationPerAckExtension
+	}
+	if settings.MinDurationPerAckExtension > 0 {
+		receiveSettings.MinDurationPerAckExtension = settings.MinDurationPerAckExtension
+	}
+	if settings.MaxOutstandingMessages > 0 {
+		receiveSettings.MaxOutstandingMessages = settings.MaxOutstandingMessages
+	}
+	if settings.MaxOutstandingBytes > 0 {
+		receiveSettings.MaxOutstandingBytes = settings.MaxOutstandingBytes
+	}
+	if settings.NumGoroutines > 0 {
+		receiveSettings.NumGoroutines = settings.NumGoroutines
+	}
+
+	return receiveSettings
+}

pkg/cloudevents/generic/options/pubsub/agentoptions_test.go

@@ -2,7 +2,10 @@ package pubsub
 
 import (
 	"testing"
+	"time"
 
+	"cloud.google.com/go/pubsub/v2"
+	"google.golang.org/grpc/keepalive"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
@@ -151,3 +154,170 @@ func TestPubsubAgentTransport_Structure(t *testing.T) {
 		t.Error("expected resyncSubscriber to be nil before Connect is called")
 	}
 }
+
+func TestToGRPCKeepaliveParameter(t *testing.T) {
+	cases := []struct {
+		name     string
+		settings *KeepaliveSettings
+		expected keepalive.ClientParameters
+	}{
+		{
+			name: "all settings provided",
+			settings: &KeepaliveSettings{
+				Time:                10 * time.Minute,
+				Timeout:             30 * time.Second,
+				PermitWithoutStream: true,
+			},
+			expected: keepalive.ClientParameters{
+				Time:                10 * time.Minute,
+				Timeout:             30 * time.Second,
+				PermitWithoutStream: true,
+			},
+		},
+		{
+			name: "only time provided",
+			settings: &KeepaliveSettings{
+				Time: 5 * time.Minute,
+			},
+			expected: keepalive.ClientParameters{
+				Time:                5 * time.Minute,
+				Timeout:             0,
+				PermitWithoutStream: false,
+			},
+		},
+		{
+			name: "only timeout provided",
+			settings: &KeepaliveSettings{
+				Timeout: 20 * time.Second,
+			},
+			expected: keepalive.ClientParameters{
+				Time:                0,
+				Timeout:             20 * time.Second,
+				PermitWithoutStream: false,
+			},
+		},
+		{
+			name: "zero values",
+			settings: &KeepaliveSettings{
+				Time:                0,
+				Timeout:             0,
+				PermitWithoutStream: false,
+			},
+			expected: keepalive.ClientParameters{
+				Time:                0,
+				Timeout:             0,
+				PermitWithoutStream: false,
+			},
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			result := toGRPCKeepaliveParamater(c.settings)
+
+			if result.Time != c.expected.Time {
+				t.Errorf("expected Time to be %v, got %v", c.expected.Time, result.Time)
+			}
+
+			if result.Timeout != c.expected.Timeout {
+				t.Errorf("expected Timeout to be %v, got %v", c.expected.Timeout, result.Timeout)
+			}
+
+			if result.PermitWithoutStream != c.expected.PermitWithoutStream {
+				t.Errorf("expected PermitWithoutStream to be %v, got %v", c.expected.PermitWithoutStream, result.PermitWithoutStream)
+			}
+		})
+	}
+}
+
+func TestToPubSubReceiveSettings(t *testing.T) {
+	cases := []struct {
+		name     string
+		settings *ReceiveSettings
+		expected pubsub.ReceiveSettings
+	}{
+		{
+			name: "all settings provided",
+			settings: &ReceiveSettings{
+				MaxExtension:               600 * time.Second,
+				MaxDurationPerAckExtension: 10 * time.Second,
+				MinDurationPerAckExtension: 1 * time.Second,
+				MaxOutstandingMessages:     1000,
+				MaxOutstandingBytes:        1000000000,
+				NumGoroutines:              10,
+			},
+			expected: pubsub.ReceiveSettings{
+				MaxExtension:               600 * time.Second,
+				MaxDurationPerAckExtension: 10 * time.Second,
+				MinDurationPerAckExtension: 1 * time.Second,
+				MaxOutstandingMessages:     1000,
+				MaxOutstandingBytes:        1000000000,
+				NumGoroutines:              10,
+			},
+		},
+		{
+			name: "partial settings",
+			settings: &ReceiveSettings{
+				MaxExtension:           300 * time.Second,
+				MaxOutstandingMessages: 500,
+			},
+			expected: pubsub.ReceiveSettings{
+				MaxExtension:               300 * time.Second,
+				MaxDurationPerAckExtension: 0,
+				MinDurationPerAckExtension: 0,
+				MaxOutstandingMessages:     500,
+				MaxOutstandingBytes:        0,
+				NumGoroutines:              0,
+			},
+		},
+		{
+			name: "zero values",
+			settings: &ReceiveSettings{
+				MaxExtension:               0,
+				MaxDurationPerAckExtension: 0,
+				MinDurationPerAckExtension: 0,
+				MaxOutstandingMessages:     0,
+				MaxOutstandingBytes:        0,
+				NumGoroutines:              0,
+			},
+			expected: pubsub.ReceiveSettings{
+				MaxExtension:               0,
+				MaxDurationPerAckExtension: 0,
+				MinDurationPerAckExtension: 0,
+				MaxOutstandingMessages:     0,
+				MaxOutstandingBytes:        0,
+				NumGoroutines:              0,
+			},
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			result := toPubSubReceiveSettings(c.settings)
+
+			if result.MaxExtension != c.expected.MaxExtension {
+				t.Errorf("expected MaxExtension to be %v, got %v", c.expected.MaxExtension, result.MaxExtension)
+			}
+
+			if result.MaxDurationPerAckExtension != c.expected.MaxDurationPerAckExtension {
+				t.Errorf("expected MaxDurationPerAckExtension to be %v, got %v", c.expected.MaxDurationPerAckExtension, result.MaxDurationPerAckExtension)
+			}
+
+			if result.MinDurationPerAckExtension != c.expected.MinDurationPerAckExtension {
+				t.Errorf("expected MinDurationPerAckExtension to be %v, got %v", c.expected.MinDurationPerAckExtension, result.MinDurationPerAckExtension)
+			}
+
+			if result.MaxOutstandingMessages != c.expected.MaxOutstandingMessages {
+				t.Errorf("expected MaxOutstandingMessages to be %v, got %v", c.expected.MaxOutstandingMessages, result.MaxOutstandingMessages)
+			}
+
+			if result.MaxOutstandingBytes != c.expected.MaxOutstandingBytes {
+				t.Errorf("expected MaxOutstandingBytes to be %v, got %v", c.expected.MaxOutstandingBytes, result.MaxOutstandingBytes)
+			}
+
+			if result.NumGoroutines != c.expected.NumGoroutines {
+				t.Errorf("expected NumGoroutines to be %v, got %v", c.expected.NumGoroutines, result.NumGoroutines)
+			}
+		})
+	}
+}