grpc transport v2

Signed-off-by: Wei Liu <[email protected]>

Author: skeeey

pkg/cloudevents/constants/constants.go

@@ -4,3 +4,7 @@ const (
 	ConfigTypeMQTT = "mqtt"
 	ConfigTypeGRPC = "grpc"
 )
+
+// GRPCSubscriptionIDKey is the key for the gRPC subscription ID.
+// This ID is generated by the gRPC server after the client subscribes to it.
+const GRPCSubscriptionIDKey = "subscription-id"

pkg/cloudevents/generic/clients/baseclient.go

@@ -79,7 +79,7 @@ func (c *baseClient) connect(ctx context.Context) error {
 				// TODO enhance the cloudevents SKD to avoid wrapping the error type to distinguish the net connection
 				// errors
 				if err != nil {
-					// failed to reconnect, try agin
+					// failed to reconnect, try again
 					runtime.HandleErrorWithContext(ctx, err, "the cloudevents client reconnect failed")
 					<-wait.RealTimer(DelayFn()).C()
 					continue

pkg/cloudevents/generic/options/v2/grpc/options.go

@@ -10,20 +10,16 @@ import (
 func NewAgentOptions(grpcOptions *grpc.GRPCOptions,
 	clusterName, agentID string, dataType types.CloudEventsDataType) *options.CloudEventsAgentOptions {
 	return &options.CloudEventsAgentOptions{
-		CloudEventsTransport: &grpcTransport{
-			opts:      grpcOptions,
-			errorChan: make(chan error),
-			getSubscriptionRequest: func() *pbv1.SubscriptionRequest {
-				return &pbv1.SubscriptionRequest{
-					// TODO: Update this code to determine the subscription source for the agent client.
-					// Currently, the grpc agent client is not utilized, and the 'Source' field serves
-					// as a placeholder with all the sources.
-					Source:      types.SourceAll,
-					ClusterName: clusterName,
-					DataType:    dataType.String(),
-				}
-			},
-		},
+		CloudEventsTransport: newTransport(grpcOptions, func() *pbv1.SubscriptionRequest {
+			return &pbv1.SubscriptionRequest{
+				// TODO: Update this code to determine the subscription source for the agent client.
+				// Currently, the grpc agent client is not utilized, and the 'Source' field serves
+				// as a placeholder with all the sources.
+				Source:      types.SourceAll,
+				ClusterName: clusterName,
+				DataType:    dataType.String(),
+			}
+		}),
 		AgentID:     agentID,
 		ClusterName: clusterName,
 	}
@@ -32,16 +28,12 @@ func NewAgentOptions(grpcOptions *grpc.GRPCOptions,
 func NewSourceOptions(gRPCOptions *grpc.GRPCOptions,
 	sourceID string, dataType types.CloudEventsDataType) *options.CloudEventsSourceOptions {
 	return &options.CloudEventsSourceOptions{
-		CloudEventsTransport: &grpcTransport{
-			opts:      gRPCOptions,
-			errorChan: make(chan error),
-			getSubscriptionRequest: func() *pbv1.SubscriptionRequest {
-				return &pbv1.SubscriptionRequest{
-					Source:   sourceID,
-					DataType: dataType.String(),
-				}
-			},
-		},
+		CloudEventsTransport: newTransport(gRPCOptions, func() *pbv1.SubscriptionRequest {
+			return &pbv1.SubscriptionRequest{
+				Source:   sourceID,
+				DataType: dataType.String(),
+			}
+		}),
 		SourceID: sourceID,
 	}
 }

pkg/cloudevents/generic/options/v2/grpc/transport.go

@@ -9,10 +9,13 @@ import (
 	"github.com/cloudevents/sdk-go/v2/binding"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/status"
 
 	"k8s.io/klog/v2"
 
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/constants"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options"
 	grpcoptions "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
 	pbv1 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protobuf/v1"
@@ -21,17 +24,28 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/heartbeat"
 )
 
+type subscriptionRequestGetter func() *pbv1.SubscriptionRequest
+
 type grpcTransport struct {
 	opts *grpcoptions.GRPCOptions
 
 	client    pbv1.CloudEventServiceClient
 	subClient pbv1.CloudEventService_SubscribeClient
+	subID     string
 
 	mu        sync.Mutex
 	closeChan chan struct{}
 	errorChan chan error
 
-	getSubscriptionRequest func() *pbv1.SubscriptionRequest
+	getSubscriptionRequest subscriptionRequestGetter
+}
+
+func newTransport(grpcOptions *grpcoptions.GRPCOptions, getter subscriptionRequestGetter) *grpcTransport {
+	return &grpcTransport{
+		opts:                   grpcOptions,
+		errorChan:              make(chan error),
+		getSubscriptionRequest: getter,
+	}
 }
 
 func (t *grpcTransport) Connect(ctx context.Context) error {
@@ -72,25 +86,37 @@ func (t *grpcTransport) Send(ctx context.Context, evt cloudevents.Event) error {
 
 func (t *grpcTransport) Subscribe(ctx context.Context) error {
 	logger := klog.FromContext(ctx)
-	subscribeOption := t.getSubscriptionRequest()
-	subClient, err := t.client.Subscribe(ctx, subscribeOption)
+	subOption := t.getSubscriptionRequest()
+	subClient, err := t.client.Subscribe(ctx, subOption)
 	if err != nil {
 		return err
 	}
 
-	logger.Info("subscribed to grpc server", "subscribeOption", subscribeOption)
+	// Wait for server to send subscription-id
+	header, err := subClient.Header()
+	if err != nil {
+		return err
+	}
+
+	values := header.Get(constants.GRPCSubscriptionIDKey)
+	if len(values) != 1 {
+		return fmt.Errorf("unexpected subscription-id")
+	}
+	t.subID = values[0]
 	t.subClient = subClient
+
+	logger.Info("subscribed to grpc server", "subID", t.subID, "subOption", subOption)
 	return nil
 }
 
 func (t *grpcTransport) Receive(ctx context.Context, handleFn options.ReceiveHandlerFn) error {
 	logger := klog.FromContext(ctx)
-	subscribeOption := t.getSubscriptionRequest()
+	subOption := t.getSubscriptionRequest()
 	subCtx, cancel := context.WithCancel(ctx)
 	healthChecker := heartbeat.NewHealthChecker(t.opts.ServerHealthinessTimeout, t.errorChan)
 
 	// start to receive the events from stream
-	logger.Info("receiving events", "subscribeOption", subscribeOption)
+	logger.Info("receiving events", "subID", t.subID, "subOption", subOption)
 	go t.startEventsReceiver(subCtx, healthChecker.Input(), handleFn)
 
 	// start to watch the stream heartbeat
@@ -105,7 +131,7 @@ func (t *grpcTransport) Receive(ctx context.Context, handleFn options.ReceiveHan
 	// ensure the event receiver and heartbeat watcher are done
 	cancel()
 
-	logger.Info("stop receiving events", "subscribeOption", subscribeOption)
+	logger.Info("stop receiving events", "subID", t.subID, "subOption", subOption)
 	return nil
 }
 
@@ -137,10 +163,16 @@ func (t *grpcTransport) startEventsReceiver(ctx context.Context,
 	for {
 		pbEvt, err := t.subClient.Recv()
 		if err != nil {
+			st, ok := status.FromError(err)
+			if ok && st.Code() == codes.Canceled {
+				// canceled by transport, return directly.
+				return
+			}
+
 			select {
-			case t.errorChan <- fmt.Errorf("subscribe stream failed: %w", err):
+			case t.errorChan <- fmt.Errorf("subscribe (subID=%s) stream failed: %w", t.subID, err):
 			default:
-				logger.Error(err, "subscribe stream failed")
+				logger.Error(err, "subscribe stream failed", "subID", t.subID)
 			}
 			return
 		}
@@ -183,7 +215,7 @@ func (t *grpcTransport) monitorConnectionState(ctx context.Context, conn *grpc.C
 			// the ctx is closed, stop this watch
 			logger.Info("stop watching grpc connection state")
 			if err := conn.Close(); err != nil {
-				logger.Error(err, "failed to close gRPC connection")
+				logger.Error(err, "failed to close grpc connection")
 			}
 			return
 		}
@@ -208,7 +240,7 @@ func (t *grpcTransport) monitorConnectionState(ctx context.Context, conn *grpc.C
 			if newState != connectivity.Shutdown {
 				// don't close the connection if it's already shutdown
 				if err := conn.Close(); err != nil {
-					logger.Error(err, "failed to close gRPC connection")
+					logger.Error(err, "failed to close grpc connection")
 				}
 			}
 

pkg/cloudevents/server/grpc/broker.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/constants"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/heartbeat"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/metrics"
 
@@ -17,6 +18,7 @@ import (
 	cloudeventstypes "github.com/cloudevents/sdk-go/v2/types"
 	"github.com/google/uuid"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -28,7 +30,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server"
 )
 
-type resourceHandler func(ctx context.Context, res *cloudevents.Event) error
+type resourceHandler func(ctx context.Context, subID string, res *cloudevents.Event) error
 
 // subscriber defines a subscriber that can receive and handle resource spec.
 type subscriber struct {
@@ -215,7 +217,7 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		}
 	}()
 
-	subscriberID := bkr.register(ctx, subReq.ClusterName, *dataType, func(ctx context.Context, evt *cloudevents.Event) error {
+	subID := bkr.register(ctx, subReq.ClusterName, *dataType, func(ctx context.Context, subID string, evt *cloudevents.Event) error {
 		// convert the cloudevents.Event to pbv1.CloudEvent
 		// WARNING: don't use "pbEvt, err := pb.ToProto(evt)" to convert cloudevent to protobuf
 		pbEvt := &pbv1.CloudEvent{}
@@ -224,7 +226,7 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		}
 
 		// send the cloudevent to the subscriber
-		logger.V(4).Info("sending the event to spec subscribers", "eventContext", evt.Context)
+		logger.V(4).Info("sending the event to spec subscribers", "subID", subID, "eventContext", evt.Context)
 		select {
 		case eventCh <- pbEvt:
 		case <-ctx.Done():
@@ -239,12 +241,19 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		return nil
 	})
 
+	// Send initial metadata to signal registration is complete
+	if err := subServer.SendHeader(metadata.Pairs(constants.GRPCSubscriptionIDKey, subID)); err != nil {
+		logger.Error(err, "failed to send subscription header, , unregister subscriber", "subID", subID)
+		bkr.unregister(ctx, subID)
+		return err
+	}
+
 	go heartbeater.Start(ctx)
 
 	select {
 	case err := <-sendErrCh:
-		logger.Error(err, "failed to send event, unregister subscriber", "id", subscriberID)
-		bkr.unregister(ctx, subscriberID)
+		logger.Error(err, "failed to send event, unregister subscriber", "subID", subID)
+		bkr.unregister(ctx, subID)
 		return err
 	case <-ctx.Done():
 		// The context of the stream has been canceled or completed.
@@ -253,7 +262,7 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		// - The server closed the stream, potentially due to a shutdown.
 		// Regardless of the reason, unregister the subscriber and stop processing.
 		// No error is returned here because the stream closure is expected.
-		bkr.unregister(ctx, subscriberID)
+		bkr.unregister(ctx, subID)
 		return nil
 	}
 }
@@ -291,13 +300,14 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 	}
 
 	if len(objs) == 0 {
-		log.V(4).Info("there are is no objs from the list, do nothing")
+		log.V(4).Info("no objs from the lister, do nothing", "eventContext", evt.Context)
 		return nil
 	}
 
 	for _, obj := range objs {
 		// respond with the deleting resource regardless of the resource version
 		if _, ok := obj.Extensions()[types.ExtensionDeletionTimestamp]; ok {
+			log.V(4).Info("respond spec resync request", "eventContext", evt.Context)
 			err = bkr.handleRes(ctx, obj, eventDataType, "delete_request")
 			if err != nil {
 				log.Error(err, "failed to handle resync spec request")
@@ -315,6 +325,7 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 		// the version of the work is not maintained on source or the source's work is newer than agent, send
 		// the newer work to agent
 		if currentResourceVersion == 0 || int64(currentResourceVersion) > lastResourceVersion {
+			log.V(4).Info("respond spec resync request", "eventContext", evt.Context)
 			err := bkr.handleRes(ctx, obj, eventDataType, "update_request")
 			if err != nil {
 				log.Error(err, "failed to handle resync spec request")
@@ -341,6 +352,7 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 			NewEvent()
 
 		// send a delete event for the current resource
+		log.V(4).Info("respond spec resync request", "eventContext", evt.Context)
 		err := bkr.handleRes(ctx, &obj, eventDataType, "delete_request")
 		if err != nil {
 			log.Error(err, "failed to handle delete request")
@@ -374,20 +386,16 @@ func (bkr *GRPCBroker) handleRes(
 	}
 	clusterName := fmt.Sprintf("%s", clusterNameValue)
 
-	// checks if the event should be processed by the current instance
-	// by verifying the resource consumer name is in the subscriber list, ensuring the
-	// event will be only processed when the consumer is subscribed to the current broker.
-	if !bkr.IsConsumerSubscribed(clusterName) {
-		log.V(4).Info("skip the event since the agent is not subscribed.",
-			"eventContext", evt.Context, "clusterName", clusterName)
-		return nil
-	}
-
-	for _, subscriber := range bkr.subscribers {
+	for id, subscriber := range bkr.subscribers {
+		// checks if the event should be processed by the current instance
+		// by verifying the resource consumer name is in the subscriber list, ensuring the
+		// event will be only processed when the consumer is subscribed to the current broker.
 		if subscriber.clusterName == clusterName && subscriber.dataType == t {
-			if err := subscriber.handler(ctx, evt); err != nil {
+			if err := subscriber.handler(ctx, id, evt); err != nil {
 				return err
 			}
+		} else {
+			log.V(4).Info("skip the event since the agent is not subscribed.", "subID", id, "eventContext", evt.Context)
 		}
 	}
 	return nil