From f118751e4d719971217b55d1292f495b273aa78f Mon Sep 17 00:00:00 2001
From: Ryan Nowak <nowakra@gmail.com>
Date: Tue, 3 Dec 2024 17:09:30 -0800
Subject: [PATCH] [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 <nowakra@gmail.com>
---
 cmd/ucpd/ucp-dev.yaml                         |   4 +
 deploy/Chart/templates/ucp/configmaps.yaml    |   3 +
 pkg/armrpc/rest/results.go                    |  13 +
 pkg/cli/cmd/resourceprovider/create/create.go |  10 +-
 .../resourcegroups/trackedresourceprocess.go  |  26 +-
 .../trackedresourceprocess_test.go            |  13 +-
 pkg/ucp/backend/service.go                    |  25 +-
 pkg/ucp/datamodel/location.go                 |   3 +
 pkg/ucp/datamodel/resourceprovider_util.go    |  72 +++
 .../datamodel/resourceprovider_util_test.go   |  54 +++
 pkg/ucp/datamodel/resourcetype.go             |   3 +
 pkg/ucp/frontend/controller/radius/proxy.go   |  53 ++-
 .../frontend/controller/radius/proxy_test.go  |  52 ++-
 .../controller/resourcegroups/util.go         | 197 +++++++-
 .../controller/resourcegroups/util_test.go    | 442 +++++++++++++++++-
 pkg/ucp/frontend/radius/module.go             |   7 +-
 pkg/ucp/frontend/radius/routes.go             |  48 +-
 pkg/ucp/hostoptions/providerconfig.go         |   8 +
 pkg/ucp/integrationtests/radius/proxy_test.go |   4 +-
 .../integrationtests/testserver/testserver.go |  29 +-
 pkg/ucp/server/server.go                      |   2 +-
 21 files changed, 975 insertions(+), 93 deletions(-)
 create mode 100644 pkg/ucp/datamodel/resourceprovider_util.go
 create mode 100644 pkg/ucp/datamodel/resourceprovider_util_test.go

diff --git a/cmd/ucpd/ucp-dev.yaml b/cmd/ucpd/ucp-dev.yaml
index 4d26937dfe..32f1a465cf 100644
--- a/cmd/ucpd/ucp-dev.yaml
+++ b/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.
diff --git a/deploy/Chart/templates/ucp/configmaps.yaml b/deploy/Chart/templates/ucp/configmaps.yaml
index 18fe1d60e1..d56d0f4f30 100644
--- a/deploy/Chart/templates/ucp/configmaps.yaml
+++ b/deploy/Chart/templates/ucp/configmaps.yaml
@@ -50,6 +50,9 @@ data:
 
     ucp:
       kind: kubernetes
+    
+    routing:
+      defaultDownstreamEndpoint: "http://dynamic-rp.radius-sytem:8082"
 
     metricsProvider:
       prometheus:
diff --git a/pkg/armrpc/rest/results.go b/pkg/armrpc/rest/results.go
index 20176b73cd..d60993136d 100644
--- a/pkg/armrpc/rest/results.go
+++ b/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{
diff --git a/pkg/cli/cmd/resourceprovider/create/create.go b/pkg/cli/cmd/resourceprovider/create/create.go
index d34c282779..b5c55b775f 100644
--- a/pkg/cli/cmd/resourceprovider/create/create.go
+++ b/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 {
diff --git a/pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess.go b/pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess.go
index 5623d88518..bb2fa2b3d0 100644
--- a/pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess.go
+++ b/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.
diff --git a/pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess_test.go b/pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess_test.go
index 1958ace976..c4db345daf 100644
--- a/pkg/ucp/backend/controller/resourcegroups/trackedresourceprocess_test.go
+++ b/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)
diff --git a/pkg/ucp/backend/service.go b/pkg/ucp/backend/service.go
index b325a3bdcf..fc6ecc1270 100644
--- a/pkg/ucp/backend/service.go
+++ b/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) {
diff --git a/pkg/ucp/datamodel/location.go b/pkg/ucp/datamodel/location.go
index f4ea49999f..719623c8d0 100644
--- a/pkg/ucp/datamodel/location.go
+++ b/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.
diff --git a/pkg/ucp/datamodel/resourceprovider_util.go b/pkg/ucp/datamodel/resourceprovider_util.go
new file mode 100644
index 0000000000..9885627430
--- /dev/null
+++ b/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
+}
diff --git a/pkg/ucp/datamodel/resourceprovider_util_test.go b/pkg/ucp/datamodel/resourceprovider_util_test.go
new file mode 100644
index 0000000000..866a7b1c22
--- /dev/null
+++ b/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)
+}
diff --git a/pkg/ucp/datamodel/resourcetype.go b/pkg/ucp/datamodel/resourcetype.go
index bc1631e2fc..38dfda47e0 100644
--- a/pkg/ucp/datamodel/resourcetype.go
+++ b/pkg/ucp/datamodel/resourcetype.go
@@ -21,6 +21,9 @@ import v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
 const (
 	// ResourceTypesResourceType is the resource type for a resource type.
 	ResourceTypeResourceType = "System.Resources/resourceProviders/resourceTypes"
+
+	// ResourceTypeResourceUnqualifiedResourceType is the unqualified resource type for a resource type.
+	ResourceTypeResourceUnqualifiedResourceType = "resourceTypes"
 )
 
 // ResourceType represents a resource type.
diff --git a/pkg/ucp/frontend/controller/radius/proxy.go b/pkg/ucp/frontend/controller/radius/proxy.go
index 8f2bbdfd3a..84cc7360a2 100644
--- a/pkg/ucp/frontend/controller/radius/proxy.go
+++ b/pkg/ucp/frontend/controller/radius/proxy.go
@@ -25,6 +25,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/go-chi/chi/v5"
 	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
 	"github.com/radius-project/radius/pkg/armrpc/asyncoperation/statusmanager"
 	armrpc_controller "github.com/radius-project/radius/pkg/armrpc/frontend/controller"
