Refactor and implement shared integration test host

This change updates implements a shared (reusable) integration test host for the Radius control-plane services. The new integration test host enables us to do in-memory testing of UCP and dynamic-rp using a "full stack" approach.

This change is a significant refactor because the "glue" code in UCP had many points of divergence with the rest of our codebase. The following major changes are the bulk of the work:

- Defining new types for configuration + options in UCP
- Updating the UCP configuration file to match the format of other components
- Making the use of databases consistent in UCP

-----

The database change is significant, and addresses a longstanding problem in the UCP code that dates back to when it was first written.

- Each instance of store.Client interface is bound to a single resource type (type-per-table).
- applications-rp was written with this restriction in mind.
- UCP was not written with this restriction in mind, and attempts to store and retrieve multiple resource types without regard to the configuration of the store.Client
- Some of our database implementations enforce this limitation and some do not ....

The new in-memory client enforces this limitation and so this was as good a time as any to update the code.

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

.vscode/launch.json

@@ -58,12 +58,7 @@
       "args": [
         "--config-file",
         "${workspaceFolder}/cmd/ucpd/ucp-dev.yaml"
-      ],
-      "env": {
-        "BASE_PATH": "/apis/api.ucp.dev/v1alpha3",
-        "PORT": "9000",
-        "UCP_CONFIG": "${workspaceFolder}/cmd/ucpd/ucp-self-hosted-dev.yaml"
-      }
+      ]
     },
     {
       "name": "Launch Controller",

cmd/applications-rp/cmd/root.go

@@ -111,7 +111,12 @@ var rootCmd = &cobra.Command{
 			Services: hostingSvc,
 		}
 
+		// Make the logger available to the services.
 		ctx := logr.NewContext(context.Background(), logger)
+
+		// Make the hosting configuration available to the services.
+		ctx = hostoptions.WithContext(ctx, options.Config)
+
 		return hosting.RunWithInterrupts(ctx, host)
 	},
 }

cmd/applications-rp/radius-self-hosted.yaml

@@ -8,7 +8,7 @@
 # - Disables metrics and profiler
 #
 environment:
-  name: Dev
+  name: self-hosted
   roleLocation: "global"
 storageProvider:
   provider: "apiserver"

cmd/ucpd/cmd/root.go

@@ -19,13 +19,15 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"os"
 
 	"github.com/go-logr/logr"
 	"github.com/spf13/cobra"
 	etcdclient "go.etcd.io/etcd/client/v3"
 	runtimelog "sigs.k8s.io/controller-runtime/pkg/log"
 
 	"github.com/radius-project/radius/pkg/armrpc/hostoptions"
+	"github.com/radius-project/radius/pkg/ucp"
 	"github.com/radius-project/radius/pkg/ucp/dataprovider"
 	"github.com/radius-project/radius/pkg/ucp/hosting"
 	"github.com/radius-project/radius/pkg/ucp/server"
