grpc broker (#115)

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

Author: skeeey

pkg/cloudevents/server/grpc/options/server.go

@@ -5,14 +5,15 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	"os"
+
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server"
 	grpcserver "open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authn"
-	"os"
 )
 
 // PreStartHook is an interface to start hook before grpc server is started.
@@ -99,7 +100,7 @@ func (s *Server) Run(ctx context.Context) error {
 		grpc.ChainStreamInterceptor(newAuthStreamInterceptor(s.authenticators...)))
 
 	grpcServer := grpc.NewServer(grpcServerOptions...)
-	grpcEventServer := grpcserver.NewGRPCBroker(grpcServer, ":"+s.options.ServerBindPort)
+	grpcEventServer := grpcserver.NewGRPCBroker(grpcServer)
 
 	for t, service := range s.services {
 		grpcEventServer.RegisterService(t, service)
@@ -110,7 +111,7 @@ func (s *Server) Run(ctx context.Context) error {
 		hook.Run(ctx)
 	}
 
-	go grpcEventServer.Start(ctx)
+	go grpcEventServer.Start(ctx, ":"+s.options.ServerBindPort)
 	<-ctx.Done()
 	return nil
 }

pkg/cloudevents/server/grpc/server.go

@@ -45,20 +45,18 @@ type GRPCBroker struct {
 	pbv1.UnimplementedCloudEventServiceServer
 	grpcServer  *grpc.Server
 	services    map[types.CloudEventsDataType]server.Service
-	bindAddress string
 	subscribers map[string]*subscriber // registered subscribers
 	mu          sync.RWMutex
 }
 
-// NewGRPCBroker creates a new gRPC broker with the given configuration.
-func NewGRPCBroker(srv *grpc.Server, addr string) server.AgentEventServer {
-	klog.Infof("Serving gRPC broker without TLS at %s", addr)
+// NewGRPCBroker creates a new gRPC broker with the given gRPC server.
+func NewGRPCBroker(srv *grpc.Server) server.AgentEventServer {
 	broker := &GRPCBroker{
 		grpcServer:  srv,
-		bindAddress: addr,
 		subscribers: make(map[string]*subscriber),
 		services:    make(map[types.CloudEventsDataType]server.Service),
 	}
+	pbv1.RegisterCloudEventServiceServer(broker.grpcServer, broker)
 	return broker
 }
 
@@ -67,15 +65,14 @@ func (bkr *GRPCBroker) RegisterService(t types.CloudEventsDataType, service serv
 	service.RegisterHandler(bkr)
 }
 
-// Start starts the gRPC broker
-func (bkr *GRPCBroker) Start(ctx context.Context) {
+// Start starts the gRPC broker at the given address
+func (bkr *GRPCBroker) Start(ctx context.Context, addr string) {
 	logger := klog.FromContext(ctx)
-	logger.Info("Starting gRPC broker")
-	lis, err := net.Listen("tcp", bkr.bindAddress)
+	logger.Info("Starting gRPC broker at %s", addr)
+	lis, err := net.Listen("tcp", addr)
 	if err != nil {
 		utilruntime.Must(fmt.Errorf("failed to listen: %v", err))
 	}
-	pbv1.RegisterCloudEventServiceServer(bkr.grpcServer, bkr)
 	go func() {
 		if err := bkr.grpcServer.Serve(lis); err != nil {
 			utilruntime.Must(fmt.Errorf("failed to serve gRPC broker: %v", err))
@@ -87,7 +84,7 @@ func (bkr *GRPCBroker) Start(ctx context.Context) {
 	klog.Infof("Shutting down gRPC broker")
 }
 
-// Publish in stub implementation for maestro agent publish resource status back to maestro server.
+// Publish in stub implementation for agent publish resource status.
 func (bkr *GRPCBroker) Publish(ctx context.Context, pubReq *pbv1.PublishRequest) (*emptypb.Empty, error) {
 	logger := klog.FromContext(ctx)
 	// WARNING: don't use "evt, err := pb.FromProto(pubReq.Event)" to convert protobuf to cloudevent
@@ -101,7 +98,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)
+	logger.Info("receive the event with grpc broker", "event", evt)
 
 	// handler resync request
 	if eventType.Action == types.ResyncRequestAction {
@@ -145,7 +142,7 @@ func (bkr *GRPCBroker) register(
 		errChan:     errChan,
 	}
 
-	klog.V(4).Infof("register a subscriber %s (cluster name = %s)", id, clusterName)
+	klog.Infof("register a subscriber %s (cluster name = %s)", id, clusterName)
 
 	return id, errChan
 }
@@ -160,9 +157,9 @@ func (bkr *GRPCBroker) unregister(id string) {
 	delete(bkr.subscribers, id)
 }
 
-// Subscribe in stub implementation for maestro agent subscribe resource spec from maestro server.
+// Subscribe in stub implementation for agent subscribe resource spec.
 // Note: It's unnecessary to send a status resync request to agent subscribers.
-// The work agent will continuously attempt to send status updates to the gRPC broker.
+// The agent will continuously attempt to send status updates to the gRPC broker.
 // If the broker is down or disconnected, the agent will resend the status once the broker is back up or reconnected.
 func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv1.CloudEventService_SubscribeServer) error {
 	if len(subReq.ClusterName) == 0 {
@@ -182,7 +179,7 @@ func (bkr *GRPCBroker) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
 		}
 
 		// send the cloudevent to the subscriber
-		klog.V(4).Infof("sending the event to spec subscribers, %s", evt)
+		klog.Infof("sending the event to spec subscribers, %s", evt)
 		if err := subServer.Send(pbEvt); err != nil {
 			klog.Errorf("failed to send grpc event, %v", err)
 			// Return the error without wrapping, as it includes the gRPC error code and message for further handling.

pkg/cloudevents/server/grpc/server_test.go

@@ -3,12 +3,13 @@ package grpc
 import (
 	"context"
 	"errors"
+	"testing"
+
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"google.golang.org/grpc"
 	grpccli "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc"
 	cetypes "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server"
-	"testing"
 )
 
 var dataType = cetypes.CloudEventsDataType{
@@ -58,14 +59,14 @@ func (s *testService) create(evt *cloudevents.Event) error {
 func TestServer(t *testing.T) {
 	grpcServerOptions := []grpc.ServerOption{}
 	grpcServer := grpc.NewServer(grpcServerOptions...)
-	grpcEventServer := NewGRPCBroker(grpcServer, ":8888")
+	grpcEventServer := NewGRPCBroker(grpcServer)
 
 	svc := &testService{evts: make(map[string]*cloudevents.Event)}
 	grpcEventServer.RegisterService(dataType, svc)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	go grpcEventServer.Start(ctx)
+	go grpcEventServer.Start(ctx, ":8888")
 
 	grpcClientOptions := grpccli.NewGRPCOptions()
 	grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: "localhost:8888"}

pkg/cloudevents/server/interface.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
 
@@ -15,7 +16,7 @@ type AgentEventServer interface {
 	RegisterService(t types.CloudEventsDataType, service Service)
 
 	// Start initiates the EventServer to listen to agents.
-	Start(ctx context.Context)
+	Start(ctx context.Context, addr string)
 }
 
 type EventHandler interface {