local

pkg/aggregator/apiserver/dataplaneservice_controller.go

@@ -147,7 +147,7 @@ func (c *DataPlaneServiceRegistrationController) processNextWorkItem() bool {
 		return true
 	}
 
-	utilruntime.HandleError(fmt.Errorf("%v failed with : %v", key, err))
+	utilruntime.HandleError(fmt.Errorf("%v failed with: %v", key, err))
 	c.queue.AddRateLimited(key)
 
 	return true

pkg/storage/unified/resource/client.go

@@ -1,12 +1,13 @@
 package resource
 
 import (
+	"context"
+	"fmt"
+
 	"github.com/fullstorydev/grpchan"
-	"github.com/fullstorydev/grpchan/inprocgrpc"
-	"github.com/grafana/authlib/authn"
 	grpcUtils "github.com/grafana/grafana/pkg/storage/unified/resource/grpc"
-	grpcAuth "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/auth"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 )
 
 type ResourceClient interface {
@@ -31,17 +32,133 @@ func NewResourceClient(channel *grpc.ClientConn) ResourceClient {
 	}
 }
 
-func NewLocalResourceClient(server ResourceStoreServer) ResourceStoreClient {
+func NewLocalResourceClient(server ResourceServer) ResourceStoreClient {
-	channel := &inprocgrpc.Channel{}
+	return &localResourceClient{server}
-	auth := authn.LocalAuthenticator{}
+}
-	channel.RegisterService(
+
-		grpchan.InterceptServer(
+var _ ResourceClient = &localResourceClient{}
-			&ResourceStore_ServiceDesc,
+
-			grpcAuth.UnaryServerInterceptor(auth.Authenticate),
+type localResourceClient struct {
-			grpcAuth.StreamServerInterceptor(auth.Authenticate),
+	server ResourceServer
-		),
+}
-		server, // Implements all the things
+
-	)
+// Create implements ResourceClient.
-	cc := grpchan.InterceptClientConn(channel, grpcUtils.UnaryClientInterceptor, grpcUtils.StreamClientInterceptor)
+func (c *localResourceClient) Create(ctx context.Context, in *CreateRequest, opts ...grpc.CallOption) (*CreateResponse, error) {
-	return NewResourceStoreClient(cc)
+	return c.server.Create(ctx, in)
+}
+
+// Delete implements ResourceClient.
+func (c *localResourceClient) Delete(ctx context.Context, in *DeleteRequest, opts ...grpc.CallOption) (*DeleteResponse, error) {
+	return c.server.Delete(ctx, in)
+}
+
+// History implements ResourceClient.
+func (c *localResourceClient) History(ctx context.Context, in *HistoryRequest, opts ...grpc.CallOption) (*HistoryResponse, error) {
+	return c.server.History(ctx, in)
+}
+
+// IsHealthy implements ResourceClient.
+func (c *localResourceClient) IsHealthy(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error) {
+	return c.server.IsHealthy(ctx, in)
+}
+
+// List implements ResourceClient.
+func (c *localResourceClient) List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error) {
+	return c.server.List(ctx, in)
+}
+
+// Origin implements ResourceClient.
+func (c *localResourceClient) Origin(ctx context.Context, in *OriginRequest, opts ...grpc.CallOption) (*OriginResponse, error) {
+	return c.server.Origin(ctx, in)
+}
+
+// Read implements ResourceClient.
+func (c *localResourceClient) Read(ctx context.Context, in *ReadRequest, opts ...grpc.CallOption) (*ReadResponse, error) {
+	return c.server.Read(ctx, in)
+}
+
+// Update implements ResourceClient.
+func (c *localResourceClient) Update(ctx context.Context, in *UpdateRequest, opts ...grpc.CallOption) (*UpdateResponse, error) {
+	return c.server.Update(ctx, in)
+}
+
+// Watch implements ResourceClient.
+func (c *localResourceClient) Watch(ctx context.Context, in *WatchRequest, opts ...grpc.CallOption) (ResourceStore_WatchClient, error) {
+	stream := &localWatchStream{
+		ctx:    ctx,
+		events: make(chan *WatchEvent),
+	}
+	err := c.server.Watch(in, stream)
+	return stream, err
+}
+
+var _ ResourceStore_WatchClient = &localWatchStream{}
+var _ ResourceStore_WatchServer = &localWatchStream{}
+
+type localWatchStream struct {
+	ctx     context.Context
+	events  chan *WatchEvent
+	closed  bool
+	trailer metadata.MD
+}
+
+// Send implements ResourceStore_WatchServer.
+func (s *localWatchStream) Send(e *WatchEvent) error {
+	if s.closed {
+		return fmt.Errorf("stream is already closed")
+	}
+	s.events <- e
+	return nil // check if the channel is OK?
+}
+
+// Recv implements ResourceStore_WatchClient.
+func (s *localWatchStream) Recv() (*WatchEvent, error) {
+	e := <-s.events
+	return e, nil
+}
+
+// Context implements ResourceStore_WatchClient.
+func (s *localWatchStream) Context() context.Context {
+	return s.ctx
+}
+
+// CloseSend implements ResourceStore_WatchClient.
+func (s *localWatchStream) CloseSend() error {
+	s.closed = true
+	return nil
+}
+
+// SendHeader implements ResourceStore_WatchServer.
+func (s *localWatchStream) SendHeader(metadata.MD) error {
+	return fmt.Errorf("local stream")
+}
+
+// SetHeader implements ResourceStore_WatchServer.
+func (s *localWatchStream) SetHeader(metadata.MD) error {
+	return fmt.Errorf("local stream")
+}
+
+// SetTrailer implements ResourceStore_WatchServer.
+func (s *localWatchStream) SetTrailer(v metadata.MD) {
+	s.trailer = v
+}
+
+// Header implements ResourceStore_WatchClient.
+func (s *localWatchStream) Header() (metadata.MD, error) {
+	return metadata.MD{}, nil
+}
+
+// RecvMsg implements ResourceStore_WatchClient.
+func (s *localWatchStream) RecvMsg(m any) error {
+	return fmt.Errorf("local stream")
+}
+
+// SendMsg implements ResourceStore_WatchClient.
+func (s *localWatchStream) SendMsg(m any) error {
+	return fmt.Errorf("local stream")
+}
+
+// Trailer implements ResourceStore_WatchClient.
+func (s *localWatchStream) Trailer() metadata.MD {
+	return s.trailer
 }