@@ -38,12 +40,23 @@ var rootCmd = &cobra.Command{
 	Long:  `Server process for the Universal Control Plane (UCP).`,
 	RunE: func(cmd *cobra.Command, args []string) error {
 		configFilePath := cmd.Flag("config-file").Value.String()
-		options, err := server.NewServerOptionsFromEnvironment(configFilePath)
+
+		bs, err := os.ReadFile(configFilePath)
 		if err != nil {
-			return err
+			return fmt.Errorf("failed to read configuration file: %w", err)
+		}
+
+		config, err := ucp.LoadConfig(bs)
+		if err != nil {
+			return fmt.Errorf("failed to parse configuration file: %w", err)
+		}
+
+		options, err := ucp.NewOptions(cmd.Context(), config)
+		if err != nil {
+			return fmt.Errorf("failed to create server options: %w", err)
 		}
 
-		logger, flush, err := ucplog.NewLogger(ucplog.LoggerName, &options.LoggingOptions)
+		logger, flush, err := ucplog.NewLogger(ucplog.LoggerName, &options.Config.Logging)
 		if err != nil {
 			return err
 		}
@@ -52,17 +65,17 @@ var rootCmd = &cobra.Command{
 		// Must set the logger before using controller-runtime.
 		runtimelog.SetLogger(logger)
 
-		if options.StorageProviderOptions.Provider == dataprovider.TypeETCD &&
-			options.StorageProviderOptions.ETCD.InMemory {
+		if options.Config.Storage.Provider == dataprovider.TypeETCD &&
+			options.Config.Storage.ETCD.InMemory {
 			// For in-memory etcd we need to register another service to manage its lifecycle.
 			//
 			// The client will be initialized asynchronously.
 			clientconfigSource := hosting.NewAsyncValue[etcdclient.Client]()
-			options.StorageProviderOptions.ETCD.Client = clientconfigSource
-			options.SecretProviderOptions.ETCD.Client = clientconfigSource
+			options.Config.Storage.ETCD.Client = clientconfigSource
+			options.Config.Storage.ETCD.Client = clientconfigSource
 		}
 
-		host, err := server.NewServer(&options)
+		host, err := server.NewServer(options)
 		if err != nil {
 			return err
 		}

cmd/ucpd/ucp-dev.yaml

@@ -9,7 +9,13 @@
 # - Talk to Portable Resources' Providers on port 8081
 # - Disables metrics and profiler
 #
-location: 'global'
+environment:
+  name: Dev
+  roleLocation: "global"
+server:
+  port: 9000
+  pathBase: /apis/api.ucp.dev/v1alpha3
+
 storageProvider:
   provider: "apiserver"
   apiserver:
@@ -32,19 +38,20 @@ profilerProvider:
 
 #Default planes configuration with which ucp starts
 # TODO: Remove azure and aws planes once rad provider commands are supported
-planes:
-  - id: "/planes/aws/aws"
-    properties:
-      kind: "AWS"
-  - id: "/planes/radius/local"
-    properties:
-      resourceProviders:
-        Applications.Core: "http://localhost:8080"
-        Applications.Messaging: "http://localhost:8080"
-        Applications.Dapr: "http://localhost:8080"
-        Applications.Datastores: "http://localhost:8080"
-        Microsoft.Resources: "http://localhost:5017"
-      kind: "UCPNative"
+initialization:
+  planes:
+    - id: "/planes/aws/aws"
+      properties:
+        kind: "AWS"
+    - id: "/planes/radius/local"
+      properties:
+        resourceProviders:
+          Applications.Core: "http://localhost:8080"
+          Applications.Messaging: "http://localhost:8080"
+          Applications.Dapr: "http://localhost:8080"
+          Applications.Datastores: "http://localhost:8080"
+          Microsoft.Resources: "http://localhost:5017"
+        kind: "UCPNative"
 
 identity:
   authMethod: default
@@ -76,4 +83,4 @@ logging:
 tracerProvider:
   serviceName: "ucp"
   zipkin:
-    url: "http://localhost:9411/api/v2/spans"
+    url: "http://localhost:9411/api/v2/spans"

deploy/Chart/templates/ucp/configmaps.yaml

@@ -10,7 +10,13 @@ data:
   ucp-config.yaml: |-
     # Radius configuration file.
     # See https://github.com/radius-project/radius/blob/main/docs/contributing/contributing-code/contributing-code-control-plane/configSettings.md for more information.
-    location: 'global'
+    environment:
+      name: Dev
+      roleLocation: "global"
+    server:
+      port: 9443
+      pathBase: /apis/api.ucp.dev/v1alpha3
+      tlsCertificateDirectory: /var/tls/cert
     storageProvider:
       provider: "apiserver"
       apiserver:
@@ -30,20 +36,20 @@ data:
     profilerProvider:
       enabled: true
       port: 6060
-
-    planes:
-      - id: "/planes/radius/local"
-        properties:
-          resourceProviders:
-            Applications.Core: "http://applications-rp.radius-system:5443"
-            Applications.Dapr: "http://applications-rp.radius-system:5443"
-            Applications.Datastores: "http://applications-rp.radius-system:5443"
-            Applications.Messaging: "http://applications-rp.radius-system:5443"
-            Microsoft.Resources: "http://bicep-de.radius-system:6443"
-          kind: "UCPNative"
-      - id: "/planes/aws/aws"
-        properties:
-          kind: "AWS"
+    initialization:
+      planes:
+        - id: "/planes/radius/local"
+          properties:
+            resourceProviders:
+              Applications.Core: "http://applications-rp.radius-system:5443"
+              Applications.Dapr: "http://applications-rp.radius-system:5443"
+              Applications.Datastores: "http://applications-rp.radius-system:5443"
+              Applications.Messaging: "http://applications-rp.radius-system:5443"
+              Microsoft.Resources: "http://bicep-de.radius-system:6443"
+            kind: "UCPNative"
+        - id: "/planes/aws/aws"
+          properties:
+            kind: "AWS"
 
     identity:
       authMethod: UCPCredential

deploy/Chart/templates/ucp/deployment.yaml

@@ -36,8 +36,6 @@ spec:
         args:
         - --config-file=/etc/config/ucp-config.yaml
         env:
-        - name: BASE_PATH
-          value: '/apis/api.ucp.dev/v1alpha3' # listen for APIService URLs
         - name: TLS_CERT_DIR
           value: '/var/tls/cert'
         - name: PORT

pkg/armrpc/asyncoperation/worker/service.go

@@ -18,57 +18,64 @@ package worker
 
 import (
 	"context"
+	"sync"
 
 	manager "github.com/radius-project/radius/pkg/armrpc/asyncoperation/statusmanager"
-	"github.com/radius-project/radius/pkg/armrpc/hostoptions"
 	"github.com/radius-project/radius/pkg/ucp/dataprovider"
-	queue "github.com/radius-project/radius/pkg/ucp/queue/client"
 	qprovider "github.com/radius-project/radius/pkg/ucp/queue/provider"
 	"github.com/radius-project/radius/pkg/ucp/ucplog"
 )
 
 // Service is the base worker service implementation to initialize and start worker.
+// All exported fields should be initialized by the caller.
 type Service struct {
-	// ProviderName is the name of provider namespace.
-	ProviderName string
-	// Options is the server hosting options.
-	Options hostoptions.HostOptions
-	// StorageProvider is the provider of storage client.
-	StorageProvider dataprovider.DataStorageProvider
 	// OperationStatusManager is the manager of the operation status.
 	OperationStatusManager manager.StatusManager
-	// Controllers is the registry of the async operation controllers.
-	Controllers *ControllerRegistry
-	// RequestQueue is the queue client for async operation request message.
-	RequestQueue queue.Client
+
+	// Options configures options for the async worker.
+	Options Options
+
+	// QueueProvider is the queue provider. Will be initialized from config if not provided.
+	QueueProvider *qprovider.QueueProvider
+
+	// StorageProvider is the provider of storage client. Will be initialized from config if not provided.
+	StorageProvider dataprovider.DataStorageProvider
+
+	// controllers is the registry of the async operation controllers.
+	controllers *ControllerRegistry
+
+	// controllersInit is used to ensure single initialization of controllers.
+	controllersInit sync.Once
 }
 
-// Init initializes worker service - it initializes the StorageProvider, RequestQueue, OperationStatusManager, Controllers, KubeClient and
-// returns an error if any of these operations fail.
-func (s *Service) Init(ctx context.Context) error {
-	s.StorageProvider = dataprovider.NewStorageProvider(s.Options.Config.StorageProvider)
-	qp := qprovider.New(s.Options.Config.QueueProvider)
-	var err error
-	s.RequestQueue, err = qp.GetClient(ctx)
-	if err != nil {
-		return err
-	}
-	s.OperationStatusManager = manager.New(s.StorageProvider, s.RequestQueue, s.Options.Config.Env.RoleLocation)
-	s.Controllers = NewControllerRegistry(s.StorageProvider)
-	return nil
+// Controllers returns the controller registry.
+func (s *Service) Controllers() *ControllerRegistry {
+	s.controllersInit.Do(func() {
+		s.controllers = NewControllerRegistry(s.StorageProvider)
+	})
+
+	return s.controllers
 }
 
-// Start creates and starts a worker, and logs any errors that occur while starting the worker.
-func (s *Service) Start(ctx context.Context, opt Options) error {
+// Start creates and starts an async worker.
+//
+// The provided context will be provided to each async controller.
+func (s *Service) Start(ctx context.Context) error {
 	logger := ucplog.FromContextOrDiscard(ctx)
-	ctx = hostoptions.WithContext(ctx, s.Options.Config)
+
+	// Create the queue reader for the worker.
+	requestQueue, err := s.QueueProvider.GetClient(ctx)
+	if err != nil {
+		return err
+	}
 
 	// Create and start worker.
-	worker := New(opt, s.OperationStatusManager, s.RequestQueue, s.Controllers)
+	worker := New(s.Options, s.OperationStatusManager, requestQueue, s.Controllers())
 
 	logger.Info("Start Worker...")
 	if err := worker.Start(ctx); err != nil {
 		logger.Error(err, "failed to start worker...")
+		return err
 	}
 
 	logger.Info("Worker stopped...")

pkg/armrpc/asyncoperation/worker/worker_runoperation_test.go

@@ -114,7 +114,7 @@ func (c *testContext) cancellable(timeout time.Duration) (context.Context, conte
 
 func newTestContext(t *testing.T, lockTime time.Duration) (*testContext, *gomock.Controller) {
 	mctrl := gomock.NewController(t)
-	inmemQ := inmemory.NewInMemQueue(lockTime)
+	inmemQ := inmemory.NewInMemQueue("test", lockTime)
 	return &testContext{
 		ctx:       context.Background(),
 		mockSC:    store.NewMockStorageClient(mctrl),

pkg/armrpc/hostoptions/providerconfig.go

@@ -17,12 +17,16 @@ limitations under the License.
 package hostoptions
 
 import (
+	"fmt"
+
 	metricsprovider "github.com/radius-project/radius/pkg/metrics/provider"
 	profilerprovider "github.com/radius-project/radius/pkg/profiler/provider"
 	"github.com/radius-project/radius/pkg/trace"
 	"github.com/radius-project/radius/pkg/ucp/config"
 	"github.com/radius-project/radius/pkg/ucp/dataprovider"
+
 	qprovider "github.com/radius-project/radius/pkg/ucp/queue/provider"
+
 	sprovider "github.com/radius-project/radius/pkg/ucp/secret/provider"
 	"github.com/radius-project/radius/pkg/ucp/ucplog"
 )
@@ -59,6 +63,18 @@ type ServerOptions struct {
 	ArmMetadataEndpoint string `yaml:"armMetadataEndpoint,omitempty"`
 	// EnableAuth when set the arm client authetication will be performed
 	EnableArmAuth bool `yaml:"enableArmAuth,omitempty"`
+
+	// TLSCertificateDirectory is the directory where the TLS certificates are stored.
+	//
+	// The server code will expect to find the following files in this directory:
+	// - tls.crt: The server's certificate.
+	// - tls.key: The server's private key.
+	TLSCertificateDirectory string `yaml:"tlsCertificateDirectory,omitempty"`
+}
+
+// Address returns the address of the server in host:port format.
+func (s ServerOptions) Address() string {
+	return s.Host + ":" + fmt.Sprint(s.Port)
 }
 
 // WorkerServerOptions includes the worker server options.