grpc transport v2

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

Author: skeeey

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

@@ -12,7 +12,6 @@ func NewAgentOptions(grpcOptions *grpc.GRPCOptions,
 	return &options.CloudEventsAgentOptions{
 		CloudEventsTransport: &grpcTransport{
 			opts:      grpcOptions,
-			closeChan: make(chan struct{}),
 			errorChan: make(chan error),
 			getSubscriptionRequest: func() *pbv1.SubscriptionRequest {
 				return &pbv1.SubscriptionRequest{
@@ -35,7 +34,6 @@ func NewSourceOptions(gRPCOptions *grpc.GRPCOptions,
 	return &options.CloudEventsSourceOptions{
 		CloudEventsTransport: &grpcTransport{
 			opts:      gRPCOptions,
-			closeChan: make(chan struct{}),
 			errorChan: make(chan error),
 			getSubscriptionRequest: func() *pbv1.SubscriptionRequest {
 				return &pbv1.SubscriptionRequest{

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

@@ -3,6 +3,7 @@ package grpc
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -26,13 +27,17 @@ type grpcTransport struct {
 	client    pbv1.CloudEventServiceClient
 	subClient pbv1.CloudEventService_SubscribeClient
 
+	mu        sync.Mutex
 	closeChan chan struct{}
 	errorChan chan error
 
 	getSubscriptionRequest func() *pbv1.SubscriptionRequest
 }
 
 func (t *grpcTransport) Connect(ctx context.Context) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
 	logger := klog.FromContext(ctx)
 	conn, err := t.opts.Dialer.Dial()
 	if err != nil {
@@ -41,6 +46,9 @@ func (t *grpcTransport) Connect(ctx context.Context) error {
 
 	t.client = pbv1.NewCloudEventServiceClient(conn)
 
+	// Initialize closeChan to support reconnect cycles
+	t.closeChan = make(chan struct{})
+
 	// Start a goroutine to monitor the gRPC connection state changes
 	go t.monitorConnectionState(ctx, conn)
 
@@ -106,8 +114,19 @@ func (t *grpcTransport) ErrorChan() <-chan error {
 }
 
 func (t *grpcTransport) Close(ctx context.Context) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
 	klog.FromContext(ctx).Info("close grpc transport")
-	close(t.closeChan)
+
+	// Guard against double-close panic
+	select {
+	case <-t.closeChan:
+		// Already closed
+	default:
+		close(t.closeChan)
+	}
+
 	return t.opts.Dialer.Close()
 }
 
@@ -163,6 +182,9 @@ func (t *grpcTransport) monitorConnectionState(ctx context.Context, conn *grpc.C
 		if !conn.WaitForStateChange(ctx, state) {
 			// 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")
+			}
 			return
 		}
 

pkg/cloudevents/server/grpc/broker.go

@@ -90,7 +90,7 @@ func (bkr *GRPCBroker) Publish(ctx context.Context, pubReq *pbv1.PublishRequest)
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("failed to parse cloud event type %s, %v", evt.Type(), err))
 	}
 
-	logger.V(4).Info("receive the event with grpc broker", "event", evt.Context)
+	logger.V(4).Info("receive the event with grpc broker", "eventContext", evt.Context)
 
 	// handler resync request
 	if eventType.Action == types.ResyncRequestAction {
@@ -134,7 +134,7 @@ func (bkr *GRPCBroker) register(
 		handler:     handler,
 	}
 
-	logger.V(4).Info("register a subscriber", "id", id)
+	logger.V(4).Info("register a subscriber", "id", id, "clusterName", clusterName, "dataType", dataType)
 	metrics.IncGRPCCESubscribersMetric(clusterName, dataType.String())
 
 	return id
@@ -146,8 +146,8 @@ func (bkr *GRPCBroker) unregister(ctx context.Context, id string) {
 	defer bkr.mu.Unlock()
 
 	logger := klog.FromContext(ctx)
-	logger.V(4).Info("unregister subscriber", "id", id)
 	if sub, exists := bkr.subscribers[id]; exists {
+		logger.V(4).Info("unregister subscriber", "id", id, "clusterName", sub.clusterName, "dataType", sub.dataType)
 		delete(bkr.subscribers, id)
 		metrics.DecGRPCCESubscribersMetric(sub.clusterName, sub.dataType.String())
 	}
@@ -308,7 +308,7 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 		lastResourceVersion := findResourceVersion(obj.ID(), resourceVersions.Versions)
 		currentResourceVersion, err := cloudeventstypes.ToInteger(obj.Extensions()[types.ExtensionResourceVersion])
 		if err != nil {
-			log.V(4).Info("ignore the obj since it has a invalid resourceVersion", "object", obj, "error", err)
+			log.V(4).Info("ignore the event since it has a invalid resourceVersion", "eventContext", obj.Context, "error", err)
 			continue
 		}
 
@@ -378,7 +378,8 @@ func (bkr *GRPCBroker) handleRes(
 	// 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.")
+		log.V(4).Info("skip the event since the agent is not subscribed.",
+			"eventContext", evt.Context, "clusterName", clusterName)
 		return nil
 	}
 

test/integration/cloudevents/cloudevetns_grpc_v2_test.go

@@ -16,7 +16,7 @@ var _ = ginkgo.Describe("CloudEvents Clients Test - GRPC V2", runCloudeventsClie
 
 // The GRPC test simulates there is a server between the source and agent, the GRPC source client
 // sends/receives events to/from server, then server forward the events to agent via GRPC broker.
-func GetGRPCSourceOptionsV2(ctx context.Context, sourceID string) (*options.CloudEventsSourceOptions, string) {
+func GetGRPCSourceOptionsV2(_ context.Context, sourceID string) (*options.CloudEventsSourceOptions, string) {
 	return grpcv2.NewSourceOptions(
 			util.NewGRPCSourceOptions(grpcServerHost), sourceID, payload.ManifestBundleEventDataType),
 		constants.ConfigTypeGRPC