WatchList(..) endpoint for the resource service

agent/grpc-external/services/resource/server.go

@@ -4,7 +4,11 @@ import (
 	"context"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/storage"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
@@ -13,6 +17,18 @@ type Server struct {
 }
 
 type Config struct {
+	registry Registry
+	backend  Backend
+}
+
+//go:generate mockery --name Registry --inpackage
+type Registry interface {
+	resource.Registry
+}
+
+//go:generate mockery --name Backend --inpackage
+type Backend interface {
+	storage.Backend
 }
 
 func NewServer(cfg Config) *Server {
@@ -50,7 +66,13 @@ func (s *Server) Delete(ctx context.Context, req *pbresource.DeleteRequest) (*pb
 	return &pbresource.DeleteResponse{}, nil
 }
 
-func (s *Server) Watch(req *pbresource.WatchRequest, ws pbresource.ResourceService_WatchServer) error {
-	// TODO
-	return nil
+func (s *Server) resolveType(typ *pbresource.Type) (*resource.Registration, error) {
+	v, ok := s.registry.Resolve(typ)
+	if ok {
+		return &v, nil
+	}
+	return nil, status.Errorf(
+		codes.InvalidArgument,
+		"resource type %s not registered", resource.ToGVK(typ),
+	)
 }

agent/grpc-external/services/resource/server_test.go

@@ -65,11 +65,3 @@ func TestDelete_TODO(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, resp)
 }
-
-func TestWatch_TODO(t *testing.T) {
-	server := NewServer(Config{})
-	client := testClient(t, server)
-	wc, err := client.Watch(context.Background(), &pbresource.WatchRequest{})
-	require.NoError(t, err)
-	require.NotNil(t, wc)
-}

agent/grpc-external/services/resource/watch.go

@@ -0,0 +1,43 @@
+package resource
+
+import (
+	"github.com/hashicorp/consul/internal/storage"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func (s *Server) WatchList(req *pbresource.WatchListRequest, stream pbresource.ResourceService_WatchListServer) error {
+	// check type exists
+	if _, err := s.resolveType(req.Type); err != nil {
+		return err
+	}
+
+	unversionedType := storage.UnversionedTypeFrom(req.Type)
+	watch, err := s.backend.WatchList(
+		stream.Context(),
+		unversionedType,
+		req.Tenancy,
+		req.NamePrefix,
+	)
+	if err != nil {
+		return err
+	}
+
+	for {
+		event, err := watch.Next(stream.Context())
+		if err != nil {
+			return err
+		}
+
+		// drop versions that don't match
+		if event.Resource.Id.Type.GroupVersion != req.Type.GroupVersion {
+			continue
+		}
+
+		if err = stream.Send(&pbresource.WatchEvent{
+			Operation: event.Operation,
+			Resource:  event.Resource,
+		}); err != nil {
+			return err
+		}
+	}
+}

agent/grpc-external/services/resource/watch_test.go

@@ -0,0 +1,222 @@
+package resource
+
+import (
+	"context"
+	"errors"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/storage/inmem"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	"github.com/hashicorp/consul/proto/private/prototest"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestWatchList_TypeNotFound(t *testing.T) {
+	t.Parallel()
+	server := testServer(t)
+	client := testClient(t, server)
+
+	stream, err := client.WatchList(context.Background(), &pbresource.WatchListRequest{
+		Type:       typev1,
+		Tenancy:    tenancy,
+		NamePrefix: "",
+	})
+	require.NoError(t, err)
+	rspCh := handleResourceStream(t, stream)
+
+	err = mustGetError(t, rspCh)
+	require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+	require.Contains(t, err.Error(), "resource type mesh/v1/service not registered")
+}
+
+func TestWatchList_GroupVersionMatches(t *testing.T) {
+	t.Parallel()
+	server := testServer(t)
+	client := testClient(t, server)
+	server.registry.Register(resource.Registration{Type: typev1})
+	ctx := context.Background()
+
+	// create a watch
+	stream, err := client.WatchList(ctx, &pbresource.WatchListRequest{
+		Type:       typev1,
+		Tenancy:    tenancy,
+		NamePrefix: "",
+	})
+	require.NoError(t, err)
+	rspCh := handleResourceStream(t, stream)
+
+	// insert and verify upsert event received
+	r1, err := server.backend.WriteCAS(ctx, resourcev1)
+	require.NoError(t, err)
+	rsp := mustGetResource(t, rspCh)
+	require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, rsp.Operation)
+	prototest.AssertDeepEqual(t, r1, rsp.Resource)
+
+	// update and verify upsert event received
+	r2 := clone(r1)
+	r2, err = server.backend.WriteCAS(ctx, r2)
+	require.NoError(t, err)
+	rsp = mustGetResource(t, rspCh)
+	require.Equal(t, pbresource.WatchEvent_OPERATION_UPSERT, rsp.Operation)
+	prototest.AssertDeepEqual(t, r2, rsp.Resource)
+
+	// delete and verify delete event received
+	err = server.backend.DeleteCAS(ctx, r2.Id, r2.Version)
+	require.NoError(t, err)
+	rsp = mustGetResource(t, rspCh)
+	require.Equal(t, pbresource.WatchEvent_OPERATION_DELETE, rsp.Operation)
+}
+
+func TestWatchList_GroupVersionMismatch(t *testing.T) {
+	// Given a watch on typev2 that only differs from typev1 by GroupVersion
+	// When a resource of typev1 is created/updated/deleted
+	// Then no watch events should be emitted
+	t.Parallel()
+	server := testServer(t)
+	client := testClient(t, server)
+	ctx := context.Background()
+	server.registry.Register(resource.Registration{Type: typev1})
+	server.registry.Register(resource.Registration{Type: typev2})
+
+	// create a watch for typev2
+	stream, err := client.WatchList(ctx, &pbresource.WatchListRequest{
+		Type:       typev2,
+		Tenancy:    tenancy,
+		NamePrefix: "",
+	})
+	require.NoError(t, err)
+	rspCh := handleResourceStream(t, stream)
+
+	// insert
+	r1, err := server.backend.WriteCAS(ctx, resourcev1)
+	require.NoError(t, err)
+
+	// update
+	r2 := clone(r1)
+	r2, err = server.backend.WriteCAS(ctx, r2)
+	require.NoError(t, err)
+
+	// delete
+	err = server.backend.DeleteCAS(ctx, r2.Id, r2.Version)
+	require.NoError(t, err)
+
+	// verify no events received
+	mustGetNoResource(t, rspCh)
+}
+
+func testServer(t *testing.T) *Server {
+	t.Helper()
+
+	backend, err := inmem.NewBackend()
+	require.NoError(t, err)
+	go backend.Run(testContext(t))
+
+	registry := resource.NewRegistry()
+	return NewServer(Config{registry: registry, backend: backend})
+}
+
+func mustGetNoResource(t *testing.T, ch <-chan resourceOrError) {
+	t.Helper()
+
+	select {
+	case rsp := <-ch:
+		require.NoError(t, rsp.err)
+		require.Nil(t, rsp.rsp, "expected nil response with no error")
+	case <-time.After(250 * time.Millisecond):
+		return
+	}
+}
+
+func mustGetResource(t *testing.T, ch <-chan resourceOrError) *pbresource.WatchEvent {
+	t.Helper()
+
+	select {
+	case rsp := <-ch:
+		require.NoError(t, rsp.err)
+		return rsp.rsp
+	case <-time.After(1 * time.Second):
+		t.Fatal("timeout waiting for WatchListResponse")
+		return nil
+	}
+}
+
+func mustGetError(t *testing.T, ch <-chan resourceOrError) error {
+	t.Helper()
+
+	select {
+	case rsp := <-ch:
+		require.Error(t, rsp.err)
+		return rsp.err
+	case <-time.After(2 * time.Second):
+		t.Fatal("timeout waiting for WatchListResponse")
+		return nil
+	}
+}
+
+func handleResourceStream(t *testing.T, stream pbresource.ResourceService_WatchListClient) <-chan resourceOrError {
+	t.Helper()
+
+	rspCh := make(chan resourceOrError)
+	go func() {
+		for {
+			rsp, err := stream.Recv()
+			if errors.Is(err, io.EOF) ||
+				errors.Is(err, context.Canceled) ||
+				errors.Is(err, context.DeadlineExceeded) {
+				return
+			}
+			rspCh <- resourceOrError{
+				rsp: rsp,
+				err: err,
+			}
+		}
+	}()
+	return rspCh
+}
+
+var (
+	tenancy = &pbresource.Tenancy{
+		Partition: "default",
+		Namespace: "default",
+		PeerName:  "local",
+	}
+	typev1 = &pbresource.Type{
+		Group:        "mesh",
+		GroupVersion: "v1",
+		Kind:         "service",
+	}
+	typev2 = &pbresource.Type{
+		Group:        "mesh",
+		GroupVersion: "v2",
+		Kind:         "service",
+	}
+	resourcev1 = &pbresource.Resource{
+		Id: &pbresource.ID{
+			Uid:     "someUid",
+			Name:    "someName",
+			Type:    typev1,
+			Tenancy: tenancy,
+		},
+		Version: "",
+	}
+)
+
+type resourceOrError struct {
+	rsp *pbresource.WatchEvent
+	err error
+}
+
+func clone[T proto.Message](v T) T { return proto.Clone(v).(T) }
+
+func testContext(t *testing.T) context.Context {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	return ctx
+}

agent/grpc-middleware/rate_limit_mappings.gen.go

@@ -25,7 +25,7 @@ var rpcRateLimitSpecs = map[string]rate.OperationType{
 	"/hashicorp.consul.resource.ResourceService/Delete":                          rate.OperationTypeWrite,
 	"/hashicorp.consul.resource.ResourceService/List":                            rate.OperationTypeRead,
 	"/hashicorp.consul.resource.ResourceService/Read":                            rate.OperationTypeRead,
-	"/hashicorp.consul.resource.ResourceService/Watch":                           rate.OperationTypeRead,
+	"/hashicorp.consul.resource.ResourceService/WatchList":                       rate.OperationTypeRead,
 	"/hashicorp.consul.resource.ResourceService/Write":                           rate.OperationTypeWrite,
 	"/hashicorp.consul.resource.ResourceService/WriteStatus":                     rate.OperationTypeWrite,
 	"/hashicorp.consul.serverdiscovery.ServerDiscoveryService/WatchServers":      rate.OperationTypeRead,