Add async operation support to dynamic-rp (#8161)

# Description

This change adds the async operationStatus and operationResult handlers
to the dynamic-rp API. This change is significant, because like all
functionality in dynamic-rp, it's dynamic and needs to work for any UDT
resource provider rather than a fixed namespace.

## Type of change

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


Fixes: #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]>

pkg/armrpc/asyncoperation/worker/worker.go

@@ -267,7 +267,12 @@ func (w *AsyncRequestProcessWorker) runOperation(ctx context.Context, message *q
 			result.SetFailed(armErr, false)
 		}
 
-		logger.Info("Operation returned", "success", result.Error == nil, "provisioningState", result.ProvisioningState(), "err", result.Error)
+		// We need the if/else here to prevent a panic inside the logger.
+		if result.Error == nil {
+			logger.Info("Operation returned", "success", "true", "provisioningState", result.ProvisioningState())
+		} else {
+			logger.Info("Operation returned", "success", "false", "provisioningState", result.ProvisioningState(), "err", result.Error)
+		}
 
 		// There are two cases when asyncReqCtx is canceled.
 		// 1. When the operation is timed out, w.completeOperation will be called in L186

pkg/dynamicrp/frontend/operations.go

@@ -0,0 +1,88 @@
+/*
+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 frontend
+
+import (
+	"net/http"
+	"strings"
+
+	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
+	"github.com/radius-project/radius/pkg/armrpc/frontend/controller"
+	"github.com/radius-project/radius/pkg/armrpc/frontend/server"
+	"github.com/radius-project/radius/pkg/armrpc/rest"
+	"github.com/radius-project/radius/pkg/ucp/resources"
+)
+
+// dynamicOperationHandler returns an http.Handler that can instantiate and run a controller.Controller for a dynamic resource.
+//
+// Usually when we register a route, we know up-front the resource type that will be handled by the controller.
+// In the dynamic-rp use-case we don't know. We need to dynamically determine the resource type based on the URL.
+//
+// For example:
+//
+// Route: /planes/radius/{planeName}/providers/{providerNamespace}/locations/{locationName}/operationResults/{operationID}
+// URL: /planes/radius/myplane/providers/Applications.Example/locations/global/operationResults/1234
+// Resource Type: Applications.Example/operationResults
+//
+// # OR
+//
+// Route: /planes/radius/{planeName}/resourceGroups/my-rg/providers/{providerNamespace}/{resourceType}/{resourceName}
+// URL: /planes/radius/myplane/resourceGroups/my-rg/providers/Applications.Example/customService/my-service
+// Resource Type: Applications.Example/customService
+//
+// This code ensures that the controller will be provided with the correct resource type.
+func dynamicOperationHandler(method v1.OperationMethod, baseOptions controller.Options, factory func(opts controller.Options) (controller.Controller, error)) http.HandlerFunc {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		id, err := resources.Parse(r.URL.Path)
+		if err != nil {
+			result := rest.NewBadRequestResponse(err.Error())
+			err = result.Apply(r.Context(), w, r)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+			}
+
+			return
+		}
+
+		operationType := v1.OperationType{Type: strings.ToUpper(id.Type()), Method: method}
+
+		// Copy the options and initalize them dynamically for this type.
+		opts := baseOptions
+		opts.ResourceType = id.Type()
+
+		// Special case the operation status and operation result types.
+		//
+		// This is special-casing that all of our resource providers do to store a single data row for both operation statuses and operation results.
+		if strings.HasSuffix(strings.ToLower(opts.ResourceType), "locations/operationstatuses") || strings.HasSuffix(strings.ToLower(opts.ResourceType), "locations/operationresults") {
+			opts.ResourceType = id.ProviderNamespace() + "/operationstatuses"
+		}
+
+		ctrl, err := factory(opts)
+		if err != nil {
+			result := rest.NewBadRequestResponse(err.Error())
+			err = result.Apply(r.Context(), w, r)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+			}
+
+			return
+		}
+
+		handler := server.HandlerForController(ctrl, operationType)
+		handler.ServeHTTP(w, r)
+	})
+}

pkg/dynamicrp/frontend/routes.go

@@ -17,14 +17,47 @@ limitations under the License.
 package frontend
 
 import (
+	"strings"
+
 	"github.com/go-chi/chi/v5"
+	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
+	"github.com/radius-project/radius/pkg/armrpc/frontend/controller"
+	"github.com/radius-project/radius/pkg/armrpc/frontend/defaultoperation"
 	"github.com/radius-project/radius/pkg/validator"
 )
 
-func (s *Service) registerRoutes(r *chi.Mux) error {
+func (s *Service) registerRoutes(r *chi.Mux, controllerOptions controller.Options) error {
+	// Return ARM errors for invalid requests.
+	r.NotFound(validator.APINotFoundHandler())
+	r.MethodNotAllowed(validator.APIMethodNotAllowedHandler())
+
 	// Return ARM errors for invalid requests.
 	r.NotFound(validator.APINotFoundHandler())
 	r.MethodNotAllowed(validator.APIMethodNotAllowedHandler())
 
+	pathBase := s.options.Config.Server.PathBase
+	if pathBase == "" {
+		pathBase = "/"
+	}
+
+	if !strings.HasSuffix(pathBase, "/") {
+		pathBase = pathBase + "/"
+	}
+
+	r.Route(pathBase+"planes/radius/{planeName}/providers/{providerNamespace}", func(r chi.Router) {
+		r.Route("/locations/{locationName}", func(r chi.Router) {
+			r.Get("/{or:operation[Rr]esults}/{operationID}", dynamicOperationHandler(v1.OperationGet, controllerOptions, makeGetOperationResultController))
+			r.Get("/{os:operation[Ss]tatuses}/{operationID}", dynamicOperationHandler(v1.OperationGet, controllerOptions, makeGetOperationStatusController))
+		})
+	})
+
 	return nil
 }
+
+func makeGetOperationResultController(opts controller.Options) (controller.Controller, error) {
+	return defaultoperation.NewGetOperationResult(opts)
+}
+
+func makeGetOperationStatusController(opts controller.Options) (controller.Controller, error) {
+	return defaultoperation.NewGetOperationStatus(opts)
+}

pkg/dynamicrp/frontend/service.go

@@ -22,6 +22,7 @@ import (
 	"net"
 	"net/http"
 
+	"github.com/radius-project/radius/pkg/armrpc/frontend/controller"
 	"github.com/radius-project/radius/pkg/armrpc/servicecontext"
 	"github.com/radius-project/radius/pkg/dynamicrp"
 	"github.com/radius-project/radius/pkg/middleware"
@@ -54,7 +55,22 @@ func (s *Service) Name() string {
 func (s *Service) initialize(ctx context.Context) (*http.Server, error) {
 	r := chi.NewRouter()
 
-	err := s.registerRoutes(r)
+	databaseClient, err := s.options.DatabaseProvider.GetClient(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get database client: %w", err)
+	}
+
+	controllerOptions := controller.Options{
+		Address:        s.options.Config.Server.Address(),
+		PathBase:       s.options.Config.Server.PathBase,
+		DatabaseClient: databaseClient,
+		StatusManager:  s.options.StatusManager,
+
+		KubeClient:   nil, // Unused by DynamicRP
+		ResourceType: "",  // Set dynamically
+	}
+
+	err = s.registerRoutes(r, controllerOptions)
 	if err != nil {
 		return nil, fmt.Errorf("failed to register routes: %w", err)
 	}

pkg/dynamicrp/integrationtest/dynamic/operations_test.go

@@ -0,0 +1,79 @@
+/*
+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 dynamic
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/google/uuid"
+	v1 "github.com/radius-project/radius/pkg/armrpc/api/v1"
+	"github.com/radius-project/radius/pkg/armrpc/asyncoperation/statusmanager"
+	"github.com/radius-project/radius/pkg/components/database"
+	"github.com/radius-project/radius/pkg/dynamicrp/testhost"
+	"github.com/radius-project/radius/test/testcontext"
+	"github.com/stretchr/testify/require"
+)
+
+// This test covers the basic functionality of the operation status/result controllers.
+//
+// This test is synthetic because we don't have a real operation to test against.
+func Test_Dynamic_OperationResultAndStatus(t *testing.T) {
+	ctx := testcontext.New(t)
+	dynamic, ucp := testhost.Start(t)
+
+	// Setup a plane & resource provider & location
+	plane := createRadiusPlane(ucp)
+	createResourceProvider(ucp)
+	createLocation(ucp)
+
+	// Now we can make a request to the operation result/status endpoints.
+	operationName := uuid.New().String()
+
+	operationResultID := fmt.Sprintf("/planes/radius/%s/providers/%s/locations/global/operationResults/%s", *plane.Name, resourceProviderNamespace, operationName)
+	operationStatusID := fmt.Sprintf("/planes/radius/%s/providers/%s/locations/global/operationStatuses/%s", *plane.Name, resourceProviderNamespace, operationName)
+
+	// This operation doesn't exist yet, so we should get a 404.
+	response := ucp.MakeRequest("GET", fmt.Sprintf("%s?api-version=%s", operationResultID, apiVersion), nil)
+	response.EqualsErrorCode(http.StatusNotFound, "NotFound")
+
+	response = ucp.MakeRequest("GET", fmt.Sprintf("%s?api-version=%s", operationStatusID, apiVersion), nil)
+	response.EqualsErrorCode(http.StatusNotFound, "NotFound")
+
+	// Now let's simulate the creation of an operation, by putting one in the database.
+	databaseClient, err := dynamic.Options().DatabaseProvider.GetClient(ctx)
+	require.NoError(t, err)
+
+	operation := &statusmanager.Status{
+		AsyncOperationStatus: v1.AsyncOperationStatus{
+			ID:     operationStatusID,
+			Name:   operationName,
+			Status: v1.ProvisioningStateUpdating,
+		},
+	}
+
+	err = databaseClient.Save(ctx, &database.Object{Data: operation, Metadata: database.Metadata{ID: operationStatusID}})
+	require.NoError(t, err)
+
+	// Now let's query it again, we should find it.
+	response = ucp.MakeRequest("GET", fmt.Sprintf("%s?api-version=%s", operationResultID, apiVersion), nil)
+	response.EqualsStatusCode(http.StatusAccepted)
+
+	response = ucp.MakeRequest("GET", fmt.Sprintf("%s?api-version=%s", operationStatusID, apiVersion), nil)
+	response.EqualsStatusCode(http.StatusOK)
+}

pkg/dynamicrp/testhost/host.go

@@ -55,6 +55,11 @@ func (f TestHostOptionFunc) Apply(options *dynamicrp.Options) {
 // TestHost provides a test host for the dynamic-rp server.
 type TestHost struct {
 	*testhost.TestHost
+	options *dynamicrp.Options
+}
+
+func (th *TestHost) Options() *dynamicrp.Options {
+	return th.options
 }
 
 func Start(t *testing.T, opts ...TestHostOption) (*TestHost, *ucptesthost.TestHost) {
@@ -124,7 +129,7 @@ func StartWithOptions(t *testing.T, options *dynamicrp.Options) (*TestHost, *ucp
 	require.NoError(t, err, "failed to create server")
 
 	th := testhost.StartHost(t, host, baseURL)
-	return &TestHost{th}, startUCP(t, baseURL, ucpPort)
+	return &TestHost{TestHost: th, options: options}, startUCP(t, baseURL, ucpPort)
 }
 
 func startUCP(t *testing.T, dynamicRPURL string, ucpPort int) *ucptesthost.TestHost {