Make CreateOrUpdate controller async for link resources

pkg/armrpc/frontend/controller/controller.go

@@ -8,6 +8,7 @@ package controller
 import (
 	"context"
 	"net/http"
+	"time"
 
 	v1 "github.com/project-radius/radius/pkg/armrpc/api/v1"
 	sm "github.com/project-radius/radius/pkg/armrpc/asyncoperation/statusmanager"
@@ -54,6 +55,9 @@ type ResourceOptions[T any] struct {
 
 	// UpdateFilters is a slice of filters that execute prior to updating a resource.
 	UpdateFilters []UpdateFilter[T]
+
+	// AsyncOperationTimeout is the default timeout duration of async put operation.
+	AsyncOperationTimeout time.Duration
 }
 
 // TODO: Remove Controller when all controller uses Operation

pkg/armrpc/frontend/controller/operation.go

@@ -20,6 +20,11 @@ import (
 	"github.com/project-radius/radius/pkg/ucp/store"
 )
 
+const (
+	// defaultAsyncPutTimeout is the default timeout duration of async put operation.
+	defaultAsyncPutTimeout = time.Duration(2) * time.Minute
+)
+
 // Operation is the base operation controller.
 type Operation[P interface {
 	*T
@@ -218,3 +223,11 @@ func (b *Operation[P, T]) DeleteFilters() []DeleteFilter[T] {
 func (b *Operation[P, T]) UpdateFilters() []UpdateFilter[T] {
 	return b.resourceOptions.UpdateFilters
 }
+
+// AsyncOperationTimeout returns the timeput for the operation.
+func (b *Operation[P, T]) AsyncOperationTimeout() time.Duration {
+	if b.resourceOptions.AsyncOperationTimeout == 0 {
+		return defaultAsyncPutTimeout
+	}
+	return b.resourceOptions.AsyncOperationTimeout
+}

pkg/armrpc/frontend/defaultoperation/defaultasyncdelete.go

@@ -8,18 +8,12 @@ package defaultoperation
 import (
 	"context"
 	"net/http"
-	"time"
 
 	v1 "github.com/project-radius/radius/pkg/armrpc/api/v1"
 	ctrl "github.com/project-radius/radius/pkg/armrpc/frontend/controller"
 	"github.com/project-radius/radius/pkg/armrpc/rest"
 )
 
-var (
-	// defaultAsyncDeleteTimeout is the default timeout duration of async delete operation.
-	defaultAsyncDeleteTimeout = time.Duration(120) * time.Second
-)
-
 // DefaultAsyncDelete is the controller implementation to delete async resource.
 type DefaultAsyncDelete[P interface {
 	*T
@@ -58,7 +52,7 @@ func (e *DefaultAsyncDelete[P, T]) Run(ctx context.Context, w http.ResponseWrite
 		}
 	}
 
-	if r, err := e.PrepareAsyncOperation(ctx, old, v1.ProvisioningStateAccepted, defaultAsyncDeleteTimeout, &etag); r != nil || err != nil {
+	if r, err := e.PrepareAsyncOperation(ctx, old, v1.ProvisioningStateAccepted, e.AsyncOperationTimeout(), &etag); r != nil || err != nil {
 		return r, err
 	}
 

pkg/armrpc/frontend/defaultoperation/defaultasyncput.go

@@ -8,18 +8,12 @@ package defaultoperation
 import (
 	"context"
 	"net/http"
-	"time"
 
 	v1 "github.com/project-radius/radius/pkg/armrpc/api/v1"
 	ctrl "github.com/project-radius/radius/pkg/armrpc/frontend/controller"
 	"github.com/project-radius/radius/pkg/armrpc/rest"
 )
 
-var (
-	// defaultAsyncPutTimeout is the default timeout duration of async put operation.
-	defaultAsyncPutTimeout = time.Duration(120) * time.Second
-)
-
 // DefaultAsyncPut is the controller implementation to create or update async resource.
 type DefaultAsyncPut[P interface {
 	*T
@@ -59,7 +53,7 @@ func (e *DefaultAsyncPut[P, T]) Run(ctx context.Context, w http.ResponseWriter,
 		}
 	}
 
-	if r, err := e.PrepareAsyncOperation(ctx, newResource, v1.ProvisioningStateAccepted, defaultAsyncPutTimeout, &etag); r != nil || err != nil {
+	if r, err := e.PrepareAsyncOperation(ctx, newResource, v1.ProvisioningStateAccepted, e.AsyncOperationTimeout(), &etag); r != nil || err != nil {
 		return r, err
 	}
 

pkg/corerp/backend/controller/createorupdateresource.go

@@ -100,8 +100,10 @@ func (c *CreateOrUpdateResource) Run(ctx context.Context, request *ctrl.Request)
 
 	oldOutputResources := deploymentDataModel.OutputResources()
 
-	deploymentDataModel.ApplyDeploymentOutput(deploymentOutput)
-
+	err = deploymentDataModel.ApplyDeploymentOutput(deploymentOutput)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
 	if !isNewResource {
 		diff := rpv1.GetGCOutputResources(deploymentDataModel.OutputResources(), oldOutputResources)
 		err = c.DeploymentProcessor().Delete(ctx, id, diff)

pkg/corerp/backend/deployment/deploymentprocessor.go

@@ -73,8 +73,8 @@ type ResourceData struct {
 	OutputResources []rpv1.OutputResource
 	ComputedValues  map[string]any
 	SecretValues    map[string]rpv1.SecretValueReference
-	AppID           *resources.ID      // Application ID for which the resource is created
-	RecipeData      link_dm.RecipeData // Relevant only for links created with recipes to find relevant connections created by that recipe
+	AppID           *resources.ID     // Application ID for which the resource is created
+	RecipeData      linkrp.RecipeData // Relevant only for links created with recipes to find relevant connections created by that recipe
 }
 
 func (dp *deploymentProcessor) Render(ctx context.Context, resourceID resources.ID, resource v1.DataModelInterface) (renderers.RendererOutput, error) {
@@ -496,25 +496,25 @@ func (dp *deploymentProcessor) getResourceDataByID(ctx context.Context, resource
 		if err = resource.As(obj); err != nil {
 			return ResourceData{}, fmt.Errorf(errMsg, resourceID.String(), err)
 		}
-		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, link_dm.RecipeData{})
+		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, linkrp.RecipeData{})
 	case strings.ToLower(gateway.ResourceType):
 		obj := &corerp_dm.Gateway{}
 		if err = resource.As(obj); err != nil {
 			return ResourceData{}, fmt.Errorf(errMsg, resourceID.String(), err)
 		}
-		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, link_dm.RecipeData{})
+		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, linkrp.RecipeData{})
 	case strings.ToLower(volume.ResourceType):
 		obj := &corerp_dm.VolumeResource{}
 		if err = resource.As(obj); err != nil {
 			return ResourceData{}, fmt.Errorf(errMsg, resourceID.String(), err)
 		}
-		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, link_dm.RecipeData{})
+		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, linkrp.RecipeData{})
 	case strings.ToLower(httproute.ResourceType):
 		obj := &corerp_dm.HTTPRoute{}
 		if err = resource.As(obj); err != nil {
 			return ResourceData{}, fmt.Errorf(errMsg, resourceID.String(), err)
 		}
-		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, link_dm.RecipeData{})
+		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, linkrp.RecipeData{})
 	case strings.ToLower(linkrp.MongoDatabasesResourceType):
 		obj := &link_dm.MongoDatabase{}
 		if err = resource.As(obj); err != nil {
@@ -544,7 +544,7 @@ func (dp *deploymentProcessor) getResourceDataByID(ctx context.Context, resource
 		if err = resource.As(obj); err != nil {
 			return ResourceData{}, fmt.Errorf(errMsg, resourceID.String(), err)
 		}
-		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, link_dm.RecipeData{})
+		return dp.buildResourceDependency(resourceID, obj.Properties.Application, obj, obj.Properties.Status.OutputResources, obj.ComputedValues, obj.SecretValues, linkrp.RecipeData{})
 	case strings.ToLower(linkrp.DaprStateStoresResourceType):
 		obj := &link_dm.DaprStateStore{}
 		if err = resource.As(obj); err != nil {
@@ -574,7 +574,7 @@ func (dp *deploymentProcessor) getResourceDataByID(ctx context.Context, resource
 	}
 }
 
-func (dp *deploymentProcessor) buildResourceDependency(resourceID resources.ID, applicationID string, resource v1.DataModelInterface, outputResources []rpv1.OutputResource, computedValues map[string]any, secretValues map[string]rpv1.SecretValueReference, recipeData link_dm.RecipeData) (ResourceData, error) {
+func (dp *deploymentProcessor) buildResourceDependency(resourceID resources.ID, applicationID string, resource v1.DataModelInterface, outputResources []rpv1.OutputResource, computedValues map[string]any, secretValues map[string]rpv1.SecretValueReference, recipeData linkrp.RecipeData) (ResourceData, error) {
 	var appID *resources.ID
 	if applicationID != "" {
 		parsedID, err := resources.ParseResource(applicationID)

pkg/corerp/backend/deployment/deploymentprocessor_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/project-radius/radius/pkg/corerp/model"
 	"github.com/project-radius/radius/pkg/corerp/renderers"
 	"github.com/project-radius/radius/pkg/corerp/renderers/container"
+	"github.com/project-radius/radius/pkg/linkrp"
 	linkrp_dm "github.com/project-radius/radius/pkg/linkrp/datamodel"
 	linkrp_renderers "github.com/project-radius/radius/pkg/linkrp/renderers"
 	"github.com/project-radius/radius/pkg/linkrp/renderers/mongodatabases"
@@ -182,9 +183,9 @@ func buildMongoDBLinkWithRecipe() linkrp_dm.MongoDatabase {
 			Mode: linkrp_dm.LinkModeRecipe,
 		},
 		LinkMetadata: linkrp_dm.LinkMetadata{
-			RecipeData: linkrp_dm.RecipeData{
-				RecipeProperties: linkrp_dm.RecipeProperties{
-					LinkRecipe: linkrp_dm.LinkRecipe{
+			RecipeData: linkrp.RecipeData{
+				RecipeProperties: linkrp.RecipeProperties{
+					LinkRecipe: linkrp.LinkRecipe{
 						Name: "mongoDB",
 						Parameters: map[string]any{
 							"ResourceGroup": "testRG",

pkg/corerp/datamodel/application.go

@@ -26,8 +26,9 @@ func (e *Application) ResourceTypeName() string {
 }
 
 // ApplyDeploymentOutput applies the properties changes based on the deployment output.
-func (c *Application) ApplyDeploymentOutput(do rpv1.DeploymentOutput) {
+func (c *Application) ApplyDeploymentOutput(do rpv1.DeploymentOutput) error {
 	c.Properties.Status.OutputResources = do.DeployedOutputResources
+	return nil
 }
 
 // OutputResources returns the output resources array.

pkg/corerp/datamodel/container.go

@@ -27,10 +27,11 @@ func (c ContainerResource) ResourceTypeName() string {
 }
 
 // ApplyDeploymentOutput applies the properties changes based on the deployment output.
-func (c *ContainerResource) ApplyDeploymentOutput(do rpv1.DeploymentOutput) {
+func (c *ContainerResource) ApplyDeploymentOutput(do rpv1.DeploymentOutput) error {
 	c.Properties.Status.OutputResources = do.DeployedOutputResources
 	c.ComputedValues = do.ComputedValues
 	c.SecretValues = do.SecretValues
+	return nil
 }
 
 // OutputResources returns the output resources array.

pkg/corerp/datamodel/gateway.go

@@ -26,13 +26,14 @@ func (g *Gateway) ResourceTypeName() string {
 }
 
 // ApplyDeploymentOutput applies the properties changes based on the deployment output.
-func (g *Gateway) ApplyDeploymentOutput(do rpv1.DeploymentOutput) {
+func (g *Gateway) ApplyDeploymentOutput(do rpv1.DeploymentOutput) error {
 	g.Properties.Status.OutputResources = do.DeployedOutputResources
 	g.ComputedValues = do.ComputedValues
 	g.SecretValues = do.SecretValues
 	if url, ok := do.ComputedValues["url"].(string); ok {
 		g.Properties.URL = url
 	}
+	return nil
 }
 
 // OutputResources returns the output resources array.

pkg/corerp/datamodel/httproute.go

@@ -26,7 +26,7 @@ func (h *HTTPRoute) ResourceTypeName() string {
 }
 
 // ApplyDeploymentOutput applies the properties changes based on the deployment output.
-func (h *HTTPRoute) ApplyDeploymentOutput(do rpv1.DeploymentOutput) {
+func (h *HTTPRoute) ApplyDeploymentOutput(do rpv1.DeploymentOutput) error {
 	if h.Properties != nil {
 		h.Properties.Status.OutputResources = do.DeployedOutputResources
 	}
@@ -46,6 +46,7 @@ func (h *HTTPRoute) ApplyDeploymentOutput(do rpv1.DeploymentOutput) {
 	if url, ok := do.ComputedValues["url"].(string); ok {
 		h.Properties.URL = url
 	}
+	return nil
 }
 
 // OutputResources returns the output resources array.

pkg/corerp/datamodel/volume.go

@@ -32,10 +32,11 @@ func (h *VolumeResource) ResourceTypeName() string {
 }
 
 // ApplyDeploymentOutput applies the properties changes based on the deployment output.
-func (h *VolumeResource) ApplyDeploymentOutput(do rpv1.DeploymentOutput) {
+func (h *VolumeResource) ApplyDeploymentOutput(do rpv1.DeploymentOutput) error {
 	h.Properties.Status.OutputResources = do.DeployedOutputResources
 	h.ComputedValues = do.ComputedValues
 	h.SecretValues = do.SecretValues
+	return nil
 }
 
 // OutputResources returns the output resources array.

pkg/linkrp/api/v20220315privatepreview/datamodel_util.go

@@ -11,7 +11,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	autorestTo "github.com/Azure/go-autorest/autorest/to"
 	v1 "github.com/project-radius/radius/pkg/armrpc/api/v1"
-	"github.com/project-radius/radius/pkg/linkrp/datamodel"
+	"github.com/project-radius/radius/pkg/linkrp"
 )
 
 func toProvisioningStateDataModel(state *ProvisioningState) v1.ProvisioningState {
@@ -78,8 +78,8 @@ func fromSystemDataModel(s v1.SystemData) *SystemData {
 	}
 }
 
-func toRecipeDataModel(r *Recipe) datamodel.LinkRecipe {
-	recipe := datamodel.LinkRecipe{
+func toRecipeDataModel(r *Recipe) linkrp.LinkRecipe {
+	recipe := linkrp.LinkRecipe{
 		Name: autorestTo.String(r.Name),
 	}
 
@@ -89,7 +89,7 @@ func toRecipeDataModel(r *Recipe) datamodel.LinkRecipe {
 	return recipe
 }
 
-func fromRecipeDataModel(r datamodel.LinkRecipe) *Recipe {
+func fromRecipeDataModel(r linkrp.LinkRecipe) *Recipe {
 	return &Recipe{
 		Name:       autorestTo.StringPtr(r.Name),
 		Parameters: r.Parameters,

pkg/linkrp/backend/controller/createorupdateresource.go

@@ -0,0 +1,103 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+// ------------------------------------------------------------
+
+package controller
+
+import (
+	"context"
+	"errors"
+	"net/http"
+
+	v1 "github.com/project-radius/radius/pkg/armrpc/api/v1"
+	ctrl "github.com/project-radius/radius/pkg/armrpc/asyncoperation/controller"
+	rpv1 "github.com/project-radius/radius/pkg/rp/v1"
+	"github.com/project-radius/radius/pkg/ucp/resources"
+	"github.com/project-radius/radius/pkg/ucp/store"
+)
+
+var _ ctrl.Controller = (*CreateOrUpdateResource)(nil)
+
+// CreateOrUpdateResource is the async operation controller to create or update Applications.Link resources.
+type CreateOrUpdateResource struct {
+	ctrl.BaseController
+}
+
+// NewCreateOrUpdateResource creates the CreateOrUpdateResource controller instance.
+func NewCreateOrUpdateResource(opts ctrl.Options) (ctrl.Controller, error) {
+	return &CreateOrUpdateResource{ctrl.NewBaseAsyncController(opts)}, nil
+}
+
+func (c *CreateOrUpdateResource) Run(ctx context.Context, req *ctrl.Request) (ctrl.Result, error) {
+	obj, err := c.StorageClient().Get(ctx, req.ResourceID)
+	if err != nil && !errors.Is(&store.ErrNotFound{}, err) {
+		return ctrl.Result{}, err
+	}
+
+	isNewResource := false
+	if errors.Is(&store.ErrNotFound{}, err) {
+		isNewResource = true
+	}
+
+	opType, _ := v1.ParseOperationType(req.OperationType)
+	if opType.Method == http.MethodPatch && errors.Is(&store.ErrNotFound{}, err) {
+		return ctrl.Result{}, err
+	}
+
+	// This code is general and we might be processing an async job for a resource or a scope, so using the general Parse function.
+	id, err := resources.Parse(req.ResourceID)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	dataModel, err := getDataModel(id)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	if err = obj.As(dataModel); err != nil {
+		return ctrl.Result{}, err
+	}
+
+	rendererOutput, err := c.LinkDeploymentProcessor().Render(ctx, id, dataModel)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	deploymentOutput, err := c.LinkDeploymentProcessor().Deploy(ctx, id, rendererOutput)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	deploymentDataModel, ok := dataModel.(rpv1.DeploymentDataModel)
+	if !ok {
+		return ctrl.NewFailedResult(v1.ErrorDetails{Message: "deployment data model conversion error"}), err
+	}
+
+	oldOutputResources := deploymentDataModel.OutputResources()
+	err = deploymentDataModel.ApplyDeploymentOutput(deploymentOutput)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	if !isNewResource {
+		diff := rpv1.GetGCOutputResources(deploymentDataModel.OutputResources(), oldOutputResources)
+		err = c.LinkDeploymentProcessor().Delete(ctx, id, diff)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+	}
+	nr := &store.Object{
+		Metadata: store.Metadata{
+			ID: req.ResourceID,
+		},
+		Data: deploymentDataModel,
+	}
+	err = c.StorageClient().Save(ctx, nr, store.WithETag(obj.ETag))
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	return ctrl.Result{}, err
+}