refactor grpc broker interface (#193)

Signed-off-by: Wei Liu <liuweixa@redhat.com>

Author: skeeey

pkg/cloudevents/clients/serviceaccount/integration_test.go

@@ -3,7 +3,6 @@ package serviceaccount
 import (
 	"context"
 	"net"
-	"sync"
 	"testing"
 	"time"
 
@@ -21,36 +20,11 @@ import (
 
 // tokenRequestService implements a service that handles token requests
 type tokenRequestService struct {
-	mu                   sync.RWMutex
-	tokenRequests        map[string]*authenticationv1.TokenRequest
 	handler              server.EventHandler
 	handleStatusUpdateFn func(context.Context, *cloudevents.Event) error
 }
 
-func (s *tokenRequestService) Get(ctx context.Context, resourceID string) (*cloudevents.Event, error) {
-	s.mu.RLock()
-	tokenReq, ok := s.tokenRequests[resourceID]
-	s.mu.RUnlock()
-	if !ok {
-		return nil, nil
-	}
-
-	codec := NewTokenRequestCodec()
-	eventType := cetypes.CloudEventsType{
-		CloudEventsDataType: TokenRequestDataType,
-		SubResource:         cetypes.SubResourceStatus,
-		Action:              cetypes.UpdateRequestAction,
-	}
-
-	evt, err := codec.Encode("test-source", eventType, tokenReq)
-	if err != nil {
-		return nil, err
-	}
-
-	return evt, nil
-}
-
-func (s *tokenRequestService) List(listOpts cetypes.ListOptions) ([]*cloudevents.Event, error) {
+func (s *tokenRequestService) List(_ context.Context, listOpts cetypes.ListOptions) ([]*cloudevents.Event, error) {
 	return nil, nil
 }
 
@@ -71,15 +45,23 @@ func (s *tokenRequestService) HandleStatusUpdate(ctx context.Context, evt *cloud
 		ExpirationTimestamp: metav1.Time{Time: time.Now().Add(1 * time.Hour)},
 	}
 
-	s.mu.Lock()
-	s.tokenRequests[string(tokenReq.UID)] = tokenReq
-	s.mu.Unlock()
-
 	// Notify the handler that we have a response ready
 	if s.handler == nil {
 		return nil
 	}
-	return s.handler.OnCreate(ctx, TokenRequestDataType, string(tokenReq.UID))
+
+	eventType := cetypes.CloudEventsType{
+		CloudEventsDataType: TokenRequestDataType,
+		SubResource:         cetypes.SubResourceSpec,
+		Action:              cetypes.UpdateRequestAction,
+	}
+
+	createEvt, err := codec.Encode("test-source", eventType, tokenReq)
+	if err != nil {
+		return err
+	}
+
+	return s.handler.HandleEvent(ctx, createEvt)
 }
 
 func (s *tokenRequestService) RegisterHandler(_ context.Context, handler server.EventHandler) {
@@ -95,9 +77,7 @@ func TestCreateToken_Integration(t *testing.T) {
 	pbv1.RegisterCloudEventServiceServer(grpcServer, grpcEventServer)
 
 	// Create and register the token request service
-	svc := &tokenRequestService{
-		tokenRequests: make(map[string]*authenticationv1.TokenRequest),
-	}
+	svc := &tokenRequestService{}
 	grpcEventServer.RegisterService(context.Background(), TokenRequestDataType, svc)
 
 	// Start listening
@@ -172,7 +152,6 @@ func TestCreateToken_Timeout(t *testing.T) {
 
 	// Create a service that receives but never responds
 	svc := &tokenRequestService{
-		tokenRequests: make(map[string]*authenticationv1.TokenRequest),
 		handleStatusUpdateFn: func(ctx context.Context, evt *cloudevents.Event) error {
 			// Just receive but don't respond
 			return nil
@@ -231,9 +210,7 @@ func TestCreateToken_MultipleRequests(t *testing.T) {
 	grpcEventServer := grpcserver.NewGRPCBroker(grpcserver.NewBrokerOptions())
 	pbv1.RegisterCloudEventServiceServer(grpcServer, grpcEventServer)
 
-	svc := &tokenRequestService{
-		tokenRequests: make(map[string]*authenticationv1.TokenRequest),
-	}
+	svc := &tokenRequestService{}
 	grpcEventServer.RegisterService(context.Background(), TokenRequestDataType, svc)
 
 	lis, err := net.Listen("tcp", "127.0.0.1:0")

pkg/cloudevents/server/grpc/broker.go

@@ -11,8 +11,6 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/heartbeat"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/metrics"
 
-	"k8s.io/apimachinery/pkg/api/errors"
-
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
 	cloudeventstypes "github.com/cloudevents/sdk-go/v2/types"
@@ -315,30 +313,36 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 		return fmt.Errorf("failed to find service for event type %s", eventDataType)
 	}
 
-	objs, err := service.List(types.ListOptions{ClusterName: clusterName, CloudEventsDataType: eventDataType})
+	evts, err := service.List(ctx, types.ListOptions{ClusterName: clusterName, CloudEventsDataType: eventDataType})
 	if err != nil {
 		return err
 	}
 
-	if len(objs) == 0 {
+	if len(evts) == 0 {
 		log.V(4).Info("no objs from the lister, do nothing")
 		return nil
 	}
 
-	for _, obj := range objs {
+	for _, evt := range evts {
 		// respond with the deleting resource regardless of the resource version
-		objLogger := log.WithValues("eventType", obj.Type(), "extensions", obj.Extensions())
-		if _, ok := obj.Extensions()[types.ExtensionDeletionTimestamp]; ok {
+		objLogger := log.WithValues("eventType", evt.Type(), "extensions", evt.Extensions())
+		if _, ok := evt.Extensions()[types.ExtensionDeletionTimestamp]; ok {
 			objLogger.V(4).Info("respond spec resync request")
-			err = bkr.handleRes(ctx, obj, eventDataType, "delete_request")
+			deleteEventTypes := types.CloudEventsType{
+				CloudEventsDataType: eventDataType,
+				SubResource:         types.SubResourceSpec,
+				Action:              types.DeleteRequestAction,
+			}
+			evt.SetType(deleteEventTypes.String())
+			err = bkr.HandleEvent(ctx, evt)
 			if err != nil {
 				objLogger.Error(err, "failed to handle resync spec request")
 			}
 			continue
 		}
 
-		lastResourceVersion := findResourceVersion(obj.ID(), resourceVersions.Versions)
-		currentResourceVersion, err := cloudeventstypes.ToInteger(obj.Extensions()[types.ExtensionResourceVersion])
+		lastResourceVersion := findResourceVersion(evt.ID(), resourceVersions.Versions)
+		currentResourceVersion, err := cloudeventstypes.ToInteger(evt.Extensions()[types.ExtensionResourceVersion])
 		if err != nil {
 			objLogger.V(4).Info("ignore the event since it has a invalid resourceVersion", "error", err)
 			continue
@@ -348,7 +352,13 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 		// the newer work to agent
 		if currentResourceVersion == 0 || int64(currentResourceVersion) > lastResourceVersion {
 			objLogger.V(4).Info("respond spec resync request")
-			err := bkr.handleRes(ctx, obj, eventDataType, "update_request")
+			updateEventTypes := types.CloudEventsType{
+				CloudEventsDataType: eventDataType,
+				SubResource:         types.SubResourceSpec,
+				Action:              types.UpdateRequestAction,
+			}
+			evt.SetType(updateEventTypes.String())
+			err := bkr.HandleEvent(ctx, evt)
 			if err != nil {
 				objLogger.Error(err, "failed to handle resync spec request")
 			}
@@ -357,14 +367,15 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 
 	// the resources do not exist on the source, but exist on the agent, delete them
 	for _, rv := range resourceVersions.Versions {
-		_, exists := getObj(rv.ResourceID, objs)
+		_, exists := getObj(rv.ResourceID, evts)
 		if exists {
 			continue
 		}
 
 		deleteEventTypes := types.CloudEventsType{
 			CloudEventsDataType: eventDataType,
 			SubResource:         types.SubResourceSpec,
+			Action:              types.DeleteRequestAction,
 		}
 		obj := types.NewEventBuilder("source", deleteEventTypes).
 			WithResourceID(rv.ResourceID).
@@ -375,7 +386,7 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 
 		// send a delete event for the current resource
 		log.V(4).Info("respond spec resync request")
-		err := bkr.handleRes(ctx, &obj, eventDataType, "delete_request")
+		err := bkr.HandleEvent(ctx, &obj)
 		if err != nil {
 			log.Error(err, "failed to handle delete request")
 		}
@@ -384,22 +395,20 @@ func (bkr *GRPCBroker) respondResyncSpecRequest(ctx context.Context, eventDataTy
 	return nil
 }
 
-// handleRes publish the resource to the correct subscriber.
-func (bkr *GRPCBroker) handleRes(
-	ctx context.Context,
-	evt *cloudevents.Event,
-	t types.CloudEventsDataType,
-	action types.EventAction) error {
+// HandleEvent publish the event to the correct subscriber.
+func (bkr *GRPCBroker) HandleEvent(ctx context.Context, evt *cloudevents.Event) error {
+	if evt == nil {
+		return fmt.Errorf("event is nil")
+	}
 
 	bkr.mu.RLock()
 	defer bkr.mu.RUnlock()
 
-	eventType := types.CloudEventsType{
-		CloudEventsDataType: t,
-		SubResource:         types.SubResourceSpec,
-		Action:              action,
+	eventType, err := types.ParseCloudEventsType(evt.Type())
+	if err != nil {
+		return err
 	}
-	evt.SetType(eventType.String())
+	evtDataType := eventType.CloudEventsDataType
 
 	clusterNameValue, err := evt.Context.GetExtension(types.ExtensionClusterName)
 	if err != nil {
@@ -412,7 +421,7 @@ func (bkr *GRPCBroker) handleRes(
 		// the resource consumer name and its data type 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 subscriber.clusterName == clusterName && subscriber.dataType == evtDataType {
 			if err := subscriber.handler(ctx, subID, evt); err != nil {
 				return err
 			}
@@ -421,63 +430,6 @@ func (bkr *GRPCBroker) handleRes(
 	return nil
 }
 
-// OnCreate is called by the controller when a resource is created on the maestro server.
-func (bkr *GRPCBroker) OnCreate(ctx context.Context, t types.CloudEventsDataType, id string) error {
-	service, ok := bkr.services[t]
-	if !ok {
-		return fmt.Errorf("failed to find service for event type %s", t)
-	}
-
-	resource, err := service.Get(ctx, id)
-	// if the resource is not found, it indicates the resource has been processed.
-	if errors.IsNotFound(err) {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-
-	return bkr.handleRes(ctx, resource, t, "create_request")
-}
-
-// OnUpdate is called by the controller when a resource is updated on the maestro server.
-func (bkr *GRPCBroker) OnUpdate(ctx context.Context, t types.CloudEventsDataType, id string) error {
-	service, ok := bkr.services[t]
-	if !ok {
-		return fmt.Errorf("failed to find service for event type %s", t)
-	}
-
-	resource, err := service.Get(ctx, id)
-	// if the resource is not found, it indicates the resource has been processed.
-	if errors.IsNotFound(err) {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-
-	return bkr.handleRes(ctx, resource, t, "update_request")
-}
-
-// OnDelete is called by the controller when a resource is deleted from the maestro server.
-func (bkr *GRPCBroker) OnDelete(ctx context.Context, t types.CloudEventsDataType, id string) error {
-	service, ok := bkr.services[t]
-	if !ok {
-		return fmt.Errorf("failed to find service for event type %s", t)
-	}
-
-	resource, err := service.Get(ctx, id)
-	// if the resource is not found, it indicates the resource has been processed.
-	if errors.IsNotFound(err) {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-
-	return bkr.handleRes(ctx, resource, t, "delete_request")
-}
-
 // IsConsumerSubscribed returns true if the consumer is subscribed to the broker for resource spec.
 func (bkr *GRPCBroker) IsConsumerSubscribed(consumerName string) bool {
 	bkr.mu.RLock()

pkg/cloudevents/server/grpc/broker_test.go

@@ -2,7 +2,6 @@ package grpc
 
 import (
 	"context"
-	"errors"
 	"net"
 	"testing"
 
@@ -26,16 +25,8 @@ type testService struct {
 	handler server.EventHandler
 }
 
-func (s *testService) Get(ctx context.Context, resourceID string) (*cloudevents.Event, error) {
-	evt, ok := s.evts[resourceID]
-	if !ok {
-		return nil, errors.New("not found")
-	}
-	return evt, nil
-}
-
 // List the cloudEvent from the service
-func (s *testService) List(listOpts cetypes.ListOptions) ([]*cloudevents.Event, error) {
+func (s *testService) List(_ context.Context, listOpts cetypes.ListOptions) ([]*cloudevents.Event, error) {
 	evts := make([]*cloudevents.Event, 0, len(s.evts))
 	for _, evt := range s.evts {
 		evts = append(evts, evt)
@@ -56,7 +47,7 @@ func (s *testService) RegisterHandler(_ context.Context, handler server.EventHan
 
 func (s *testService) create(evt *cloudevents.Event) error {
 	s.evts[evt.ID()] = evt
-	return s.handler.OnCreate(context.TODO(), dataType, evt.ID())
+	return s.handler.HandleEvent(context.TODO(), evt)
 }
 
 func TestServer(t *testing.T) {

pkg/cloudevents/server/interface.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 
+	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
@@ -22,14 +23,8 @@ type AgentEventServer interface {
 }
 
 type EventHandler interface {
-	// OnCreate is the callback when resource is created in the service.
-	OnCreate(ctx context.Context, t types.CloudEventsDataType, resourceID string) error
-
-	// OnUpdate is the callback when resource is updated in the service.
-	OnUpdate(ctx context.Context, t types.CloudEventsDataType, resourceID string) error
-
-	// OnDelete is the callback when resource is deleted from the service.
-	OnDelete(ctx context.Context, t types.CloudEventsDataType, resourceID string) error
+	// HandleEvent publish the event to the correct subscriber.
+	HandleEvent(ctx context.Context, evt *cloudevents.Event) error
 }
 
 // TODO SourceEventServer to handle the grpc conversation between consumers and grpcserver.

pkg/cloudevents/server/store.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cetypes "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 )
@@ -11,11 +12,8 @@ import (
 
 // TODO need a method to check if an event has been processed already.
 type Service interface {
-	// Get the cloudEvent based on resourceID from the service
-	Get(ctx context.Context, resourceID string) (*cloudevents.Event, error)
-
 	// List the cloudEvent from the service
-	List(listOpts cetypes.ListOptions) ([]*cloudevents.Event, error)
+	List(ctx context.Context, listOpts cetypes.ListOptions) ([]*cloudevents.Event, error)
 
 	// HandleStatusUpdate processes the resource status update from the agent.
 	HandleStatusUpdate(ctx context.Context, evt *cloudevents.Event) error

pkg/server/grpc/metrics/metrics_test.go

@@ -207,11 +207,7 @@ func newMockWorkService() *mockWorkService {
 	return &mockWorkService{}
 }
 
-func (s *mockWorkService) Get(ctx context.Context, resourceID string) (*cloudevents.Event, error) {
-	return nil, nil
-}
-
-func (s *mockWorkService) List(listOpts cetypes.ListOptions) ([]*cloudevents.Event, error) {
+func (s *mockWorkService) List(_ context.Context, listOpts cetypes.ListOptions) ([]*cloudevents.Event, error) {
 	return nil, nil
 }