expose grpc as http endpoint (#18221)

expose resource grpc endpoints as http endpoints

agent/agent_endpoint_test.go

@@ -1601,14 +1601,35 @@ func TestAgent_Metrics_ACLDeny(t *testing.T) {
 	})
 }
 
+func newDefaultBaseDeps(t *testing.T) BaseDeps {
+	dataDir := testutil.TempDir(t, "acl-agent")
+	logBuffer := testutil.NewLogBuffer(t)
+	logger := hclog.NewInterceptLogger(nil)
+	loader := func(source config.Source) (config.LoadResult, error) {
+		dataDir := fmt.Sprintf(`data_dir = "%s"`, dataDir)
+		opts := config.LoadOpts{
+			HCL:           []string{TestConfigHCL(NodeID()), "", dataDir},
+			DefaultConfig: source,
+		}
+		result, err := config.Load(opts)
+		if result.RuntimeConfig != nil {
+			result.RuntimeConfig.Telemetry.Disable = true
+		}
+		return result, err
+	}
+	bd, err := NewBaseDeps(loader, logBuffer, logger)
+	require.NoError(t, err)
+	return bd
+}
+
 func TestHTTPHandlers_AgentMetricsStream_ACLDeny(t *testing.T) {
-	bd := BaseDeps{}
+	bd := newDefaultBaseDeps(t)
 	bd.Tokens = new(tokenStore.Store)
 	sink := metrics.NewInmemSink(30*time.Millisecond, time.Second)
 	bd.MetricsConfig = &lib.MetricsConfig{
 		Handler: sink,
 	}
-	d := fakeResolveTokenDelegate{authorizer: acl.DenyAll()}
+	d := fakeResolveTokenDelegate{delegate: &delegateMock{}, authorizer: acl.DenyAll()}
 	agent := &Agent{
 		baseDeps: bd,
 		delegate: d,
@@ -1631,13 +1652,13 @@ func TestHTTPHandlers_AgentMetricsStream_ACLDeny(t *testing.T) {
 }
 
 func TestHTTPHandlers_AgentMetricsStream(t *testing.T) {
-	bd := BaseDeps{}
+	bd := newDefaultBaseDeps(t)
 	bd.Tokens = new(tokenStore.Store)
 	sink := metrics.NewInmemSink(20*time.Millisecond, time.Second)
 	bd.MetricsConfig = &lib.MetricsConfig{
 		Handler: sink,
 	}
-	d := fakeResolveTokenDelegate{authorizer: acl.ManageAll()}
+	d := fakeResolveTokenDelegate{delegate: &delegateMock{}, authorizer: acl.ManageAll()}
 	agent := &Agent{
 		baseDeps: bd,
 		delegate: d,

agent/agent_test.go

@@ -57,6 +57,7 @@ import (
 	"github.com/hashicorp/consul/agent/token"
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/consul/internal/go-sso/oidcauth/oidcauthtest"
+	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/ipaddr"
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/consul/proto/private/pbautoconf"
@@ -322,6 +323,7 @@ func TestAgent_HTTPMaxHeaderBytes(t *testing.T) {
 					Tokens:          new(token.Store),
 					TLSConfigurator: tlsConf,
 					GRPCConnPool:    &fakeGRPCConnPool{},
+					Registry:        resource.NewRegistry(),
 				},
 				RuntimeConfig: &config.RuntimeConfig{
 					HTTPAddrs: []net.Addr{
@@ -344,6 +346,7 @@ func TestAgent_HTTPMaxHeaderBytes(t *testing.T) {
 			require.NoError(t, err)
 
 			a, err := New(bd)
+			a.delegate = &delegateMock{}
 			require.NoError(t, err)
 
 			a.startLicenseManager(testutil.TestContext(t))
@@ -5477,6 +5480,7 @@ func TestAgent_ListenHTTP_MultipleAddresses(t *testing.T) {
 			Tokens:          new(token.Store),
 			TLSConfigurator: tlsConf,
 			GRPCConnPool:    &fakeGRPCConnPool{},
+			Registry:        resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{
 			HTTPAddrs: []net.Addr{
@@ -5499,6 +5503,7 @@ func TestAgent_ListenHTTP_MultipleAddresses(t *testing.T) {
 	require.NoError(t, err)
 
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	agent.startLicenseManager(testutil.TestContext(t))
@@ -6073,6 +6078,7 @@ func TestAgent_startListeners(t *testing.T) {
 			Logger:       hclog.NewInterceptLogger(nil),
 			Tokens:       new(token.Store),
 			GRPCConnPool: &fakeGRPCConnPool{},
+			Registry:     resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{
 			HTTPAddrs: []net.Addr{},
@@ -6091,6 +6097,7 @@ func TestAgent_startListeners(t *testing.T) {
 	require.NoError(t, err)
 
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	// use up an address
@@ -6213,6 +6220,7 @@ func TestAgent_startListeners_scada(t *testing.T) {
 			HCP: hcp.Deps{
 				Provider: pvd,
 			},
+			Registry: resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{},
 		Cache:         cache.New(cache.Options{}),
@@ -6230,6 +6238,7 @@ func TestAgent_startListeners_scada(t *testing.T) {
 	require.NoError(t, err)
 
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	_, err = agent.startListeners([]net.Addr{c})
@@ -6273,6 +6282,7 @@ func TestAgent_checkServerLastSeen(t *testing.T) {
 			Logger:       hclog.NewInterceptLogger(nil),
 			Tokens:       new(token.Store),
 			GRPCConnPool: &fakeGRPCConnPool{},
+			Registry:     resource.NewRegistry(),
 		},
 		RuntimeConfig: &config.RuntimeConfig{},
 		Cache:         cache.New(cache.Options{}),
@@ -6284,6 +6294,7 @@ func TestAgent_checkServerLastSeen(t *testing.T) {
 		Config:      leafcert.Config{},
 	})
 	agent, err := New(bd)
+	agent.delegate = &delegateMock{}
 	require.NoError(t, err)
 
 	// Test that an ErrNotExist OS error is treated as ok.

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

@@ -8,18 +8,50 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"github.com/hashicorp/go-uuid"
+
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/acl/resolver"
 	svc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
 	internal "github.com/hashicorp/consul/agent/grpc-internal"
+	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/storage/inmem"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
+func randomACLIdentity(t *testing.T) structs.ACLIdentity {
+	id, err := uuid.GenerateUUID()
+	require.NoError(t, err)
+
+	return &structs.ACLToken{AccessorID: id}
+}
+
+func AuthorizerFrom(t *testing.T, policyStrs ...string) resolver.Result {
+	policies := []*acl.Policy{}
+	for _, policyStr := range policyStrs {
+		policy, err := acl.NewPolicyFromSource(policyStr, nil, nil)
+		require.NoError(t, err)
+		policies = append(policies, policy)
+	}
+
+	authz, err := acl.NewPolicyAuthorizerWithDefaults(acl.DenyAll(), policies, nil)
+	require.NoError(t, err)
+
+	return resolver.Result{
+		Authorizer:  authz,
+		ACLIdentity: randomACLIdentity(t),
+	}
+}
+
 // RunResourceService runs a Resource Service for the duration of the test and
 // returns a client to interact with it. ACLs will be disabled.
 func RunResourceService(t *testing.T, registerFns ...func(resource.Registry)) pbresource.ResourceServiceClient {
+	return RunResourceServiceWithACL(t, resolver.DANGER_NO_AUTH{}, registerFns...)
+}
+
+func RunResourceServiceWithACL(t *testing.T, aclResolver svc.ACLResolver, registerFns ...func(resource.Registry)) pbresource.ResourceServiceClient {
 	t.Helper()
 
 	backend, err := inmem.NewBackend()
@@ -40,7 +72,7 @@ func RunResourceService(t *testing.T, registerFns ...func(resource.Registry)) pb
 		Backend:     backend,
 		Registry:    registry,
 		Logger:      testutil.Logger(t),
-		ACLResolver: resolver.DANGER_NO_AUTH{},
+		ACLResolver: aclResolver,
 	}).Register(server)
 
 	pipe := internal.NewPipeListener()

agent/http.go

@@ -36,6 +36,7 @@ import (
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/agent/uiserver"
 	"github.com/hashicorp/consul/api"
+	resourcehttp "github.com/hashicorp/consul/internal/resource/http"
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/consul/logging"
 	"github.com/hashicorp/consul/proto/private/pbcommon"
@@ -259,6 +260,17 @@ func (s *HTTPHandlers) handler() http.Handler {
 	handlePProf("/debug/pprof/symbol", pprof.Symbol)
 	handlePProf("/debug/pprof/trace", pprof.Trace)
 
+	mux.Handle("/api/",
+		http.StripPrefix("/api",
+			resourcehttp.NewHandler(
+				s.agent.delegate.ResourceServiceClient(),
+				s.agent.baseDeps.Registry,
+				s.parseToken,
+				s.agent.logger.Named(logging.HTTP),
+			),
+		),
+	)
+
 	if s.IsUIEnabled() {
 		// Note that we _don't_ support reloading ui_config.{enabled, content_dir,
 		// content_path} since this only runs at initial startup.

internal/resource/http/http.go

@@ -0,0 +1,170 @@
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"path"
+	"strings"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/hashicorp/go-hclog"
+
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func NewHandler(
+	client pbresource.ResourceServiceClient,
+	registry resource.Registry,
+	parseToken func(req *http.Request, token *string),
+	logger hclog.Logger) http.Handler {
+	mux := http.NewServeMux()
+	for _, t := range registry.Types() {
+		// Individual Resource Endpoints.
+		prefix := strings.ToLower(fmt.Sprintf("/%s/%s/%s/", t.Type.Group, t.Type.GroupVersion, t.Type.Kind))
+		logger.Info("Registered resource endpoint", "endpoint", prefix)
+		mux.Handle(prefix, http.StripPrefix(prefix, &resourceHandler{t, client, parseToken, logger}))
+	}
+
+	return mux
+}
+
+type writeRequest struct {
+	Metadata map[string]string `json:"metadata"`
+	Data     json.RawMessage   `json:"data"`
+	Owner    *pbresource.ID    `json:"owner"`
+}
+
+type resourceHandler struct {
+	reg        resource.Registration
+	client     pbresource.ResourceServiceClient
+	parseToken func(req *http.Request, token *string)
+	logger     hclog.Logger
+}
+
+func (h *resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	var token string
+	h.parseToken(r, &token)
+	ctx := metadata.AppendToOutgoingContext(r.Context(), "x-consul-token", token)
+	switch r.Method {
+	case http.MethodPut:
+		h.handleWrite(w, r, ctx)
+	default:
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return
+	}
+}
+
+func (h *resourceHandler) handleWrite(w http.ResponseWriter, r *http.Request, ctx context.Context) {
+	var req writeRequest
+	// convert req body to writeRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte("Request body didn't follow schema."))
+	}
+	// convert data struct to proto message
+	data := h.reg.Proto.ProtoReflect().New().Interface()
+	if err := protojson.Unmarshal(req.Data, data); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte("Request body didn't follow schema."))
+	}
+	// proto message to any
+	anyProtoMsg, err := anypb.New(data)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to convert proto message to any type", "error", err)
+		return
+	}
+
+	tenancyInfo, resourceName, version := checkURL(r)
+
+	rsp, err := h.client.Write(ctx, &pbresource.WriteRequest{
+		Resource: &pbresource.Resource{
+			Id: &pbresource.ID{
+				Type:    h.reg.Type,
+				Tenancy: tenancyInfo,
+				Name:    resourceName,
+			},
+			Owner:    req.Owner,
+			Version:  version,
+			Metadata: req.Metadata,
+			Data:     anyProtoMsg,
+		},
+	})
+	if err != nil {
+		handleResponseError(err, w, h)
+		return
+	}
+
+	output, err := jsonMarshal(rsp.Resource)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to unmarshal GRPC resource response", "error", err)
+		return
+	}
+	w.Write(output)
+}
+
+func checkURL(r *http.Request) (tenancy *pbresource.Tenancy, resourceName string, version string) {
+	params := r.URL.Query()
+	tenancy = &pbresource.Tenancy{
+		Partition: params.Get("partition"),
+		PeerName:  params.Get("peer_name"),
+		Namespace: params.Get("namespace"),
+	}
+	resourceName = path.Base(r.URL.Path)
+	if resourceName == "." || resourceName == "/" {
+		resourceName = ""
+	}
+	version = params.Get("version")
+
+	return
+}
+
+func jsonMarshal(res *pbresource.Resource) ([]byte, error) {
+	output, err := protojson.Marshal(res)
+	if err != nil {
+		return nil, err
+	}
+
+	var stuff map[string]any
+	if err := json.Unmarshal(output, &stuff); err != nil {
+		return nil, err
+	}
+
+	delete(stuff["data"].(map[string]any), "@type")
+	return json.MarshalIndent(stuff, "", "  ")
+}
+
+func handleResponseError(err error, w http.ResponseWriter, h *resourceHandler) {
+	if e, ok := status.FromError(err); ok {
+		switch e.Code() {
+		case codes.InvalidArgument:
+			w.WriteHeader(http.StatusBadRequest)
+			h.logger.Info("User has mal-formed request", "error", err)
+		case codes.NotFound:
+			w.WriteHeader(http.StatusNotFound)
+			h.logger.Info("Failed to write to GRPC resource: Not found", "error", err)
+		case codes.PermissionDenied:
+			w.WriteHeader(http.StatusForbidden)
+			h.logger.Info("Failed to write to GRPC resource: User not authenticated", "error", err)
+		case codes.Aborted:
+			w.WriteHeader(http.StatusConflict)
+			h.logger.Info("Failed to write to GRPC resource: the request conflict with the current state of the target resource", "error", err)
+		default:
+			w.WriteHeader(http.StatusInternalServerError)
+			h.logger.Error("Failed to write to GRPC resource", "error", err)
+		}
+	} else {
+		w.WriteHeader(http.StatusInternalServerError)
+		h.logger.Error("Failed to write to GRPC resource: not able to parse error returned", "error", err)
+	}
+	w.Write([]byte(err.Error()))
+}

internal/resource/http/http_test.go

@@ -0,0 +1,299 @@
+package http
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/hashicorp/go-hclog"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	resourceSvc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
+	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
+	pbdemov1 "github.com/hashicorp/consul/proto/private/pbdemo/v1"
+
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/resource/demo"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	pbdemov2 "github.com/hashicorp/consul/proto/private/pbdemo/v2"
+	"github.com/hashicorp/consul/sdk/testutil"
+)
+
+const testACLTokenArtistReadPolicy = "00000000-0000-0000-0000-000000000001"
+const testACLTokenArtistWritePolicy = "00000000-0000-0000-0000-000000000002"
+
+func parseToken(req *http.Request, token *string) {
+	*token = req.Header.Get("X-Consul-Token")
+}
+
+func TestResourceHandler_InputValidation(t *testing.T) {
+	type testCase struct {
+		description          string
+		request              *http.Request
+		response             *httptest.ResponseRecorder
+		expectedResponseCode int
+	}
+	client := svctest.RunResourceService(t, demo.RegisterTypes)
+	resourceHandler := resourceHandler{
+		resource.Registration{
+			Type:  demo.TypeV2Artist,
+			Proto: &pbdemov2.Artist{},
+		},
+		client,
+		func(req *http.Request, token *string) { return },
+		hclog.NewNullLogger(),
+	}
+
+	testCases := []testCase{
+		{
+			description: "missing resource name",
+			request: httptest.NewRequest("PUT", "/?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+				{
+					"metadata": {
+						"foo": "bar"
+					},
+					"data": {
+						"name": "Keith Urban",
+						"genre": "GENRE_COUNTRY"
+					}
+				}
+			`)),
+			response:             httptest.NewRecorder(),
+			expectedResponseCode: http.StatusBadRequest,
+		},
+		{
+			description: "wrong schema",
+			request: httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+				{
+					"metadata": {
+						"foo": "bar"
+					},
+					"dada": {
+						"name": "Keith Urban",
+						"genre": "GENRE_COUNTRY"
+					}
+				}
+			`)),
+			response:             httptest.NewRecorder(),
+			expectedResponseCode: http.StatusBadRequest,
+		},
+		{
+			description: "missing tenancy info",
+			request: httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local", strings.NewReader(`
+				{
+					"metadata": {
+						"foo": "bar"
+					},
+					"data": {
+						"name": "Keith Urban",
+						"genre": "GENRE_COUNTRY"
+					}
+				}
+			`)),
+			response:             httptest.NewRecorder(),
+			expectedResponseCode: http.StatusBadRequest,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.description, func(t *testing.T) {
+			resourceHandler.ServeHTTP(tc.response, tc.request)
+
+			require.Equal(t, tc.expectedResponseCode, tc.response.Result().StatusCode)
+		})
+	}
+}
+
+func TestResourceWriteHandler(t *testing.T) {
+	aclResolver := &resourceSvc.MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistReadPolicy, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV1ReadPolicy, demo.ArtistV2ReadPolicy), nil)
+	aclResolver.On("ResolveTokenAndDefaultMeta", testACLTokenArtistWritePolicy, mock.Anything, mock.Anything).
+		Return(svctest.AuthorizerFrom(t, demo.ArtistV1WritePolicy, demo.ArtistV2WritePolicy), nil)
+
+	client := svctest.RunResourceServiceWithACL(t, aclResolver, demo.RegisterTypes)
+
+	v1ArtistHandler := resourceHandler{
+		resource.Registration{
+			Type:  demo.TypeV1Artist,
+			Proto: &pbdemov1.Artist{},
+		},
+		client,
+		parseToken,
+		hclog.NewNullLogger(),
+	}
+
+	v2ArtistHandler := resourceHandler{
+		resource.Registration{
+			Type:  demo.TypeV2Artist,
+			Proto: &pbdemov2.Artist{},
+		},
+		client,
+		parseToken,
+		hclog.NewNullLogger(),
+	}
+
+	t.Run("should be blocked if the token is not authorized", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistReadPolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusForbidden, rsp.Result().StatusCode)
+	})
+
+	t.Run("should write to the resource backend", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban", result["data"].(map[string]any)["name"])
+		require.Equal(t, "keith-urban", result["id"].(map[string]any)["name"])
+
+		readRsp, err := client.Read(testutil.TestContext(t), &pbresource.ReadRequest{
+			Id: &pbresource.ID{
+				Type:    demo.TypeV2Artist,
+				Tenancy: demo.TenancyDefault,
+				Name:    "keith-urban",
+			},
+		})
+		require.NoError(t, err)
+		require.NotNil(t, readRsp.Resource)
+
+		var artist pbdemov2.Artist
+		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
+		require.Equal(t, "Keith Urban", artist.Name)
+	})
+
+	t.Run("should update the record with version parameter", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&version=1", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban Two",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban Two", result["data"].(map[string]any)["name"])
+		require.Equal(t, "keith-urban", result["id"].(map[string]any)["name"])
+	})
+
+	t.Run("should fail the update if the resource's version doesn't match the version of the existing resource", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v2/artist/keith-urban?partition=default&peer_name=local&namespace=default&version=1", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban",
+					"genre": "GENRE_COUNTRY"
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+
+		v2ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusConflict, rsp.Result().StatusCode)
+	})
+
+	t.Run("should write to the resource backend with owner", func(t *testing.T) {
+		rsp := httptest.NewRecorder()
+		req := httptest.NewRequest("PUT", "/demo/v1/artist/keith-urban-v1?partition=default&peer_name=local&namespace=default", strings.NewReader(`
+			{
+				"metadata": {
+					"foo": "bar"
+				},
+				"data": {
+					"name": "Keith Urban V1",
+					"genre": "GENRE_COUNTRY"
+				},
+				"owner": {
+					"name": "keith-urban",
+					"type": {
+						"group": "demo",
+						"group_version": "v2",
+						"kind": "Artist"
+					},
+					"tenancy": {
+						"partition": "default",
+						"peer_name": "local",
+						"namespace": "default"
+					}
+				}
+			}
+		`))
+
+		req.Header.Add("x-consul-token", testACLTokenArtistWritePolicy)
+
+		v1ArtistHandler.ServeHTTP(rsp, req)
+
+		require.Equal(t, http.StatusOK, rsp.Result().StatusCode)
+
+		var result map[string]any
+		require.NoError(t, json.NewDecoder(rsp.Body).Decode(&result))
+		require.Equal(t, "Keith Urban V1", result["data"].(map[string]any)["name"])
+		require.Equal(t, "keith-urban-v1", result["id"].(map[string]any)["name"])
+
+		readRsp, err := client.Read(testutil.TestContext(t), &pbresource.ReadRequest{
+			Id: &pbresource.ID{
+				Type:    demo.TypeV1Artist,
+				Tenancy: demo.TenancyDefault,
+				Name:    "keith-urban-v1",
+			},
+		})
+		require.NoError(t, err)
+		require.NotNil(t, readRsp.Resource)
+		require.Equal(t, "keith-urban", readRsp.Resource.Owner.Name)
+
+		var artist pbdemov1.Artist
+		require.NoError(t, readRsp.Resource.Data.UnmarshalTo(&artist))
+		require.Equal(t, "Keith Urban V1", artist.Name)
+	})
+}

internal/resource/registry.go

@@ -54,6 +54,8 @@ type Registry interface {
 
 	// Resolve the given resource type and its hooks.
 	Resolve(typ *pbresource.Type) (reg Registration, ok bool)
+
+	Types() []Registration
 }
 
 type Registration struct {
@@ -183,6 +185,17 @@ func (r *TypeRegistry) Resolve(typ *pbresource.Type) (reg Registration, ok bool)
 	return Registration{}, false
 }
 
+func (r *TypeRegistry) Types() []Registration {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
+	types := make([]Registration, 0, len(r.registrations))
+	for _, v := range r.registrations {
+		types = append(types, v)
+	}
+	return types
+}
+
 func ToGVK(resourceType *pbresource.Type) string {
 	return fmt.Sprintf("%s.%s.%s", resourceType.Group, resourceType.GroupVersion, resourceType.Kind)
 }