diff --git a/cmd/ateapi/internal/controlapi/actor.go b/cmd/ateapi/internal/controlapi/actor.go index f89bf5296..ad7c89caa 100644 --- a/cmd/ateapi/internal/controlapi/actor.go +++ b/cmd/ateapi/internal/controlapi/actor.go @@ -34,7 +34,7 @@ import ( "k8s.io/apimachinery/pkg/util/validation/field" ) -func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (created *ateapipb.Actor, err error) { +func (s *RPCService) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (created *ateapipb.Actor, err error) { if errs := validateCreateActorRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -118,7 +118,7 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ // resolveSnapshotSource resolves a CreateActor request's source snapshot tag // and checks that its scope and ActorSnapshot are compatible with creating // an Actor in actorAtespace from template. -func (s *Service) resolveSnapshotSource(ctx context.Context, actorAtespace string, tagRef *ateapipb.ObjectRef, template *atev1alpha1.ActorTemplate) (*ateapipb.ActorSourceSnapshotStatus, error) { +func (s *RPCService) resolveSnapshotSource(ctx context.Context, actorAtespace string, tagRef *ateapipb.ObjectRef, template *atev1alpha1.ActorTemplate) (*ateapipb.ActorSourceSnapshotStatus, error) { tag, err := s.persistence.GetActorSnapshotTag(ctx, tagRef.GetAtespace(), tagRef.GetName()) if errors.Is(err, store.ErrNotFound) { return nil, status.Error(codes.NotFound, "ActorSnapshot not found") @@ -210,7 +210,7 @@ func validateCreateActorRequest(req *ateapipb.CreateActorRequest) field.ErrorLis return errs } -func (s *Service) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (*ateapipb.Actor, error) { +func (s *RPCService) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (*ateapipb.Actor, error) { if errs := validateGetActorRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -237,7 +237,7 @@ func validateGetActorRequest(req *ateapipb.GetActorRequest) field.ErrorList { return errs } -func (s *Service) ListActors(ctx context.Context, req *ateapipb.ListActorsRequest) (*ateapipb.ListActorsResponse, error) { +func (s *RPCService) ListActors(ctx context.Context, req *ateapipb.ListActorsRequest) (*ateapipb.ListActorsResponse, error) { if errs := validateListActorsRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -275,7 +275,7 @@ var actorMutableFields = fieldmask.NewMutableFields( "worker_selector.match_labels", ) -func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequest) (*ateapipb.Actor, error) { +func (s *RPCService) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequest) (*ateapipb.Actor, error) { if errs := validateUpdateActorRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -328,7 +328,7 @@ func validateUpdateActorRequest(req *ateapipb.UpdateActorRequest) field.ErrorLis return errs } -func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequest) (deleted *ateapipb.Actor, err error) { +func (s *RPCService) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequest) (deleted *ateapipb.Actor, err error) { if errs := validateDeleteActorRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -370,7 +370,7 @@ func validateDeleteActorRequest(req *ateapipb.DeleteActorRequest) field.ErrorLis return errs } -func (s *Service) PauseActor(ctx context.Context, req *ateapipb.PauseActorRequest) (*ateapipb.PauseActorResponse, error) { +func (s *RPCService) PauseActor(ctx context.Context, req *ateapipb.PauseActorRequest) (*ateapipb.PauseActorResponse, error) { if errs := validatePauseActorRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -405,7 +405,7 @@ func validatePauseActorRequest(req *ateapipb.PauseActorRequest) field.ErrorList return errs } -func (s *Service) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequest) (*ateapipb.ResumeActorResponse, error) { +func (s *RPCService) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequest) (*ateapipb.ResumeActorResponse, error) { if errs := validateResumeActorRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -440,7 +440,7 @@ func validateResumeActorRequest(req *ateapipb.ResumeActorRequest) field.ErrorLis return errs } -func (s *Service) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) { +func (s *RPCService) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRequest) (*ateapipb.SuspendActorResponse, error) { if errs := validateSuspendActorRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } diff --git a/cmd/ateapi/internal/controlapi/actor_snapshot.go b/cmd/ateapi/internal/controlapi/actor_snapshot.go index 6323b2064..f18ad47cb 100644 --- a/cmd/ateapi/internal/controlapi/actor_snapshot.go +++ b/cmd/ateapi/internal/controlapi/actor_snapshot.go @@ -47,7 +47,7 @@ var actorSnapshotTagScopeNames = func() []string { return names }() -func (s *Service) GetActorSnapshot(ctx context.Context, req *ateapipb.GetActorSnapshotRequest) (*ateapipb.ActorSnapshot, error) { +func (s *RPCService) GetActorSnapshot(ctx context.Context, req *ateapipb.GetActorSnapshotRequest) (*ateapipb.ActorSnapshot, error) { if errs := validateGetActorSnapshotRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -74,7 +74,7 @@ func validateGetActorSnapshotRequest(req *ateapipb.GetActorSnapshotRequest) fiel return errs } -func (s *Service) GetActorSnapshotTag(ctx context.Context, req *ateapipb.GetActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { +func (s *RPCService) GetActorSnapshotTag(ctx context.Context, req *ateapipb.GetActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { if errs := validateGetActorSnapshotTagRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -101,7 +101,7 @@ func validateGetActorSnapshotTagRequest(req *ateapipb.GetActorSnapshotTagRequest return errs } -func (s *Service) ListActorSnapshots(ctx context.Context, req *ateapipb.ListActorSnapshotsRequest) (*ateapipb.ListActorSnapshotsResponse, error) { +func (s *RPCService) ListActorSnapshots(ctx context.Context, req *ateapipb.ListActorSnapshotsRequest) (*ateapipb.ListActorSnapshotsResponse, error) { if errs := validateListActorSnapshotsRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -128,7 +128,7 @@ func validateListActorSnapshotsRequest(req *ateapipb.ListActorSnapshotsRequest) return errs } -func (s *Service) CreateActorSnapshotTag(ctx context.Context, req *ateapipb.CreateActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { +func (s *RPCService) CreateActorSnapshotTag(ctx context.Context, req *ateapipb.CreateActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { if errs := validateCreateActorSnapshotTagRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -180,7 +180,7 @@ func validateCreateActorSnapshotTagRequest(req *ateapipb.CreateActorSnapshotTagR // may name in an UpdateActorSnapshotTag update_mask. var actorSnapshotTagMutableFields = fieldmask.NewMutableFields("scope") -func (s *Service) UpdateActorSnapshotTag(ctx context.Context, req *ateapipb.UpdateActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { +func (s *RPCService) UpdateActorSnapshotTag(ctx context.Context, req *ateapipb.UpdateActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { if errs := validateUpdateActorSnapshotTagRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -228,7 +228,7 @@ func validateUpdateActorSnapshotTagRequest(req *ateapipb.UpdateActorSnapshotTagR return errs } -func (s *Service) DeleteActorSnapshotTag(ctx context.Context, req *ateapipb.DeleteActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { +func (s *RPCService) DeleteActorSnapshotTag(ctx context.Context, req *ateapipb.DeleteActorSnapshotTagRequest) (*ateapipb.ActorSnapshotTag, error) { if errs := validateDeleteActorSnapshotTagRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } diff --git a/cmd/ateapi/internal/controlapi/actor_snapshot_test.go b/cmd/ateapi/internal/controlapi/actor_snapshot_test.go index 2f05cb6f3..681ffb7b3 100644 --- a/cmd/ateapi/internal/controlapi/actor_snapshot_test.go +++ b/cmd/ateapi/internal/controlapi/actor_snapshot_test.go @@ -295,7 +295,7 @@ func TestUpdateActorSnapshotTag_FieldMasks(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { tt.stored.Metadata = &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "tag1"} - svc, stored := serviceWithActorSnapshotTag(t, tt.stored) + svc, stored := rpcServiceWithActorSnapshotTag(t, tt.stored) tt.req.Metadata = stored.GetMetadata() // Sent but not in the mask - must be ignored. @@ -322,7 +322,7 @@ func TestUpdateActorSnapshotTag_FieldMasks(t *testing.T) { // scope without populating it is rejected. func TestUpdateActorSnapshotTag_UnsetScopeDoesNotUnpublish(t *testing.T) { ctx := context.Background() - svc, stored := serviceWithActorSnapshotTag(t, &ateapipb.ActorSnapshotTag{ + svc, stored := rpcServiceWithActorSnapshotTag(t, &ateapipb.ActorSnapshotTag{ Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "tag1"}, Scope: ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_PUBLISHED, }) @@ -354,7 +354,7 @@ func TestUpdateActorSnapshotTag_UnsetScopeDoesNotUnpublish(t *testing.T) { // creation. func TestCreateActorSnapshotTag_RejectsUnsetScope(t *testing.T) { ctx := context.Background() - svc, stored := serviceWithActorSnapshotTag(t, &ateapipb.ActorSnapshotTag{ + svc, stored := rpcServiceWithActorSnapshotTag(t, &ateapipb.ActorSnapshotTag{ Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "tag1"}, Scope: ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_ATESPACE, }) @@ -370,9 +370,9 @@ func TestCreateActorSnapshotTag_RejectsUnsetScope(t *testing.T) { } } -// serviceWithActorSnapshotTag seeds an ActorSnapshot and a tag pointing at it -// in a miniredis-backed store, and returns a Service over it. -func serviceWithActorSnapshotTag(t *testing.T, tag *ateapipb.ActorSnapshotTag) (*Service, *ateapipb.ActorSnapshotTag) { +// rpcServiceWithActorSnapshotTag seeds an ActorSnapshot and a tag pointing at it +// in a miniredis-backed store, and returns a RPCService over it. +func rpcServiceWithActorSnapshotTag(t *testing.T, tag *ateapipb.ActorSnapshotTag) (*RPCService, *ateapipb.ActorSnapshotTag) { t.Helper() persistence, cleanup := storetest.SetupTestStore(t) t.Cleanup(cleanup) @@ -390,7 +390,7 @@ func serviceWithActorSnapshotTag(t *testing.T, tag *ateapipb.ActorSnapshotTag) ( if err != nil { t.Fatalf("Failed to CreateActorSnapshotTag: %v", err) } - return &Service{persistence: persistence}, created + return &RPCService{persistence: persistence}, created } // TestUpdateActorSnapshotTag_DeleteRecreateRace checks that an update is not @@ -438,7 +438,7 @@ func TestUpdateActorSnapshotTag_DeleteRecreateRace(t *testing.T) { } }, } - svc := &Service{persistence: racing} + svc := &RPCService{persistence: racing} // The client asserts "only update the tag with uid A". Its version guard is // satisfied by B as well, because re-tagging resets the version to 1: the @@ -503,7 +503,7 @@ func TestUpdateActorSnapshotTag_ConcurrentUpdate(t *testing.T) { } }, } - svc := &Service{persistence: racing} + svc := &RPCService{persistence: racing} originalTag.Scope = ateapipb.ActorSnapshotTagScope_ACTOR_SNAPSHOT_TAG_SCOPE_PUBLISHED _, err = svc.UpdateActorSnapshotTag(ctx, &ateapipb.UpdateActorSnapshotTagRequest{ diff --git a/cmd/ateapi/internal/controlapi/actor_template.go b/cmd/ateapi/internal/controlapi/actor_template.go index 2f4f122bb..7e3beb6b2 100644 --- a/cmd/ateapi/internal/controlapi/actor_template.go +++ b/cmd/ateapi/internal/controlapi/actor_template.go @@ -28,7 +28,7 @@ import ( "k8s.io/apimachinery/pkg/util/validation/field" ) -func (s *Service) CreateActorTemplate(ctx context.Context, req *ateapipb.CreateActorTemplateRequest) (*ateapipb.ActorTemplate, error) { +func (s *RPCService) CreateActorTemplate(ctx context.Context, req *ateapipb.CreateActorTemplateRequest) (*ateapipb.ActorTemplate, error) { if errs := validateCreateActorTemplateRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -148,7 +148,7 @@ func validateCreateActorTemplateRequest(req *ateapipb.CreateActorTemplateRequest return errs } -func (s *Service) GetActorTemplate(ctx context.Context, req *ateapipb.GetActorTemplateRequest) (*ateapipb.ActorTemplate, error) { +func (s *RPCService) GetActorTemplate(ctx context.Context, req *ateapipb.GetActorTemplateRequest) (*ateapipb.ActorTemplate, error) { if errs := validateGetActorTemplateRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -177,7 +177,7 @@ func validateGetActorTemplateRequest(req *ateapipb.GetActorTemplateRequest) fiel return errs } -func (s *Service) ListActorTemplates(ctx context.Context, req *ateapipb.ListActorTemplatesRequest) (*ateapipb.ListActorTemplatesResponse, error) { +func (s *RPCService) ListActorTemplates(ctx context.Context, req *ateapipb.ListActorTemplatesRequest) (*ateapipb.ListActorTemplatesResponse, error) { if errs := validateListActorTemplatesRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -208,7 +208,7 @@ func validateListActorTemplatesRequest(req *ateapipb.ListActorTemplatesRequest) return errs } -func (s *Service) DeleteActorTemplate(ctx context.Context, req *ateapipb.DeleteActorTemplateRequest) (*ateapipb.ActorTemplate, error) { +func (s *RPCService) DeleteActorTemplate(ctx context.Context, req *ateapipb.DeleteActorTemplateRequest) (*ateapipb.ActorTemplate, error) { if errs := validateDeleteActorTemplateRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } diff --git a/cmd/ateapi/internal/controlapi/actor_template_test.go b/cmd/ateapi/internal/controlapi/actor_template_test.go index 45967d598..ce3ec7550 100644 --- a/cmd/ateapi/internal/controlapi/actor_template_test.go +++ b/cmd/ateapi/internal/controlapi/actor_template_test.go @@ -172,7 +172,7 @@ func TestValidateCreateActorTemplateRequest(t *testing.T) { // while the atespace is missing, and succeeds once the atespace exists. func TestCreateActorTemplate(t *testing.T) { persistence := newTestPersistence(t) - s := &Service{persistence: persistence} + s := &RPCService{persistence: persistence} ctx := context.Background() req := func(atespace, name string) *ateapipb.CreateActorTemplateRequest { return &ateapipb.CreateActorTemplateRequest{ActorTemplate: validActorTemplate(func(tmpl *ateapipb.ActorTemplate) { @@ -202,7 +202,7 @@ func TestCreateActorTemplate(t *testing.T) { // guard. func TestCreateActorTemplateIgnoresServerOwnedFields(t *testing.T) { persistence := newTestPersistence(t) - s := &Service{persistence: persistence} + s := &RPCService{persistence: persistence} ctx := context.Background() if _, err := persistence.CreateAtespace(ctx, &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: "ns1"}}); err != nil { diff --git a/cmd/ateapi/internal/controlapi/actor_test.go b/cmd/ateapi/internal/controlapi/actor_test.go index 173836b4e..3a9c7934e 100644 --- a/cmd/ateapi/internal/controlapi/actor_test.go +++ b/cmd/ateapi/internal/controlapi/actor_test.go @@ -361,7 +361,7 @@ func TestUpdateActor_FieldMasks(t *testing.T) { tt.stored.Metadata = &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: testActorID} tt.stored.ActorTemplateNamespace = "ns1" tt.stored.ActorTemplateName = "tmpl1" - svc, created := serviceWithActor(t, tt.stored) + svc, created := rpcServiceWithActor(t, tt.stored) tt.req.Metadata = created.GetMetadata() updated, err := svc.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{ @@ -433,7 +433,7 @@ func TestUpdateActor_DeleteRecreateRace(t *testing.T) { } }, } - svc := &Service{persistence: racing} + svc := &RPCService{persistence: racing} // The client asserts "only update the actor with uid A". original.WorkerSelector = &ateapipb.Selector{MatchLabels: map[string]string{"tier": "paid"}} @@ -501,7 +501,7 @@ func TestUpdateActor_ConcurrentDisjointUpdates(t *testing.T) { } }, } - svc := &Service{persistence: racing} + svc := &RPCService{persistence: racing} // Update operation is changing the worker_selector field, not the actor's state (like the concurrent op) // This update must fail: the racing update bumped the version. @@ -561,9 +561,9 @@ func withSelector(labels map[string]string) func(*ateapipb.UpdateActorRequest) { } } -// serviceWithActor seeds one actor in a miniredis-backed store and returns a -// Service over it. -func serviceWithActor(t *testing.T, actor *ateapipb.Actor) (*Service, *ateapipb.Actor) { +// rpcServiceWithActor seeds one actor in a miniredis-backed store and returns a +// RPCService over it. +func rpcServiceWithActor(t *testing.T, actor *ateapipb.Actor) (*RPCService, *ateapipb.Actor) { t.Helper() persistence, cleanup := storetest.SetupTestStore(t) t.Cleanup(cleanup) @@ -572,7 +572,7 @@ func serviceWithActor(t *testing.T, actor *ateapipb.Actor) (*Service, *ateapipb. if err != nil { t.Fatalf("Failed to CreateActor: %v", err) } - return &Service{persistence: persistence}, created + return &RPCService{persistence: persistence}, created } func TestValidateDeleteActorRequest(t *testing.T) { diff --git a/cmd/ateapi/internal/controlapi/atespace.go b/cmd/ateapi/internal/controlapi/atespace.go index 220275791..d45be0d69 100644 --- a/cmd/ateapi/internal/controlapi/atespace.go +++ b/cmd/ateapi/internal/controlapi/atespace.go @@ -27,7 +27,7 @@ import ( "k8s.io/apimachinery/pkg/util/validation/field" ) -func (s *Service) CreateAtespace(ctx context.Context, req *ateapipb.CreateAtespaceRequest) (*ateapipb.Atespace, error) { +func (s *RPCService) CreateAtespace(ctx context.Context, req *ateapipb.CreateAtespaceRequest) (*ateapipb.Atespace, error) { if errs := validateCreateAtespaceRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -74,7 +74,7 @@ func validateCreateAtespaceRequest(req *ateapipb.CreateAtespaceRequest) field.Er return errs } -func (s *Service) GetAtespace(ctx context.Context, req *ateapipb.GetAtespaceRequest) (*ateapipb.Atespace, error) { +func (s *RPCService) GetAtespace(ctx context.Context, req *ateapipb.GetAtespaceRequest) (*ateapipb.Atespace, error) { if errs := validateGetAtespaceRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -103,7 +103,7 @@ func validateGetAtespaceRequest(req *ateapipb.GetAtespaceRequest) field.ErrorLis return errs } -func (s *Service) ListAtespaces(ctx context.Context, req *ateapipb.ListAtespacesRequest) (*ateapipb.ListAtespacesResponse, error) { +func (s *RPCService) ListAtespaces(ctx context.Context, req *ateapipb.ListAtespacesRequest) (*ateapipb.ListAtespacesResponse, error) { if errs := validateListAtespacesRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -129,7 +129,7 @@ func validateListAtespacesRequest(req *ateapipb.ListAtespacesRequest) field.Erro return errs } -func (s *Service) DeleteAtespace(ctx context.Context, req *ateapipb.DeleteAtespaceRequest) (*ateapipb.Atespace, error) { +func (s *RPCService) DeleteAtespace(ctx context.Context, req *ateapipb.DeleteAtespaceRequest) (*ateapipb.Atespace, error) { if errs := validateDeleteAtespaceRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } diff --git a/cmd/ateapi/internal/controlapi/functionaltest/common_test.go b/cmd/ateapi/internal/controlapi/functionaltest/common_test.go index 1cd642a03..a40e715b9 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/common_test.go +++ b/cmd/ateapi/internal/controlapi/functionaltest/common_test.go @@ -76,7 +76,7 @@ var ( type testContext struct { mr *miniredis.Miniredis - service *controlapi.Service + service *controlapi.RPCService client ateapipb.ControlClient k8sClient kubernetes.Interface substrateClient versioned.Interface @@ -103,7 +103,7 @@ func setupTest(t *testing.T, ns string) *testContext { // setupTestWithVolumePlugins is setupTest with the default mock volume plugin // replaced by plugins, keyed by driver name. Tests that need a failure-injecting -// plugin pass it here rather than swapping it into the running Service, so each +// plugin pass it here rather than swapping it into the running RPCService, so each // test owns its own plugin set. func setupTestWithVolumePlugins(t *testing.T, ns string, plugins map[string]volume.VolumePluginControlPlane) *testContext { t.Helper() @@ -191,7 +191,7 @@ func setupTestWithVolumePlugins(t *testing.T, ns string, plugins map[string]volu mockDriverName: mockPlugin, } } - service := controlapi.NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins) + service := controlapi.NewRPCService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins) // 5. Start REAL gRPC Server for ATE API grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ateinterceptors.ServerUnaryInterceptor)) diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go index c50ccb755..ab34f6500 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -28,8 +28,9 @@ import ( storagev1listers "k8s.io/client-go/listers/storage/v1" ) -// Service implements ateapipb.Control -type Service struct { +// RPCService implements ateapipb.ControlServer. This is the user-facing RPC +// interface. +type RPCService struct { ateapipb.UnimplementedControlServer persistence serviceStore workerCache *workercache.Cache @@ -44,15 +45,15 @@ type Service struct { volumePlugins map[string]volume.VolumePluginControlPlane } -var _ ateapipb.ControlServer = (*Service)(nil) +var _ ateapipb.ControlServer = (*RPCService)(nil) // VolumePluginRegistry defines the interface for dynamic CSI plugin resolution. type VolumePluginRegistry interface { GetPlugin(ctx context.Context, name string) (volume.VolumePluginControlPlane, error) } -// NewService creates a service. instruments may be nil; the record helpers no-op. -func NewService( +// NewRPCService creates a RPC service. instruments may be nil; the record helpers no-op. +func NewRPCService( persistence store.Interface, workerCache *workercache.Cache, actorTemplateLister listersv1alpha1.ActorTemplateLister, @@ -64,8 +65,8 @@ func NewService( instruments *Instruments, egressGatewayAddress string, volumePlugins map[string]volume.VolumePluginControlPlane, -) *Service { - s := &Service{ +) *RPCService { + s := &RPCService{ persistence: persistence, workerCache: workerCache, actorTemplateLister: actorTemplateLister, @@ -107,7 +108,7 @@ type serviceStore interface { } // GetPlugin retrieves a CSI volume plugin by driver name, dynamically discovering it if not present. -func (s *Service) GetPlugin(ctx context.Context, driverName string) (volume.VolumePluginControlPlane, error) { +func (s *RPCService) GetPlugin(ctx context.Context, driverName string) (volume.VolumePluginControlPlane, error) { s.mu.RLock() plugin, ok := s.volumePlugins[driverName] s.mu.RUnlock() diff --git a/cmd/ateapi/internal/controlapi/worker.go b/cmd/ateapi/internal/controlapi/worker.go index 1577ae00d..85836e646 100644 --- a/cmd/ateapi/internal/controlapi/worker.go +++ b/cmd/ateapi/internal/controlapi/worker.go @@ -25,7 +25,7 @@ import ( "k8s.io/apimachinery/pkg/util/validation/field" ) -func (s *Service) ListWorkers(ctx context.Context, req *ateapipb.ListWorkersRequest) (*ateapipb.ListWorkersResponse, error) { +func (s *RPCService) ListWorkers(ctx context.Context, req *ateapipb.ListWorkersRequest) (*ateapipb.ListWorkersResponse, error) { if errs := validateListWorkersRequest(req); len(errs) > 0 { return nil, toGRPCStatusError(errs) } @@ -51,22 +51,22 @@ func validateListWorkersRequest(req *ateapipb.ListWorkersRequest) field.ErrorLis return errs } -func (s *Service) GetWorker(ctx context.Context, req *ateapipb.GetWorkerRequest) (*ateapipb.Worker, error) { +func (s *RPCService) GetWorker(ctx context.Context, req *ateapipb.GetWorkerRequest) (*ateapipb.Worker, error) { return nil, status.Error(codes.Unimplemented, "GetWorker is not implemented yet") } -func (s *Service) CreateWorker(ctx context.Context, req *ateapipb.CreateWorkerRequest) (*ateapipb.Worker, error) { +func (s *RPCService) CreateWorker(ctx context.Context, req *ateapipb.CreateWorkerRequest) (*ateapipb.Worker, error) { return nil, status.Error(codes.Unimplemented, "CreateWorker is not implemented yet") } -func (s *Service) UpdateWorker(ctx context.Context, req *ateapipb.UpdateWorkerRequest) (*ateapipb.Worker, error) { +func (s *RPCService) UpdateWorker(ctx context.Context, req *ateapipb.UpdateWorkerRequest) (*ateapipb.Worker, error) { return nil, status.Error(codes.Unimplemented, "UpdateWorker is not implemented yet") } -func (s *Service) DeleteWorker(ctx context.Context, req *ateapipb.DeleteWorkerRequest) (*ateapipb.Worker, error) { +func (s *RPCService) DeleteWorker(ctx context.Context, req *ateapipb.DeleteWorkerRequest) (*ateapipb.Worker, error) { return nil, status.Error(codes.Unimplemented, "DeleteWorker is not implemented yet") } -func (s *Service) DrainWorker(ctx context.Context, req *ateapipb.DrainWorkerRequest) (*ateapipb.Worker, error) { +func (s *RPCService) DrainWorker(ctx context.Context, req *ateapipb.DrainWorkerRequest) (*ateapipb.Worker, error) { return nil, status.Error(codes.Unimplemented, "DrainWorker is not implemented yet") } diff --git a/cmd/ateapi/internal/controlapi/worker_test.go b/cmd/ateapi/internal/controlapi/worker_test.go index 1871bcb68..3809837f7 100644 --- a/cmd/ateapi/internal/controlapi/worker_test.go +++ b/cmd/ateapi/internal/controlapi/worker_test.go @@ -54,7 +54,7 @@ func TestValidateListWorkersRequest(t *testing.T) { // will fail loudly as each one is filled in — at which point the corresponding // case moves to a real test. func TestWorkerAPIUnimplemented(t *testing.T) { - s := &Service{} + s := &RPCService{} ctx := context.Background() tests := []struct { diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go index 5d1d3e0d5..5e782f6e9 100644 --- a/cmd/ateapi/main.go +++ b/cmd/ateapi/main.go @@ -193,7 +193,7 @@ func main() { volPlugins := make(map[string]volume.VolumePluginControlPlane) ateletDialer := controlapi.NewAteletDialer(workerPodInformer.GetIndexer(), ateletPodInformer.GetIndexer(), *ateletClientCredBundle, *podIdentityCACerts) - sm := controlapi.NewService(persistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, storageClassLister, ateletDialer, instruments, *egressGatewayAddress, volPlugins) + controlSrv := controlapi.NewRPCService(persistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, storageClassLister, ateletDialer, instruments, *egressGatewayAddress, volPlugins) actorIdentitySrv := actoridentity.New(actorIdentityJWTIssuer, *actorIDJWTPoolFile, *actorIDCAPoolFile, persistence, workerCache) debugSrv := debugapi.NewService(persistence) @@ -228,7 +228,7 @@ func main() { ), ) reflection.Register(mux) - ateapipb.RegisterControlServer(mux, sm) + ateapipb.RegisterControlServer(mux, controlSrv) ateapipb.RegisterActorIdentityServer(mux, actorIdentitySrv) ateapipb.RegisterDebugServer(mux, debugSrv)