[UDT PR 7 / N] Implement routing for UDT (#8070)

# Description

This change adds support routing decisions based on UDT registrations.
The logic in UCP that deals with request routing will now check for
resource types and resource provider locations, with a fallback to the
previous logic for compatibility.

This is powered by a new configuration setting in UCP's configuration,
that will be configured to the URL of dynamic-rp.

Note that there's nothing in dynamic-rp yet to serve these requests.
This will be done in a follow-up.


## Type of change


- This pull request adds or changes features of Radius and has an
approved issue (issue link required).


Part of: #6688 

## Contributor checklist
Please verify that the PR meets the following requirements, where
applicable:

- [ ] An overview of proposed schema changes is included in a linked
GitHub issue.
- [ ] A design document PR is created in the [design-notes
repository](https://github.com/radius-project/design-notes/), if new
APIs are being introduced.
- [ ] If applicable, design document has been reviewed and approved by
Radius maintainers/approvers.
- [ ] A PR for the [samples
repository](https://github.com/radius-project/samples) is created, if
existing samples are affected by the changes in this PR.
- [ ] A PR for the [documentation
repository](https://github.com/radius-project/docs) is created, if the
changes in this PR affect the documentation or any user facing updates
are made.
- [ ] A PR for the [recipes
repository](https://github.com/radius-project/recipes) is created, if
existing recipes are affected by the changes in this PR.

Signed-off-by: Ryan Nowak <[email protected]>

cmd/ucpd/ucp-dev.yaml

@@ -54,6 +54,10 @@ ucp:
   direct:
     endpoint: "http://localhost:9000/apis/api.ucp.dev/v1alpha3"
 
+routing:
+  # This is the default downstream (dynamic-rp) for UDT implementations.
+  defaultDownstreamEndpoint: "http://localhost:8082"
+
 # Metrics configuration
 # port is not the same as metrics configuration in radius-self-hosted.yaml
 # so that we can run both services in debug mode.

deploy/Chart/templates/ucp/configmaps.yaml

@@ -50,6 +50,9 @@ data:
 
     ucp:
       kind: kubernetes
+    
+    routing:
+      defaultDownstreamEndpoint: "http://dynamic-rp.radius-sytem:8082"
 
     metricsProvider:
       prometheus:

pkg/armrpc/rest/results.go

@@ -537,6 +537,19 @@ func NewNotFoundResponse(id resources.ID) Response {
 	}
 }
 
+// NewNotFoundResponse creates a NotFoundResponse with resource id and an additional message.
+func NewNotFoundResponseWithCause(id resources.ID, cause string) Response {
+	return &NotFoundResponse{
+		Body: v1.ErrorResponse{
+			Error: v1.ErrorDetails{
+				Code:    v1.CodeNotFound,
+				Message: fmt.Sprintf("the resource with id '%s' was not found: %s", id.String(), cause),
+				Target:  id.String(),
+			},
+		},
+	}
+}
+
 // NewNotFoundAPIVersionResponse creates Response for unsupported api version. (message is consistent with ARM).
 func NewNotFoundAPIVersionResponse(resourceType string, namespace string, apiVersion string) Response {
 	return &NotFoundResponse{

pkg/cli/cmd/resourceprovider/create/create.go

@@ -33,7 +33,7 @@ import (
 	"github.com/spf13/cobra"
 )
 
-// NewCommand creates an instance of the `rad resourceprovider create` command and runner.
+// NewCommand creates an instance of the `rad resource-provider create` command and runner.
 func NewCommand(factory framework.Factory) (*cobra.Command, framework.Runner) {
 	runner := NewRunner(factory)
 
@@ -68,7 +68,7 @@ rad resource-provider create --from-file /path/to/input.json
 	return cmd, runner
 }
 
-// Runner is the Runner implementation for the `rad resourceprovider create` command.
+// Runner is the Runner implementation for the `rad resource-provider create` command.
 type Runner struct {
 	ConnectionFactory connections.Factory
 	ConfigHolder      *framework.ConfigHolder
@@ -80,7 +80,7 @@ type Runner struct {
 	ResourceProvider                 *manifest.ResourceProvider
 }
 
-// NewRunner creates an instance of the runner for the `rad resourceprovider create` command.
+// NewRunner creates an instance of the runner for the `rad resource-provider create` command.
 func NewRunner(factory framework.Factory) *Runner {
 	return &Runner{
 		ConnectionFactory: factory.GetConnectionFactory(),
@@ -89,7 +89,7 @@ func NewRunner(factory framework.Factory) *Runner {
 	}
 }
 
-// Validate runs validation for the `rad resourceprovider create` command.
+// Validate runs validation for the `rad resource-provider create` command.
 func (r *Runner) Validate(cmd *cobra.Command, args []string) error {
 	// Validate command line args and
 	workspace, err := cli.RequireWorkspace(cmd, r.ConfigHolder.Config, r.ConfigHolder.DirectoryConfig)
@@ -112,7 +112,7 @@ func (r *Runner) Validate(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
-// Run runs the `rad resourceprovider create` command.
+// Run runs the `rad resource-provider create` command.
 func (r *Runner) Run(ctx context.Context) error {
 	client, err := r.ConnectionFactory.CreateApplicationsManagementClient(ctx, *r.Workspace)
 	if err != nil {

pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"net/url"
 
 	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
 	ctrl "github.com/radius-project/radius/pkg/armrpc/asyncoperation/controller"
@@ -30,7 +31,6 @@ import (
 	"github.com/radius-project/radius/pkg/ucp/store"
 	"github.com/radius-project/radius/pkg/ucp/trackedresource"
 	"github.com/radius-project/radius/pkg/ucp/ucplog"
-	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
 var _ ctrl.Controller = (*TrackedResourceProcessController)(nil)
@@ -45,12 +45,18 @@ type TrackedResourceProcessController struct {
 
 	// Updater is the utility struct that can perform updates on tracked resources. This can be modified for testing.
 	updater updater
+
+	// defaultDownstream is the address of the dynamic resource provider to proxy requests to.
+	defaultDownstream *url.URL
 }
 
 // NewTrackedResourceProcessController creates a new TrackedResourceProcessController controller which is used to process resources asynchronously.
-func NewTrackedResourceProcessController(opts ctrl.Options) (ctrl.Controller, error) {
-	transport := otelhttp.NewTransport(http.DefaultTransport)
-	return &TrackedResourceProcessController{ctrl.NewBaseAsyncController(opts), trackedresource.NewUpdater(opts.StorageClient, &http.Client{Transport: transport})}, nil
+func NewTrackedResourceProcessController(opts ctrl.Options, transport http.RoundTripper, defaultDownstream *url.URL) (ctrl.Controller, error) {
+	return &TrackedResourceProcessController{
+		BaseController:    ctrl.NewBaseAsyncController(opts),
+		updater:           trackedresource.NewUpdater(opts.StorageClient, &http.Client{Transport: transport}),
+		defaultDownstream: defaultDownstream,
+	}, nil
 }
 
 // Run retrieves a resource from storage, parses the resource ID, and updates our tracked resource entry in the background.
@@ -67,7 +73,7 @@ func (c *TrackedResourceProcessController) Run(ctx context.Context, request *ctr
 		return ctrl.Result{}, err
 	}
 
-	downstreamURL, err := resourcegroups.ValidateDownstream(ctx, c.StorageClient(), originalID)
+	downstreamURL, err := resourcegroups.ValidateDownstream(ctx, c.StorageClient(), originalID, v1.LocationGlobal, resource.Properties.APIVersion)
 	if errors.Is(err, &resourcegroups.NotFoundError{}) {
 		return ctrl.NewFailedResult(v1.ErrorDetails{Code: v1.CodeNotFound, Message: err.Error(), Target: request.ResourceID}), nil
 	} else if errors.Is(err, &resourcegroups.InvalidError{}) {
@@ -76,8 +82,18 @@ func (c *TrackedResourceProcessController) Run(ctx context.Context, request *ctr
 		return ctrl.Result{}, fmt.Errorf("failed to validate downstream: %w", err)
 	}
 
+	if downstreamURL == nil {
+		downstreamURL = c.defaultDownstream
+	}
+
+	if downstreamURL == nil {
+		message := "No downstream address was configured for the resource provider, and no default downstream address was provided"
+		return ctrl.NewFailedResult(v1.ErrorDetails{Code: v1.CodeInvalid, Message: message, Target: resource.Properties.ID}), nil
+	}
+
 	logger := ucplog.FromContextOrDiscard(ctx)
 	logger.Info("Processing tracked resource", "resourceID", originalID)
+
 	err = c.updater.Update(ctx, downstreamURL.String(), originalID, resource.Properties.APIVersion)
 	if errors.Is(err, &trackedresource.InProgressErr{}) {
 		// The resource is still being processed, so we can sleep for a while.

pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess_test.go

@@ -37,7 +37,7 @@ func Test_Run(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		storageClient := store.NewMockStorageClient(ctrl)
 
-		pc, err := NewTrackedResourceProcessController(controller.Options{StorageClient: storageClient})
+		pc, err := NewTrackedResourceProcessController(controller.Options{StorageClient: storageClient}, nil, nil)
 		require.NoError(t, err)
 
 		updater := mockUpdater{}
@@ -48,6 +48,9 @@ func Test_Run(t *testing.T) {
 	id := resources.MustParse("/planes/test/local/resourceGroups/test-rg/providers/Applications.Test/testResources/my-resource")
 	trackingID := trackedresource.IDFor(id)
 
+	resourceTypeID, err := datamodel.ResourceTypeIDFromResourceID(id)
+	require.NoError(t, err)
+
 	plane := datamodel.RadiusPlane{
 		Properties: datamodel.RadiusPlaneProperties{
 			ResourceProviders: map[string]string{
@@ -70,6 +73,10 @@ func Test_Run(t *testing.T) {
 			Get(gomock.Any(), "/planes/"+trackingID.PlaneNamespace(), gomock.Any()).
 			Return(&store.Object{Data: plane}, nil).Times(1)
 
+		storageClient.EXPECT().
+			Get(gomock.Any(), resourceTypeID.String(), gomock.Any()).
+			Return(nil, &store.ErrNotFound{}).Times(1)
+
 		storageClient.EXPECT().
 			Get(gomock.Any(), trackingID.RootScope(), gomock.Any()).
 			Return(&store.Object{Data: resourceGroup}, nil).Times(1)
@@ -90,6 +97,10 @@ func Test_Run(t *testing.T) {
 			Get(gomock.Any(), "/planes/"+trackingID.PlaneNamespace(), gomock.Any()).
 			Return(&store.Object{Data: plane}, nil).Times(1)
 
+		storageClient.EXPECT().
+			Get(gomock.Any(), resourceTypeID.String(), gomock.Any()).
+			Return(nil, &store.ErrNotFound{}).Times(1)
+
 		storageClient.EXPECT().
 			Get(gomock.Any(), trackingID.RootScope(), gomock.Any()).
 			Return(&store.Object{Data: resourceGroup}, nil).Times(1)

pkg/ucp/backend/service.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
+	"net/url"
 
 	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
 	ctrl "github.com/radius-project/radius/pkg/armrpc/asyncoperation/controller"
@@ -30,6 +32,8 @@ import (
 	"github.com/radius-project/radius/pkg/ucp/backend/controller/resourcegroups"
 	"github.com/radius-project/radius/pkg/ucp/backend/controller/resourceproviders"
 	"github.com/radius-project/radius/pkg/ucp/datamodel"
+	ucpoptions "github.com/radius-project/radius/pkg/ucp/hostoptions"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
 const (
@@ -39,15 +43,18 @@ const (
 // Service is a service to run AsyncReqeustProcessWorker.
 type Service struct {
 	worker.Service
+
+	config ucpoptions.UCPConfig
 }
 
 // NewService creates new service instance to run AsyncRequestProcessWorker.
-func NewService(options hostoptions.HostOptions) *Service {
+func NewService(options hostoptions.HostOptions, config ucpoptions.UCPConfig) *Service {
 	return &Service{
-		worker.Service{
+		Service: worker.Service{
 			ProviderName: UCPProviderName,
 			Options:      options,
 		},
+		config: config,
 	}
 }
 
@@ -77,7 +84,13 @@ func (w *Service) Run(ctx context.Context) error {
 		DataProvider: w.StorageProvider,
 	}
 
-	err := RegisterControllers(ctx, w.Controllers, w.Options.UCPConnection, opts)
+	defaultDownstream, err := url.Parse(w.config.Routing.DefaultDownstreamEndpoint)
+	if err != nil {
+		return err
+	}
+
+	transport := otelhttp.NewTransport(http.DefaultTransport)
+	err = RegisterControllers(ctx, w.Controllers, w.Options.UCPConnection, transport, opts, defaultDownstream)
 	if err != nil {
 		return err
 	}
@@ -86,9 +99,11 @@ func (w *Service) Run(ctx context.Context) error {
 }
 
 // RegisterControllers registers the controllers for the UCP backend.
-func RegisterControllers(ctx context.Context, registry *worker.ControllerRegistry, connection sdk.Connection, opts ctrl.Options) error {
+func RegisterControllers(ctx context.Context, registry *worker.ControllerRegistry, connection sdk.Connection, transport http.RoundTripper, opts ctrl.Options, defaultDownstream *url.URL) error {
 	// Tracked resources
-	err := errors.Join(nil, registry.Register(ctx, v20231001preview.ResourceType, v1.OperationMethod(datamodel.OperationProcess), resourcegroups.NewTrackedResourceProcessController, opts))
+	err := errors.Join(nil, registry.Register(ctx, v20231001preview.ResourceType, v1.OperationMethod(datamodel.OperationProcess), func(opts ctrl.Options) (ctrl.Controller, error) {
+		return resourcegroups.NewTrackedResourceProcessController(opts, transport, defaultDownstream)
+	}, opts))
 
 	// Resource providers and related types
 	err = errors.Join(err, registry.Register(ctx, datamodel.ResourceProviderResourceType, v1.OperationPut, func(opts ctrl.Options) (ctrl.Controller, error) {

pkg/ucp/datamodel/location.go

@@ -21,6 +21,9 @@ import v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
 const (
 	// LocationResourceType is the resource type for a resource provider location
 	LocationResourceType = "System.Resources/resourceProviders/locations"
+
+	// LocationUnqualifiedResourceType is the unqualified resource type for a resource provider location.
+	LocationUnqualifiedResourceType = "locations"
 )
 
 // Location represents a location.

pkg/ucp/datamodel/resourceprovider_util.go

@@ -0,0 +1,72 @@
+/*
+Copyright 2023 The Radius Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package datamodel
+
+import (
+	"strings"
+
+	"github.com/radius-project/radius/pkg/ucp/resources"
+)
+
+// ResourceProviderIDFromResourceID converts an inbound resource id to the resource ID
+// of the resource provider.
+func ResourceProviderIDFromResourceID(id resources.ID) (resources.ID, error) {
+	// Ex:
+	// /planes/radius/local/providers/Applications.Test/testResources/foo
+	// => /planes/radius/local/providers/System.Resources/resourceProviders/Applications.Test
+	return resources.ParseResource(
+		id.PlaneScope() +
+			resources.SegmentSeparator + resources.ProvidersSegment +
+			resources.SegmentSeparator + ResourceProviderResourceType +
+			resources.SegmentSeparator + id.ProviderNamespace())
+}
+
+// ResourceTypeIDFromResourceID converts an inbound resource id to the resource ID
+// of the resource type.
+func ResourceTypeIDFromResourceID(id resources.ID) (resources.ID, error) {
+	// Ex:
+	// /planes/radius/local/providers/Applications.Test/testResources/foo
+	// => /planes/radius/local/providers/System.Resources/resourceProviders/Applications.Test/resourceTypes/testResources
+
+	// Ex: Applications.Test/testResources
+	fullyQualifiedResourceType := id.Type()
+
+	// Ex: testResources
+	unqualifiedResourceType := strings.TrimPrefix(fullyQualifiedResourceType, id.ProviderNamespace()+"/")
+
+	return resources.ParseResource(
+		id.PlaneScope() +
+			resources.SegmentSeparator + resources.ProvidersSegment +
+			resources.SegmentSeparator + ResourceProviderResourceType +
+			resources.SegmentSeparator + id.ProviderNamespace() +
+			resources.SegmentSeparator + ResourceTypeResourceUnqualifiedResourceType +
+			resources.SegmentSeparator + unqualifiedResourceType)
+}
+
+// ResourceProviderLocationIDFromResourceID converts an inbound resource id to the resource ID
+// of the resource provider's location.
+func ResourceProviderLocationIDFromResourceID(id resources.ID, location string) (resources.ID, error) {
+	// Ex:
+	// /planes/radius/local/providers/Applications.Test/testResources/foo + east
+	// => /planes/radius/local/providers/System.Resources/resourceProviders/Applications.Test/locations/east
+	base, err := ResourceProviderIDFromResourceID(id)
+	if err != nil {
+		return resources.ID{}, err
+	}
+
+	return base.Append(resources.TypeSegment{Type: LocationUnqualifiedResourceType, Name: location}), nil
+}

pkg/ucp/datamodel/resourceprovider_util_test.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2023 The Radius Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package datamodel
+
+import (
+	"testing"
+
+	"github.com/radius-project/radius/pkg/ucp/resources"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_ResourceProviderIDFromResourceID(t *testing.T) {
+	id := resources.MustParse("/planes/radius/local/resourceGroups/test-group/providers/Applications.Test/testResources/foo")
+
+	result, err := ResourceProviderIDFromResourceID(id)
+	require.NoError(t, err)
+
+	expected := resources.MustParse("/planes/radius/local/providers/System.Resources/resourceProviders/Applications.Test")
+	require.Equal(t, expected, result)
+}
+
+func Test_ResourceTypeIDFromResourceID(t *testing.T) {
+	id := resources.MustParse("/planes/radius/local/resourceGroups/test-group/providers/Applications.Test/testResources/foo")
+
+	result, err := ResourceTypeIDFromResourceID(id)
+	require.NoError(t, err)
+
+	expected := resources.MustParse("/planes/radius/local/providers/System.Resources/resourceProviders/Applications.Test/resourceTypes/testResources")
+	require.Equal(t, expected, result)
+}
+
+func Test_ResourceProviderLocationIDFromResourceID(t *testing.T) {
+	id := resources.MustParse("/planes/radius/local/resourceGroups/test-group/providers/Applications.Test/testResources/foo")
+
+	result, err := ResourceProviderLocationIDFromResourceID(id, "east")
+	require.NoError(t, err)
+
+	expected := resources.MustParse("/planes/radius/local/providers/System.Resources/resourceProviders/Applications.Test/locations/east")
+	require.Equal(t, expected, result)
+}