pkg/storage/unified/resource/go.mod

@@ -19,7 +19,6 @@ require (
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/bufbuild/protocompile v0.4.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
@@ -27,7 +26,6 @@ require (
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.13.0 // indirect
 	github.com/jhump/protoreflect v1.15.1 // indirect

pkg/storage/unified/resource/server.go

@@ -306,7 +306,7 @@ func (s *server) Create(ctx context.Context, req *CreateRequest) (*CreateRespons
 	user, ok := claims.From(ctx)
 	if !ok || user == nil {
 		rsp.Error = &ErrorResult{
-			Message: "no user found in context",
+			Message: "no user found in context (create)",
 			Code:    http.StatusUnauthorized,
 		}
 		return rsp, nil
@@ -347,7 +347,7 @@ func (s *server) Update(ctx context.Context, req *UpdateRequest) (*UpdateRespons
 	user, ok := claims.From(ctx)
 	if !ok || user == nil {
 		rsp.Error = &ErrorResult{
-			Message: "no user found in context",
+			Message: "no user found in context (update)",
 			Code:    http.StatusUnauthorized,
 		}
 		return rsp, nil
@@ -396,6 +396,10 @@ func (s *server) Delete(ctx context.Context, req *DeleteRequest) (*DeleteRespons
 	if err := s.Init(ctx); err != nil {
 		return nil, err
 	}
+	requester, ok := claims.From(ctx)
+	if !ok {
+		return nil, apierrors.NewBadRequest("no user found in context (delete)")
+	}
 
 	rsp := &DeleteResponse{}
 	if req.ResourceVersion < 0 {
@@ -420,10 +424,6 @@ func (s *server) Delete(ctx context.Context, req *DeleteRequest) (*DeleteRespons
 		Type:       WatchEvent_DELETED,
 		PreviousRV: latest.ResourceVersion,
 	}
-	requester, ok := claims.From(ctx)
-	if !ok {
-		return nil, apierrors.NewBadRequest("unable to get user")
-	}
 	marker := &DeletedMarker{}
 	err := json.Unmarshal(latest.Value, marker)
 	if err != nil {