@@ -37,7 +38,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"
 )
 
 const (
@@ -55,7 +55,7 @@ const (
 )
 
 type updater interface {
-	Update(ctx context.Context, downstream string, id resources.ID, version string) error
+	Update(ctx context.Context, downstreamURL string, originalID resources.ID, version string) error
 }
 
 var _ armrpc_controller.Controller = (*ProxyController)(nil)
@@ -64,6 +64,9 @@ var _ armrpc_controller.Controller = (*ProxyController)(nil)
 type ProxyController struct {
 	armrpc_controller.Operation[*datamodel.RadiusPlane, datamodel.RadiusPlane]
 
+	// defaultDownstream is the address of the dynamic resource provider to proxy requests to.
+	defaultDownstream *url.URL
+
 	// transport is the http.RoundTripper to use for proxying requests. Can be overridden for testing.
 	transport http.RoundTripper
 
@@ -71,17 +74,20 @@ type ProxyController struct {
 	updater updater
 }
 
-// # Function Explanation
-//
 // NewProxyController creates a new ProxyPlane controller with the given options and returns it, or returns an error if the
 // controller cannot be created.
-func NewProxyController(opts armrpc_controller.Options) (armrpc_controller.Controller, error) {
-	transport := otelhttp.NewTransport(http.DefaultTransport)
+func NewProxyController(opts armrpc_controller.Options, transport http.RoundTripper, defaultDownstream string) (armrpc_controller.Controller, error) {
+	parsedDefaultDownstream, err := url.Parse(defaultDownstream)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse default downstream URL: %w", err)
+	}
+
 	updater := trackedresource.NewUpdater(opts.StorageClient, &http.Client{Transport: transport})
 	return &ProxyController{
-		Operation: armrpc_controller.NewOperation(opts, armrpc_controller.ResourceOptions[datamodel.RadiusPlane]{}),
-		transport: transport,
-		updater:   updater,
+		Operation:         armrpc_controller.NewOperation(opts, armrpc_controller.ResourceOptions[datamodel.RadiusPlane]{}),
+		transport:         transport,
+		defaultDownstream: parsedDefaultDownstream,
+		updater:           updater,
 	}, nil
 }
 
@@ -100,9 +106,16 @@ func (p *ProxyController) Run(ctx context.Context, w http.ResponseWriter, req *h
 	id := requestCtx.ResourceID
 	relativePath := middleware.GetRelativePath(p.Options().PathBase, requestCtx.OriginalURL.Path)
 
-	downstreamURL, err := resourcegroups.ValidateDownstream(ctx, p.StorageClient(), id)
+	apiVersion := requestCtx.APIVersion
+	if apiVersion == "" {
+		message := "the api-version query parameter is required"
+		response := v1.ErrorResponse{Error: v1.ErrorDetails{Code: v1.CodeInvalid, Message: message, Target: id.String()}}
+		return armrpc_rest.NewBadRequestARMResponse(response), nil
+	}
+
+	downstreamURL, err := resourcegroups.ValidateDownstream(ctx, p.StorageClient(), id, v1.LocationGlobal, apiVersion)
 	if errors.Is(err, &resourcegroups.NotFoundError{}) {
-		return armrpc_rest.NewNotFoundResponse(id), nil
+		return armrpc_rest.NewNotFoundResponseWithCause(id, err.Error()), nil
 	} else if errors.Is(err, &resourcegroups.InvalidError{}) {
 		response := v1.ErrorResponse{Error: v1.ErrorDetails{Code: v1.CodeInvalid, Message: err.Error(), Target: id.String()}}
 		return armrpc_rest.NewBadRequestARMResponse(response), nil
@@ -110,13 +123,22 @@ func (p *ProxyController) Run(ctx context.Context, w http.ResponseWriter, req *h
 		return nil, fmt.Errorf("failed to validate downstream: %w", err)
 	}
 
+	if downstreamURL == nil {
+		downstreamURL = p.defaultDownstream
+	}
+
+	if downstreamURL == nil {
+		message := "No downstream address was configured for the resource provider, and no default downstream address was provided"
+		response := v1.ErrorResponse{Error: v1.ErrorDetails{Code: v1.CodeInvalid, Message: message, Target: id.String()}}
+		return armrpc_rest.NewInternalServerErrorARMResponse(response), nil
+	}
+
 	proxyReq, err := p.PrepareProxyRequest(ctx, req, downstreamURL.String(), relativePath)
 	if err != nil {
 		return nil, err
 	}
 
 	interceptor := &responseInterceptor{Inner: p.transport}
-
 	sender := proxy.NewARMProxy(proxy.ReverseProxyOptions{RoundTripper: interceptor}, downstreamURL, nil)
 	sender.ServeHTTP(w, proxyReq)
 
@@ -192,6 +214,9 @@ func (p *ProxyController) PrepareProxyRequest(ctx context.Context, originalReq *
 	proxyReq.Header.Set("X-Forwarded-Proto", refererURL.Scheme)
 	proxyReq.Header.Set(v1.RefererHeader, refererURL.String())
 
+	// Clear route context, we don't want to inherit any state from Chi.
+	proxyReq = proxyReq.WithContext(context.WithValue(ctx, chi.RouteCtxKey, nil))
+
 	return proxyReq, nil
 }
 
@@ -220,8 +245,8 @@ func (p *ProxyController) IsTerminalResponse(resp *http.Response) bool {
 }
 
 // UpdateTrackedResource updates the tracked resource synchronously.
-func (p *ProxyController) UpdateTrackedResource(ctx context.Context, downstream string, id resources.ID, apiVersion string) error {
-	return p.updater.Update(ctx, downstream, id, apiVersion)
+func (p *ProxyController) UpdateTrackedResource(ctx context.Context, downstreamURL string, originalID resources.ID, apiVersion string) error {
+	return p.updater.Update(ctx, downstreamURL, originalID, apiVersion)
 }
 
 // EnqueueTrackedResourceUpdate enqueues an async operation to update the tracked resource.
diff --git a/pkg/ucp/frontend/controller/radius/proxy_test.go b/pkg/ucp/frontend/controller/radius/proxy_test.go
index f596210c8a..8e51472637 100644
--- a/pkg/ucp/frontend/controller/radius/proxy_test.go
+++ b/pkg/ucp/frontend/controller/radius/proxy_test.go
@@ -37,6 +37,11 @@ import (
 	"go.uber.org/mock/gomock"
 )
 
+const (
+	apiVersion = "2025-01-01"
+	location   = "global"
+)
+
 // The Run function is also tested by integration tests in the pkg/ucp/integrationtests/radius package.
 
 func createController(t *testing.T) (*ProxyController, *store.MockStorageClient, *mockUpdater, *mockRoundTripper, *statusmanager.MockStatusManager) {
@@ -44,15 +49,18 @@ func createController(t *testing.T) (*ProxyController, *store.MockStorageClient,
 	storageClient := store.NewMockStorageClient(ctrl)
 	statusManager := statusmanager.NewMockStatusManager(ctrl)
 
-	p, err := NewProxyController(controller.Options{StorageClient: storageClient, StatusManager: statusManager})
+	roundTripper := mockRoundTripper{}
+
+	p, err := NewProxyController(
+		controller.Options{StorageClient: storageClient, StatusManager: statusManager},
+		&roundTripper,
+		"http://localhost:1234")
 	require.NoError(t, err)
 
 	updater := mockUpdater{}
-	roundTripper := mockRoundTripper{}
 
 	pc := p.(*ProxyController)
 	pc.updater = &updater
-	pc.transport = &roundTripper
 
 	return pc, storageClient, &updater, &roundTripper, statusManager
 }
@@ -60,6 +68,11 @@ func createController(t *testing.T) (*ProxyController, *store.MockStorageClient,
 func Test_Run(t *testing.T) {
 	id := resources.MustParse("/planes/test/local/resourceGroups/test-rg/providers/Applications.Test/testResources/my-resource")
 
+	// This test covers the legacy (pre-UDT) behavior for looking up the downstream URL. Update
+	// this when the old behavior is removed.
+	resourceTypeID, err := datamodel.ResourceTypeIDFromResourceID(id)
+	require.NoError(t, err)
+
 	plane := datamodel.RadiusPlane{
 		Properties: datamodel.RadiusPlaneProperties{
 			ResourceProviders: map[string]string{
@@ -73,6 +86,7 @@ func Test_Run(t *testing.T) {
 		p, storageClient, _, roundTripper, _ := createController(t)
 
 		svcContext := &v1.ARMRequestContext{
+			APIVersion: apiVersion,
 			ResourceID: id,
 		}
 		ctx := testcontext.New(t)
@@ -81,7 +95,11 @@ func Test_Run(t *testing.T) {
 		w := httptest.NewRecorder()
 
 		// Not a mutating request
-		req := httptest.NewRequest(http.MethodGet, id.String(), nil)
+		req := httptest.NewRequest(http.MethodGet, id.String()+"?api-version="+apiVersion, nil)
+
+		storageClient.EXPECT().
+			Get(gomock.Any(), resourceTypeID.String(), gomock.Any()).
+			Return(nil, &store.ErrNotFound{}).Times(1)
 
 		storageClient.EXPECT().
 			Get(gomock.Any(), "/planes/"+id.PlaneNamespace(), gomock.Any()).
@@ -104,6 +122,7 @@ func Test_Run(t *testing.T) {
 		p, storageClient, updater, roundTripper, _ := createController(t)
 
 		svcContext := &v1.ARMRequestContext{
+			APIVersion: apiVersion,
 			ResourceID: id,
 		}
 		ctx := testcontext.New(t)
@@ -112,7 +131,11 @@ func Test_Run(t *testing.T) {
 		w := httptest.NewRecorder()
 
 		// Mutating request that will complete synchronously
-		req := httptest.NewRequest(http.MethodDelete, id.String(), nil)
+		req := httptest.NewRequest(http.MethodDelete, id.String()+"?api-version="+apiVersion, nil)
+
+		storageClient.EXPECT().
+			Get(gomock.Any(), resourceTypeID.String(), gomock.Any()).
+			Return(nil, &store.ErrNotFound{}).Times(1)
 
 		storageClient.EXPECT().
 			Get(gomock.Any(), "/planes/"+id.PlaneNamespace(), gomock.Any()).
@@ -138,6 +161,7 @@ func Test_Run(t *testing.T) {
 		p, storageClient, updater, roundTripper, statusManager := createController(t)
 
 		svcContext := &v1.ARMRequestContext{
+			APIVersion: apiVersion,
 			ResourceID: id,
 		}
 		ctx := testcontext.New(t)
@@ -146,7 +170,11 @@ func Test_Run(t *testing.T) {
 		w := httptest.NewRecorder()
 
 		// Mutating request that will complete synchronously
-		req := httptest.NewRequest(http.MethodDelete, id.String(), nil)
+		req := httptest.NewRequest(http.MethodDelete, id.String()+"?api-version="+apiVersion, nil)
+
+		storageClient.EXPECT().
+			Get(gomock.Any(), resourceTypeID.String(), gomock.Any()).
+			Return(nil, &store.ErrNotFound{}).Times(1)
 
 		storageClient.EXPECT().
 			Get(gomock.Any(), "/planes/"+id.PlaneNamespace(), gomock.Any()).
@@ -184,6 +212,7 @@ func Test_Run(t *testing.T) {
 		p, storageClient, updater, roundTripper, _ := createController(t)
 
 		svcContext := &v1.ARMRequestContext{
+			APIVersion: apiVersion,
 			ResourceID: id,
 		}
 		ctx := testcontext.New(t)
@@ -192,7 +221,11 @@ func Test_Run(t *testing.T) {
 		w := httptest.NewRecorder()
 
 		// Mutating request that will complete asynchronously
-		req := httptest.NewRequest(http.MethodDelete, id.String(), nil)
+		req := httptest.NewRequest(http.MethodDelete, id.String()+"?api-version="+apiVersion, nil)
+
+		storageClient.EXPECT().
+			Get(gomock.Any(), resourceTypeID.String(), gomock.Any()).
+			Return(nil, &store.ErrNotFound{}).Times(1)
 
 		storageClient.EXPECT().
 			Get(gomock.Any(), "/planes/"+id.PlaneNamespace(), gomock.Any()).
@@ -237,19 +270,20 @@ func Test_Run(t *testing.T) {
 		p, storageClient, _, _, _ := createController(t)
 
 		svcContext := &v1.ARMRequestContext{
+			APIVersion: apiVersion,
 			ResourceID: id,
 		}
 		ctx := testcontext.New(t)
 		ctx = v1.WithARMRequestContext(ctx, svcContext)
 
 		w := httptest.NewRecorder()
-		req := httptest.NewRequest(http.MethodPut, id.String(), nil)
+		req := httptest.NewRequest(http.MethodPut, id.String()+"?api-version="+apiVersion, nil)
 
 		storageClient.EXPECT().
 			Get(gomock.Any(), "/planes/"+id.PlaneNamespace(), gomock.Any()).
 			Return(nil, &store.ErrNotFound{}).Times(1)
 
-		expected := rest.NewNotFoundResponse(id)
+		expected := rest.NewNotFoundResponseWithCause(id, "plane \"/planes/test/local\" not found")
 
 		response, err := p.Run(ctx, w, req)
 		require.NoError(t, err)
diff --git a/pkg/ucp/frontend/controller/resourcegroups/util.go b/pkg/ucp/frontend/controller/resourcegroups/util.go
index ac8117e0ee..a1d5f8c507 100644
--- a/pkg/ucp/frontend/controller/resourcegroups/util.go
+++ b/pkg/ucp/frontend/controller/resourcegroups/util.go
@@ -13,6 +13,7 @@ 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 resourcegroups
 
 import (
@@ -20,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
+	"strings"
 
 	"github.com/radius-project/radius/pkg/ucp/datamodel"
 	"github.com/radius-project/radius/pkg/ucp/resources"
@@ -59,38 +61,174 @@ func (e *InvalidError) Is(err error) bool {
 	return ok
 }
 
-// ValidateDownstream can be used to find and validate the downstream URL for a resource.
-// Returns NotFoundError for the case where the plane or resource group does not exist.
-// Returns InvalidError for cases where the data is invalid, like when the resource provider is not configured.
-func ValidateDownstream(ctx context.Context, client store.StorageClient, id resources.ID) (*url.URL, error) {
+// ValidateRadiusPlane validates that the plane specified in the id exists. Returns NotFoundError if the plane does not exist.
+func ValidateRadiusPlane(ctx context.Context, client store.StorageClient, id resources.ID) (*datamodel.RadiusPlane, error) {
 	planeID, err := resources.ParseScope(id.PlaneScope())
 	if err != nil {
 		// Not expected to happen.
 		return nil, err
 	}
+
 	plane, err := store.GetResource[datamodel.RadiusPlane](ctx, client, planeID.String())
 	if errors.Is(err, &store.ErrNotFound{}) {
 		return nil, &NotFoundError{Message: fmt.Sprintf("plane %q not found", planeID.String())}
 	} else if err != nil {
-		return nil, fmt.Errorf("failed to find plane %q: %w", planeID.String(), err)
+		return nil, fmt.Errorf("failed to fetch plane %q: %w", planeID.String(), err)
 	}
 
+	return plane, nil
+}
+
+// ValidateResourceGroup validates that the resource group specified in the id exists (if applicable).
+// Returns NotFoundError if the resource group does not exist.
+func ValidateResourceGroup(ctx context.Context, client store.StorageClient, id resources.ID) error {
 	// If the ID contains a resource group, validate it now.
-	if id.FindScope(resources_radius.ScopeResourceGroups) != "" {
-		resourceGroupID, err := resources.ParseScope(id.RootScope())
+	if id.FindScope(resources_radius.ScopeResourceGroups) == "" {
+		return nil
+	}
+
+	resourceGroupID, err := resources.ParseScope(id.RootScope())
+	if err != nil {
+		// Not expected to happen.
+		return err
+	}
+
+	_, err = store.GetResource[datamodel.ResourceGroup](ctx, client, resourceGroupID.String())
+	if errors.Is(err, &store.ErrNotFound{}) {
+		return &NotFoundError{Message: fmt.Sprintf("resource group %q not found", resourceGroupID.String())}
+	} else if err != nil {
+		return fmt.Errorf("failed to fetch resource group %q: %w", resourceGroupID.String(), err)
+	}
+
+	return nil
+}
+
+// ValidateResourceType performs semantic validation of a proxy request against registered
+// resource types.
+//
+// Returns NotFoundError if the resource type does not exist.
+// Returns InvalidError if the request cannot be routed due to an invalid configuration.
+func ValidateResourceType(ctx context.Context, client store.StorageClient, id resources.ID, locationName string, apiVersion string) (*url.URL, error) {
+	// The strategy is to:
+	// - Look up the resource type and validate that it exists .. then
+	// - Look up the location resource, and validate that it supports the requested resource type and API version.
+
+	// We need to do both because they may not be in sync. This can be be the case if a resource type is being added or deleted.
+
+	if !isOperationResourceType(id) {
+		resourceTypeID, err := datamodel.ResourceTypeIDFromResourceID(id)
 		if err != nil {
-			// Not expected to happen.
 			return nil, err
 		}
 
-		_, err = store.GetResource[datamodel.ResourceGroup](ctx, client, resourceGroupID.String())
+		_, err = store.GetResource[datamodel.ResourceType](ctx, client, resourceTypeID.String())
 		if errors.Is(err, &store.ErrNotFound{}) {
-			return nil, &NotFoundError{Message: fmt.Sprintf("resource group %q not found", resourceGroupID.String())}
+
+			// Return the error as-is to fallback to the legacy routing behavior.
+			return nil, err
+
+			// Uncomment this when we remove the legacy routing behavior.
+			// return nil, &InvalidError{Message: fmt.Sprintf("resource type %q not found", id.Type())}
 		} else if err != nil {
-			return nil, fmt.Errorf("failed to find resource group %q: %w", resourceGroupID.String(), err)
+			return nil, fmt.Errorf("failed to fetch resource type %q: %w", id.Type(), err)
+		}
+	}
+
+	locationID, err := datamodel.ResourceProviderLocationIDFromResourceID(id, locationName)
+	if err != nil {
+		return nil, err
+	}
+
+	location, err := store.GetResource[datamodel.Location](ctx, client, locationID.String())
+	if errors.Is(err, &store.ErrNotFound{}) {
+
+		// Return the error as-is to fallback to the legacy routing behavior.
+		return nil, err
+
+		// Uncomment this when we remove the legacy routing behavior.
+		// return nil, &InvalidError{Message: fmt.Sprintf("location %q not found for resource provider %q", locationName, id.ProviderNamespace())}
+	} else if err != nil {
+		return nil, fmt.Errorf("failed to fetch location %q: %w", locationID.String(), err)
+	}
+
+	// Check if the location supports the resource type.
+	// Resource types are case-insensitive so we have to iterate.
+	var locationResourceType *datamodel.LocationResourceTypeConfiguration
+
+	// We special-case two pseudo-resource types: "locations/operationstatuses" and "locations/operationresults".
+	// If the resource type is one of these, we can return the downstream URL directly.
+	if isOperationResourceType(id) {
+		locationResourceType = &datamodel.LocationResourceTypeConfiguration{
+			APIVersions: map[string]datamodel.LocationAPIVersionConfiguration{
+				apiVersion: {}, // Assume this API version is supported.
+			},
 		}
+	} else {
+		// Ex: Applications.Test/testResources => testResources
+		search := strings.TrimPrefix(strings.ToLower(id.Type()), strings.ToLower(id.ProviderNamespace())+resources.SegmentSeparator)
+		for name, rt := range location.Properties.ResourceTypes {
+			if strings.EqualFold(name, search) {
+				copy := rt
+				locationResourceType = &copy
+				break
+			}
+		}
+	}
+
+	// Now check if the location supports the resource type and API version. If it does, we can return the downstream URL.
+	if locationResourceType == nil {
+		return nil, &InvalidError{Message: fmt.Sprintf("resource type %q not supported by location %q", id.Type(), locationName)}
+	}
+
+	_, ok := locationResourceType.APIVersions[apiVersion]
+	if !ok {
+		return nil, &InvalidError{Message: fmt.Sprintf("api version %q is not supported for resource type %q by location %q", apiVersion, id.Type(), locationName)}
+	}
+
+	// If we get to here, then we're all good.
+	//
+	// The address might be nil which means that we're using the default address (dynamic RP)
+	if location.Properties.Address == nil {
+		return nil, nil
+	}
+
+	// If the address was provided, then use that instead.
+	u, err := url.Parse(*location.Properties.Address)
+	if err != nil {
+		return nil, &InvalidError{Message: fmt.Sprintf("failed to parse location address: %v", err.Error())}
 	}
 
+	return u, nil
+}
+
+// isOperationResourceType returns true if the resource type is an operation resource type (operationResults/operationStatuses).
+//
+// We special-case these types, and don't require the resource provider to register them.
+func isOperationResourceType(id resources.ID) bool {
+	// For a resource provider "Applications.Test" the operation resource types are:
+	// - Applications.Test/locations/operationStatuses
+	// - Applications.Test/locations/operationResults
+
+	// Radius resource providers include the location name in the resource id
+	if strings.EqualFold(id.Type(), id.ProviderNamespace()+"/locations/operationstatuses") ||
+		strings.EqualFold(id.Type(), id.ProviderNamespace()+"/locations/operationresults") {
+		return true
+	}
+
+	// An older pattern is to use a child resource
+	typeSegments := id.TypeSegments()
+	if len(typeSegments) >= 2 && (strings.EqualFold(typeSegments[len(typeSegments)-1].Type, "operationstatuses") ||
+		strings.EqualFold(typeSegments[len(typeSegments)-1].Type, "operationresults")) {
+		return true
+	}
+
+	// Not an operation.
+	return false
+}
+
+// ValidateLegacyResourceProvider validates that the resource provider specified in the id exists. Returns InvalidError if the plane
+// contains invalid data.
+func ValidateLegacyResourceProvider(ctx context.Context, client store.StorageClient, id resources.ID, plane *datamodel.RadiusPlane) (*url.URL, error) {
 	downstream := plane.LookupResourceProvider(id.ProviderNamespace())
 	if downstream == "" {
 		return nil, &InvalidError{Message: fmt.Sprintf("resource provider %s not configured", id.ProviderNamespace())}
@@ -103,3 +241,40 @@ func ValidateDownstream(ctx context.Context, client store.StorageClient, id reso
 
 	return downstreamURL, nil
 }
+
+// ValidateDownstream can be used to find and validate the downstream URL for a resource.
+// Returns NotFoundError for the case where the plane or resource group does not exist.
+// Returns InvalidError for cases where the data is invalid, like when the resource provider is not configured.
+func ValidateDownstream(ctx context.Context, client store.StorageClient, id resources.ID, location string, apiVersion string) (*url.URL, error) {
+	// There are a few steps to validation:
+	//
+	// - The plane exists
+	// - The resource group exists
+	// - The resource provider is configured .. either:
+	// 		- As part of the plane (legacy routing)
+	// 		- As part of a resource provider resource (System.Resources/resourceProviders) (new/UDT routing)
+	//
+
+	// The plane exists.
+	plane, err := ValidateRadiusPlane(ctx, client, id)
+	if err != nil {
+		return nil, err
+	}
+
+	// The resource group exists (if applicable).
+	err = ValidateResourceGroup(ctx, client, id)
+	if err != nil {
+		return nil, err
+	}
+
+	// If this returns success, it means the resource type is configured using new/UDT routing.
+	downstreamURL, err := ValidateResourceType(ctx, client, id, location, apiVersion)
+	if errors.Is(err, &store.ErrNotFound{}) {
+		// If the resource provider is not found, treat it like a legacy provider.
+		return ValidateLegacyResourceProvider(ctx, client, id, plane)
+	} else if err != nil {
+		return nil, err
+	}
+
+	return downstreamURL, nil
+}
diff --git a/pkg/ucp/frontend/controller/resourcegroups/util_test.go b/pkg/ucp/frontend/controller/resourcegroups/util_test.go
index f2e530ea62..f266458196 100644
--- a/pkg/ucp/frontend/controller/resourcegroups/util_test.go
+++ b/pkg/ucp/frontend/controller/resourcegroups/util_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
+	"github.com/radius-project/radius/pkg/to"
 	"github.com/radius-project/radius/pkg/ucp/datamodel"
 	"github.com/radius-project/radius/pkg/ucp/resources"
 	"github.com/radius-project/radius/pkg/ucp/store"
@@ -30,6 +31,11 @@ import (
 	"go.uber.org/mock/gomock"
 )
 
+const (
+	location   = "east"
+	apiVersion = "2025-01-01"
+)
+
 func Test_ValidateDownstream(t *testing.T) {
 	id, err := resources.ParseResource("/planes/radius/local/resourceGroups/test-group/providers/System.TestRP/testResources/name")
 	require.NoError(t, err)
@@ -37,6 +43,12 @@ func Test_ValidateDownstream(t *testing.T) {
 	idWithoutResourceGroup, err := resources.Parse("/planes/radius/local/providers/System.TestRP/testResources")
 	require.NoError(t, err)
 
+	resourceTypeID, err := datamodel.ResourceTypeIDFromResourceID(id)
+	require.NoError(t, err)
+
+	locationID, err := datamodel.ResourceProviderLocationIDFromResourceID(id, location)
+	require.NoError(t, err)
+
 	downstream := "http://localhost:7443"
 
 	plane := &datamodel.RadiusPlane{
@@ -52,6 +64,35 @@ func Test_ValidateDownstream(t *testing.T) {
 		},
 	}
 
+	resourceTypeResource := &datamodel.ResourceType{
+		BaseResource: v1.BaseResource{
+			TrackedResource: v1.TrackedResource{
+				Name: "testResources",
+				ID:   resourceTypeID.String(),
+			},
+		},
+		Properties: datamodel.ResourceTypeProperties{},
+	}
+
+	locationResource := &datamodel.Location{
+		BaseResource: v1.BaseResource{
+			TrackedResource: v1.TrackedResource{
+				Name: location,
+				ID:   locationID.String(),
+			},
+		},
+		Properties: datamodel.LocationProperties{
+			Address: to.Ptr("http://localhost:7443"),
+			ResourceTypes: map[string]datamodel.LocationResourceTypeConfiguration{
+				"testResources": {
+					APIVersions: map[string]datamodel.LocationAPIVersionConfiguration{
+						apiVersion: {},
+					},
+				},
+			},
+		},
+	}
+
 	setup := func(t *testing.T) *store.MockStorageClient {
 		ctrl := gomock.NewController(t)
 		return store.NewMockStorageClient(ctrl)
@@ -69,11 +110,13 @@ func Test_ValidateDownstream(t *testing.T) {
 		mock := setup(t)
 		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
 		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeResource.ID).Return(&store.Object{Data: resourceTypeResource}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
 
 		expectedURL, err := url.Parse(downstream)
 		require.NoError(t, err)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
 		require.NoError(t, err)
 		require.Equal(t, expectedURL, downstreamURL)
 	})
@@ -81,11 +124,77 @@ func Test_ValidateDownstream(t *testing.T) {
 	t.Run("success (non resource group)", func(t *testing.T) {
 		mock := setup(t)
 		mock.EXPECT().Get(gomock.Any(), idWithoutResourceGroup.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeResource.ID).Return(&store.Object{Data: resourceTypeResource}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
 
 		expectedURL, err := url.Parse(downstream)
 		require.NoError(t, err)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, idWithoutResourceGroup)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, idWithoutResourceGroup, location, apiVersion)
+		require.NoError(t, err)
+		require.Equal(t, expectedURL, downstreamURL)
+	})
+
+	// The deployment engine models its operation status resources as child resources of the deployment resource.
+	t.Run("success (operationstatuses as child resource)", func(t *testing.T) {
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), idWithoutResourceGroup.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
+
+		operationStatusID := resources.MustParse("/planes/radius/local/providers/System.TestRP/deployments/xzy/operationStatuses/abcd")
+
+		expectedURL, err := url.Parse(downstream)
+		require.NoError(t, err)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, operationStatusID, location, apiVersion)
+		require.NoError(t, err)
+		require.Equal(t, expectedURL, downstreamURL)
+	})
+
+	// All of the Radius RPs include a location in the operation status child resource.
+	t.Run("success (operationstatuses with location)", func(t *testing.T) {
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), idWithoutResourceGroup.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
+
+		operationStatusID := resources.MustParse("/planes/radius/local/providers/System.TestRP/locations/east/operationStatuses/abcd")
+
+		expectedURL, err := url.Parse(downstream)
+		require.NoError(t, err)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, operationStatusID, location, apiVersion)
+		require.NoError(t, err)
+		require.Equal(t, expectedURL, downstreamURL)
+	})
+
+	// The deployment engine models its operation result resources as child resources of the deployment resource.
+	t.Run("success (operationresults as child resource)", func(t *testing.T) {
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), idWithoutResourceGroup.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
+
+		operationResultID := resources.MustParse("/planes/radius/local/providers/System.TestRP/deployments/xzy/operationResults/abcd")
+
+		expectedURL, err := url.Parse(downstream)
+		require.NoError(t, err)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, operationResultID, location, apiVersion)
+		require.NoError(t, err)
+		require.Equal(t, expectedURL, downstreamURL)
+	})
+
+	// All of the Radius RPs include a location in the operation result child resource.
+	t.Run("success (operationresults with location)", func(t *testing.T) {
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), idWithoutResourceGroup.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
+
+		operationResultID := resources.MustParse("/planes/radius/local/providers/System.TestRP/locations/east/operationResults/abcd")
+
+		expectedURL, err := url.Parse(downstream)
+		require.NoError(t, err)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, operationResultID, location, apiVersion)
 		require.NoError(t, err)
 		require.Equal(t, expectedURL, downstreamURL)
 	})
@@ -94,7 +203,7 @@ func Test_ValidateDownstream(t *testing.T) {
 		mock := setup(t)
 		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(nil, &store.ErrNotFound{}).Times(1)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
 		require.Error(t, err)
 		require.Equal(t, &NotFoundError{Message: "plane \"/planes/radius/local\" not found"}, err)
 		require.Nil(t, downstreamURL)
@@ -103,10 +212,10 @@ func Test_ValidateDownstream(t *testing.T) {
 	t.Run("plane retreival failure", func(t *testing.T) {
 		mock := setup(t)
 
-		expected := fmt.Errorf("failed to find plane \"/planes/radius/local\": %w", errors.New("test error"))
+		expected := fmt.Errorf("failed to fetch plane \"/planes/radius/local\": %w", errors.New("test error"))
 		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(nil, errors.New("test error")).Times(1)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
 		require.Error(t, err)
 		require.Equal(t, expected, err)
 		require.Nil(t, downstreamURL)
@@ -117,7 +226,7 @@ func Test_ValidateDownstream(t *testing.T) {
 		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
 		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(nil, &store.ErrNotFound{}).Times(1)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
 		require.Error(t, err)
 		require.Equal(t, &NotFoundError{Message: "resource group \"/planes/radius/local/resourceGroups/test-group\" not found"}, err)
 		require.Nil(t, downstreamURL)
@@ -129,13 +238,322 @@ func Test_ValidateDownstream(t *testing.T) {
 		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
 		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(nil, errors.New("test error")).Times(1)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, "failed to fetch resource group \"/planes/radius/local/resourceGroups/test-group\": test error", err.Error())
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("resource type error", func(t *testing.T) {
+		resourceGroup := &datamodel.ResourceGroup{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.RootScope(),
+				},
+			},
+		}
+
+		expected := fmt.Errorf("failed to fetch resource type %q: %w", "System.TestRP/testResources", errors.New("test error"))
+
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeResource.ID).Return(nil, errors.New("test error")).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, expected, err)
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("location error", func(t *testing.T) {
+		resourceGroup := &datamodel.ResourceGroup{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.RootScope(),
+				},
+			},
+		}
+
+		expected := fmt.Errorf("failed to fetch location %q: %w", locationResource.ID, errors.New("test error"))
+
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeResource.ID).Return(&store.Object{Data: resourceTypeID}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(nil, errors.New("test error")).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
 		require.Error(t, err)
-		require.Equal(t, "failed to find resource group \"/planes/radius/local/resourceGroups/test-group\": test error", err.Error())
+		require.Equal(t, expected, err)
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("resource type not found in location", func(t *testing.T) {
+		resourceGroup := &datamodel.ResourceGroup{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.RootScope(),
+				},
+			},
+		}
+
+		locationResource := &datamodel.Location{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					Name: location,
+					ID:   locationResource.ID,
+				},
+			},
+			Properties: datamodel.LocationProperties{
+				Address: to.Ptr("http://localhost:7443"),
+				ResourceTypes: map[string]datamodel.LocationResourceTypeConfiguration{
+					"testResources2": {
+						APIVersions: map[string]datamodel.LocationAPIVersionConfiguration{
+							apiVersion: {},
+						},
+					},
+				},
+			},
+		}
+
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeResource.ID).Return(&store.Object{Data: resourceTypeID}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, &InvalidError{Message: "resource type \"System.TestRP/testResources\" not supported by location \"east\""}, err)
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("api-version not found in location", func(t *testing.T) {
+		resourceGroup := &datamodel.ResourceGroup{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.RootScope(),
+				},
+			},
+		}
+
+		locationResource := &datamodel.Location{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					Name: location,
+					ID:   locationResource.ID,
+				},
+			},
+			Properties: datamodel.LocationProperties{
+				Address: to.Ptr("http://localhost:7443"),
+				ResourceTypes: map[string]datamodel.LocationResourceTypeConfiguration{
+					"testResources": {
+						APIVersions: map[string]datamodel.LocationAPIVersionConfiguration{
+							apiVersion + "-preview": {},
+						},
+					},
+				},
+			},
+		}
+
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeResource.ID).Return(&store.Object{Data: resourceTypeID}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, &InvalidError{Message: "api version \"2025-01-01\" is not supported for resource type \"System.TestRP/testResources\" by location \"east\""}, err)
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("location invalid URL", func(t *testing.T) {
+		resourceGroup := &datamodel.ResourceGroup{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.RootScope(),
+				},
+			},
+		}
+
+		locationResource := &datamodel.Location{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					Name: location,
+					ID:   locationResource.ID,
+				},
+			},
+			Properties: datamodel.LocationProperties{
+				Address: to.Ptr("\ninvalid"),
+				ResourceTypes: map[string]datamodel.LocationResourceTypeConfiguration{
+					"testResources": {
+						APIVersions: map[string]datamodel.LocationAPIVersionConfiguration{
+							apiVersion: {},
+						},
+					},
+				},
+			},
+		}
+
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeResource.ID).Return(&store.Object{Data: resourceTypeID}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), locationResource.ID).Return(&store.Object{Data: locationResource}, nil).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, &InvalidError{Message: "failed to parse location address: parse \"\\ninvalid\": net/url: invalid control character in URL"}, err)
+		require.Nil(t, downstreamURL)
+	})
+}
+
+// This test validates the pre-UDT before where resource providers are registered as part of the plane.
+// This can be deleted once the legacy routing behavior is removed.
+func Test_ValidateDownstream_Legacy(t *testing.T) {
+	id, err := resources.ParseResource("/planes/radius/local/resourceGroups/test-group/providers/System.TestRP/testResources/name")
+	require.NoError(t, err)
+
+	idWithoutResourceGroup, err := resources.Parse("/planes/radius/local/providers/System.TestRP/testResources")
+	require.NoError(t, err)
+
+	resourceTypeID, err := datamodel.ResourceTypeIDFromResourceID(id)
+	require.NoError(t, err)
+
+	downstream := "http://localhost:7443"
+
+	plane := &datamodel.RadiusPlane{
+		BaseResource: v1.BaseResource{
+			TrackedResource: v1.TrackedResource{
+				ID: id.PlaneScope(),
+			},
+		},
+		Properties: datamodel.RadiusPlaneProperties{
+			ResourceProviders: map[string]string{
+				"System.TestRP": downstream,
+			},
+		},
+	}
+
+	setup := func(t *testing.T) *store.MockStorageClient {
+		ctrl := gomock.NewController(t)
+		return store.NewMockStorageClient(ctrl)
+	}
+
+	t.Run("success (resource group)", func(t *testing.T) {
+		resourceGroup := &datamodel.ResourceGroup{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.RootScope(),
+				},
+			},
+		}
+
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeID.String()).Return(nil, &store.ErrNotFound{}).Times(1)
+
+		expectedURL, err := url.Parse(downstream)
+		require.NoError(t, err)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.NoError(t, err)
+		require.Equal(t, expectedURL, downstreamURL)
+	})
+
+	t.Run("success (non resource group)", func(t *testing.T) {
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), idWithoutResourceGroup.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeID.String()).Return(nil, &store.ErrNotFound{}).Times(1)
+
+		expectedURL, err := url.Parse(downstream)
+		require.NoError(t, err)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, idWithoutResourceGroup, location, apiVersion)
+		require.NoError(t, err)
+		require.Equal(t, expectedURL, downstreamURL)
+	})
+
+	t.Run("plane not found", func(t *testing.T) {
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(nil, &store.ErrNotFound{}).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, &NotFoundError{Message: "plane \"/planes/radius/local\" not found"}, err)
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("plane retrieval failure", func(t *testing.T) {
+		mock := setup(t)
+
+		expected := fmt.Errorf("failed to fetch plane \"/planes/radius/local\": %w", errors.New("test error"))
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(nil, errors.New("test error")).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, expected, err)
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("resource group not found", func(t *testing.T) {
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(nil, &store.ErrNotFound{}).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, &NotFoundError{Message: "resource group \"/planes/radius/local/resourceGroups/test-group\" not found"}, err)
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("resource group err", func(t *testing.T) {
+		mock := setup(t)
+
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(nil, errors.New("test error")).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, "failed to fetch resource group \"/planes/radius/local/resourceGroups/test-group\": test error", err.Error())
+		require.Nil(t, downstreamURL)
+	})
+
+	t.Run("legacy resource provider not configured", func(t *testing.T) {
+		plane := &datamodel.RadiusPlane{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.PlaneScope(),
+				},
+			},
+			Properties: datamodel.RadiusPlaneProperties{
+				ResourceProviders: map[string]string{},
+			},
+		}
+
+		resourceGroup := &datamodel.ResourceGroup{
+			BaseResource: v1.BaseResource{
+				TrackedResource: v1.TrackedResource{
+					ID: id.RootScope(),
+				},
+			},
+		}
+
+		mock := setup(t)
+		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeID.String()).Return(nil, &store.ErrNotFound{}).Times(1)
+
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
+		require.Error(t, err)
+		require.Equal(t, &InvalidError{Message: "resource provider System.TestRP not configured"}, err)
 		require.Nil(t, downstreamURL)
 	})
 
-	t.Run("resource provider not found", func(t *testing.T) {
+	t.Run("location not found", func(t *testing.T) {
 		plane := &datamodel.RadiusPlane{
 			BaseResource: v1.BaseResource{
 				TrackedResource: v1.TrackedResource{
@@ -158,8 +576,9 @@ func Test_ValidateDownstream(t *testing.T) {
 		mock := setup(t)
 		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
 		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeID.String()).Return(nil, &store.ErrNotFound{}).Times(1)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
 		require.Error(t, err)
 		require.Equal(t, &InvalidError{Message: "resource provider System.TestRP not configured"}, err)
 		require.Nil(t, downstreamURL)
@@ -190,8 +609,9 @@ func Test_ValidateDownstream(t *testing.T) {
 		mock := setup(t)
 		mock.EXPECT().Get(gomock.Any(), id.PlaneScope()).Return(&store.Object{Data: plane}, nil).Times(1)
 		mock.EXPECT().Get(gomock.Any(), id.RootScope()).Return(&store.Object{Data: resourceGroup}, nil).Times(1)
+		mock.EXPECT().Get(gomock.Any(), resourceTypeID.String()).Return(nil, &store.ErrNotFound{}).Times(1)
 
-		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id)
+		downstreamURL, err := ValidateDownstream(testcontext.New(t), mock, id, location, apiVersion)
 		require.Error(t, err)
 		require.Equal(t, &InvalidError{Message: "failed to parse downstream URL: parse \"\\ninvalid\": net/url: invalid control character in URL"}, err)
 		require.Nil(t, downstreamURL)
diff --git a/pkg/ucp/frontend/radius/module.go b/pkg/ucp/frontend/radius/module.go
index 1dae8a5641..811e5692d4 100644
--- a/pkg/ucp/frontend/radius/module.go
+++ b/pkg/ucp/frontend/radius/module.go
@@ -28,15 +28,16 @@ func NewModule(options modules.Options) *Module {
 	router.NotFound(validator.APINotFoundHandler())
 	router.MethodNotAllowed(validator.APIMethodNotAllowedHandler())
 
-	return &Module{options: options, router: router}
+	return &Module{options: options, router: router, defaultDownstream: options.Config.Routing.DefaultDownstreamEndpoint}
 }
 
 var _ modules.Initializer = &Module{}
 
 // Module defines the module for Radius functionality.
 type Module struct {
-	options modules.Options
-	router  chi.Router
+	options           modules.Options
+	router            chi.Router
+	defaultDownstream string
 }
 
 // PlaneType returns the type of plane this module is for.
diff --git a/pkg/ucp/frontend/radius/routes.go b/pkg/ucp/frontend/radius/routes.go
index 22bf6288b6..2631bd9afd 100644
--- a/pkg/ucp/frontend/radius/routes.go
+++ b/pkg/ucp/frontend/radius/routes.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"net/http"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
@@ -34,11 +35,15 @@ import (
 	resourcegroups_ctrl "github.com/radius-project/radius/pkg/ucp/frontend/controller/resourcegroups"
 	resourceproviders_ctrl "github.com/radius-project/radius/pkg/ucp/frontend/controller/resourceproviders"
 	"github.com/radius-project/radius/pkg/validator"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
 const (
 	// OperationTypeUCPRadiusProxy is the operation type for proxying Radius API calls.
 	OperationTypeUCPRadiusProxy = "UCPRADIUSPROXY"
+
+	// operationRetryAfter tells clients to poll in 1 second intervals. Our operations are fast.
+	operationRetryAfter = time.Second * 1
 )
 
 func (m *Module) Initialize(ctx context.Context) (http.Handler, error) {
@@ -47,6 +52,8 @@ func (m *Module) Initialize(ctx context.Context) (http.Handler, error) {
 		ResourceTypeGetter: validator.UCPResourceTypeGetter,
 	})
 
+	transport := otelhttp.NewTransport(http.DefaultTransport)
+
 	// More convienent way to capture errors
 	var err error
 	capture := func(handler http.HandlerFunc, e error) http.HandlerFunc {
@@ -124,7 +131,7 @@ func (m *Module) Initialize(ctx context.Context) (http.Handler, error) {
 				// Proxy to plane-scoped ResourceProvider APIs
 				//
 				// NOTE: DO NOT validate schema for proxy routes.
-				r.Handle("/*", capture(planeScopedProxyHandler(ctx, ctrlOptions)))
+				r.Handle("/*", capture(planeScopedProxyHandler(ctx, ctrlOptions, transport, m.defaultDownstream)))
 			})
 
 			r.Route("/resourcegroups", func(r chi.Router) {
@@ -141,7 +148,7 @@ func (m *Module) Initialize(ctx context.Context) (http.Handler, error) {
 						// Proxy to resource-group-scoped ResourceProvider APIs
 						//
 						// NOTE: DO NOT validate schema for proxy routes.
-						r.Handle("/*", capture(resourceGroupScopedProxyHandler(ctx, ctrlOptions)))
+						r.Handle("/*", capture(resourceGroupScopedProxyHandler(ctx, ctrlOptions, transport, m.defaultDownstream)))
 					})
 				})
 
@@ -153,8 +160,9 @@ func (m *Module) Initialize(ctx context.Context) (http.Handler, error) {
 }
 
 var planeResourceOptions = controller.ResourceOptions[datamodel.RadiusPlane]{
-	RequestConverter:  converter.RadiusPlaneDataModelFromVersioned,
-	ResponseConverter: converter.RadiusPlaneDataModelToVersioned,
+	RequestConverter:         converter.RadiusPlaneDataModelFromVersioned,
+	ResponseConverter:        converter.RadiusPlaneDataModelToVersioned,
+	AsyncOperationRetryAfter: operationRetryAfter,
 }
 
 var planeResourceType = "System.Radius/planes"
@@ -231,8 +239,9 @@ func resourceProviderSummaryGetHandler(ctx context.Context, ctrlOptions controll
 }
 
 var resourceProviderResourceOptions = controller.ResourceOptions[datamodel.ResourceProvider]{
-	RequestConverter:  converter.ResourceProviderDataModelFromVersioned,
-	ResponseConverter: converter.ResourceProviderDataModelToVersioned,
+	RequestConverter:         converter.ResourceProviderDataModelFromVersioned,
+	ResponseConverter:        converter.ResourceProviderDataModelToVersioned,
+	AsyncOperationRetryAfter: operationRetryAfter,
 }
 
 func resourceProviderListHandler(ctx context.Context, ctrlOptions controller.Options) (http.HandlerFunc, error) {
@@ -260,8 +269,9 @@ func resourceProviderDeleteHandler(ctx context.Context, ctrlOptions controller.O
 }
 
 var resourceTypeResourceOptions = controller.ResourceOptions[datamodel.ResourceType]{
-	RequestConverter:  converter.ResourceTypeDataModelFromVersioned,
-	ResponseConverter: converter.ResourceTypeDataModelToVersioned,
+	RequestConverter:         converter.ResourceTypeDataModelFromVersioned,
+	ResponseConverter:        converter.ResourceTypeDataModelToVersioned,
+	AsyncOperationRetryAfter: operationRetryAfter,
 }
 
 func resourceTypeListHandler(ctx context.Context, ctrlOptions controller.Options) (http.HandlerFunc, error) {
@@ -289,8 +299,9 @@ func resourceTypeDeleteHandler(ctx context.Context, ctrlOptions controller.Optio
 }
 
 var apiVersionResourceOptions = controller.ResourceOptions[datamodel.APIVersion]{
-	RequestConverter:  converter.APIVersionDataModelFromVersioned,
-	ResponseConverter: converter.APIVersionDataModelToVersioned,
+	RequestConverter:         converter.APIVersionDataModelFromVersioned,
+	ResponseConverter:        converter.APIVersionDataModelToVersioned,
+	AsyncOperationRetryAfter: operationRetryAfter,
 }
 
 func apiVersionListHandler(ctx context.Context, ctrlOptions controller.Options) (http.HandlerFunc, error) {
@@ -318,8 +329,9 @@ func apiVersionDeleteHandler(ctx context.Context, ctrlOptions controller.Options
 }
 
 var locationResourceOptions = controller.ResourceOptions[datamodel.Location]{
-	RequestConverter:  converter.LocationDataModelFromVersioned,
-	ResponseConverter: converter.LocationDataModelToVersioned,
+	RequestConverter:         converter.LocationDataModelFromVersioned,
+	ResponseConverter:        converter.LocationDataModelToVersioned,
+	AsyncOperationRetryAfter: operationRetryAfter,
 }
 
 func locationListHandler(ctx context.Context, ctrlOptions controller.Options) (http.HandlerFunc, error) {
@@ -346,12 +358,16 @@ func locationDeleteHandler(ctx context.Context, ctrlOptions controller.Options)
 	})
 }
 
-func planeScopedProxyHandler(ctx context.Context, ctrlOptions controller.Options) (http.HandlerFunc, error) {
-	return server.CreateHandler(ctx, OperationTypeUCPRadiusProxy, v1.OperationProxy, ctrlOptions, radius_ctrl.NewProxyController)
+func planeScopedProxyHandler(ctx context.Context, ctrlOptions controller.Options, transport http.RoundTripper, defaultDownstream string) (http.HandlerFunc, error) {
+	return server.CreateHandler(ctx, OperationTypeUCPRadiusProxy, v1.OperationProxy, ctrlOptions, func(o controller.Options) (controller.Controller, error) {
+		return radius_ctrl.NewProxyController(o, transport, defaultDownstream)
+	})
 }
 
-func resourceGroupScopedProxyHandler(ctx context.Context, ctrlOptions controller.Options) (http.HandlerFunc, error) {
-	return server.CreateHandler(ctx, OperationTypeUCPRadiusProxy, v1.OperationProxy, ctrlOptions, radius_ctrl.NewProxyController)
+func resourceGroupScopedProxyHandler(ctx context.Context, ctrlOptions controller.Options, transport http.RoundTripper, defaultDownstream string) (http.HandlerFunc, error) {
+	return server.CreateHandler(ctx, OperationTypeUCPRadiusProxy, v1.OperationProxy, ctrlOptions, func(o controller.Options) (controller.Controller, error) {
+		return radius_ctrl.NewProxyController(o, transport, defaultDownstream)
+	})
 }
 
 func operationStatusGetHandler(ctx context.Context, ctrlOptions controller.Options) (http.HandlerFunc, error) {
diff --git a/pkg/ucp/hostoptions/providerconfig.go b/pkg/ucp/hostoptions/providerconfig.go
index 40907d7082..4b37e6a8cf 100644
--- a/pkg/ucp/hostoptions/providerconfig.go
+++ b/pkg/ucp/hostoptions/providerconfig.go
@@ -41,6 +41,7 @@ type UCPConfig struct {
 	Identity         Identity                                 `yaml:"identity,omitempty"`
 	UCP              config.UCPOptions                        `yaml:"ucp"`
 	Location         string                                   `yaml:"location"`
+	Routing          RoutingConfig                            `yaml:"routing"`
 }
 
 const (
@@ -55,3 +56,10 @@ type Identity struct {
 	// AuthMethod represents the method of authentication for authenticating with external systems like Azure and AWS.
 	AuthMethod string `yaml:"authMethod"`
 }
+
+// RoutingConfig provides configuration for UCP routing.
+type RoutingConfig struct {
+	// DefaultDownstreamEndpoint is the default destination when a resource provider does not provide a downstream endpoint.
+	// In practice, this points to the URL of dynamic-rp.
+	DefaultDownstreamEndpoint string `yaml:"defaultDownstreamEndpoint"`
+}
diff --git a/pkg/ucp/integrationtests/radius/proxy_test.go b/pkg/ucp/integrationtests/radius/proxy_test.go
index 3d019e8ec4..371a79af0a 100644
--- a/pkg/ucp/integrationtests/radius/proxy_test.go
+++ b/pkg/ucp/integrationtests/radius/proxy_test.go
@@ -56,9 +56,9 @@ func Test_RadiusPlane_Proxy_ResourceGroupDoesNotExist(t *testing.T) {
 	}
 	createRadiusPlane(ucp, rps)
 
-	response := ucp.MakeRequest(http.MethodGet, testResourceID, nil)
+	response := ucp.MakeRequest(http.MethodGet, testResourceID+"?api-version="+testrp.Version, nil)
 	response.EqualsErrorCode(http.StatusNotFound, "NotFound")
-	require.Equal(t, "the resource with id '/planes/radius/test/resourceGroups/test-rg/providers/System.Test/testResources/test-resource' was not found", response.Error.Error.Message)
+	require.Equal(t, "the resource with id '/planes/radius/test/resourceGroups/test-rg/providers/System.Test/testResources/test-resource' was not found: resource group \"/planes/radius/test/resourceGroups/test-rg\" not found", response.Error.Error.Message)
 }
 
 func Test_RadiusPlane_ResourceSync(t *testing.T) {
diff --git a/pkg/ucp/integrationtests/testserver/testserver.go b/pkg/ucp/integrationtests/testserver/testserver.go
index c0134c0db6..f1eba83023 100644
--- a/pkg/ucp/integrationtests/testserver/testserver.go
+++ b/pkg/ucp/integrationtests/testserver/testserver.go
@@ -298,16 +298,6 @@ func StartWithETCD(t *testing.T, configureModules func(options modules.Options)
 
 	statusManager := statusmanager.New(dataProvider, queueClient, v1.LocationGlobal)
 
-	registry := worker.NewControllerRegistry(dataProvider)
-	err = backend.RegisterControllers(ctx, registry, connection, backend_ctrl.Options{DataProvider: dataProvider})
-	require.NoError(t, err)
-
-	w := worker.New(worker.Options{}, statusManager, queueClient, registry)
-	go func() {
-		err = w.Start(ctx)
-		require.NoError(t, err)
-	}()
-
 	specLoader, err := validator.LoadSpec(ctx, "ucp", swagger.SpecFilesUCP, []string{pathBase}, "")
 	require.NoError(t, err, "failed to load OpenAPI spec")
 
@@ -328,6 +318,25 @@ func StartWithETCD(t *testing.T, configureModules func(options modules.Options)
 
 	modules := configureModules(options)
 
+	// The URL for the dynamic-rp needs to be specified in configuration, however not all of our tests
+	// need to use the dynamic-rp. We can just use a placeholder value here.
+	if options.Config.Routing.DefaultDownstreamEndpoint == "" {
+		options.Config.Routing.DefaultDownstreamEndpoint = "http://localhost:65535"
+	}
+
+	defaultDownstream, err := url.Parse(options.Config.Routing.DefaultDownstreamEndpoint)
+	require.NoError(t, err)
+
+	registry := worker.NewControllerRegistry(dataProvider)
+	err = backend.RegisterControllers(ctx, registry, connection, http.DefaultTransport, backend_ctrl.Options{DataProvider: dataProvider}, defaultDownstream)
+	require.NoError(t, err)
+
+	w := worker.New(worker.Options{}, statusManager, queueClient, registry)
+	go func() {
+		err = w.Start(ctx)
+		require.NoError(t, err)
+	}()
+
 	err = api.Register(ctx, router, modules, options)
 	require.NoError(t, err)
 
diff --git a/pkg/ucp/server/server.go b/pkg/ucp/server/server.go
index 421fd45d1c..83a1e9cd8b 100644
--- a/pkg/ucp/server/server.go
+++ b/pkg/ucp/server/server.go
@@ -199,7 +199,7 @@ func NewServer(options *Options) (*hosting.Host, error) {
 			ProfilerProvider: options.ProfilerProviderOptions,
 		},
 	}
-	hostingServices = append(hostingServices, backend.NewService(backendServiceOptions))
+	hostingServices = append(hostingServices, backend.NewService(backendServiceOptions, *options.Config))
 
 	options.TracerProviderOptions.ServiceName = "ucp"
 	hostingServices = append(hostingServices, &trace.Service{Options: options.TracerProviderOptions})