feat: Provisioner for FTL modules (#3050)

Apologies, this grow a bit bigger than I hoped for.

Major parts in this PR:
- a new in memory provisioner for creating modules by calling
`ftl-controller`
- an in memory resource graph to represent that modules can depend on
databases. The direct dependants are passed to the provisioner when
provisioning, with outputs filled in from previous runs. Eventually,
this graph will be saved to DB

Next: pass the new database DSNs to the controller at deployment.

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

backend/protos/xyz/block/ftl/v1beta1/provisioner/resource.proto

@@ -2,6 +2,10 @@ syntax = "proto3";
 
 package xyz.block.ftl.v1beta1.provisioner;
 
+import "google/protobuf/struct.proto";
+import "xyz/block/ftl/v1/ftl.proto";
+import "xyz/block/ftl/v1/schema/schema.proto";
+
 option go_package = "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1beta1/provisioner;provisioner";
 option java_multiple_files = true;
 
@@ -13,6 +17,7 @@ message Resource {
   oneof resource {
     PostgresResource postgres = 102;
     MysqlResource mysql = 103;
+    ModuleResource module = 104;
   }
 }
 
@@ -35,3 +40,15 @@ message MysqlResource {
   }
   MysqlResourceOutput output = 1;
 }
+
+message ModuleResource {
+  message ModuleResourceOutput {
+    string deployment_key = 1;
+  }
+  ModuleResourceOutput output = 1;
+
+  xyz.block.ftl.v1.schema.Module schema = 2;
+  repeated xyz.block.ftl.v1.DeploymentArtefact artefacts = 3;
+  // Runner labels required to run this deployment.
+  optional google.protobuf.Struct labels = 4;
+}

backend/provisioner/controller_provisioner.go

@@ -0,0 +1,41 @@
+package provisioner
+
+import (
+	"context"
+	"fmt"
+
+	"connectrpc.com/connect"
+
+	ftlv1 "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1"
+	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
+	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1beta1/provisioner"
+	"github.com/TBD54566975/ftl/internal/log"
+)
+
+// NewControllerProvisioner creates a new provisioner that uses the FTL controller to provision modules
+func NewControllerProvisioner(client ftlv1connect.ControllerServiceClient) *InMemProvisioner {
+	return NewEmbeddedProvisioner(map[ResourceType]InMemResourceProvisionerFn{
+		ResourceTypeModule: func(ctx context.Context, rc *provisioner.ResourceContext, module, _ string) (*provisioner.Resource, error) {
+			mod, ok := rc.Resource.Resource.(*provisioner.Resource_Module)
+			if !ok {
+				panic(fmt.Errorf("unexpected resource type: %T", rc.Resource.Resource))
+			}
+			logger := log.FromContext(ctx)
+			logger.Infof("provisioning module: %s", module)
+
+			resp, err := client.CreateDeployment(ctx, connect.NewRequest(&ftlv1.CreateDeploymentRequest{
+				Schema:    mod.Module.Schema,
+				Artefacts: mod.Module.Artefacts,
+				Labels:    mod.Module.Labels,
+			}))
+			if err != nil {
+				return nil, fmt.Errorf("failed to create deployment: %w", err)
+			}
+			if mod.Module.Output == nil {
+				mod.Module.Output = &provisioner.ModuleResource_ModuleResourceOutput{}
+			}
+			mod.Module.Output.DeploymentKey = resp.Msg.DeploymentKey
+			return rc.Resource, nil
+		},
+	})
+}

backend/provisioner/deployment.go

@@ -25,8 +25,8 @@ type Task struct {
 	handler  provisionerconnect.ProvisionerPluginServiceClient
 	module   string
 	state    TaskState
-	desired  []*provisioner.Resource
-	existing []*provisioner.Resource
+	desired  *ResourceGraph
+	existing *ResourceGraph
 	// populated only when the task is done
 	output []*provisioner.Resource
 
@@ -44,7 +44,7 @@ func (t *Task) Start(ctx context.Context) error {
 		Module: t.module,
 		// TODO: We need a proper cluster specific ID here
 		FtlClusterId:      "ftl",
-		ExistingResources: t.existing,
+		ExistingResources: t.existing.Roots(),
 		DesiredResources:  t.constructResourceContext(t.desired),
 	}))
 	if err != nil {
@@ -53,18 +53,18 @@ func (t *Task) Start(ctx context.Context) error {
 	}
 	if resp.Msg.Status == provisioner.ProvisionResponse_NO_CHANGES {
 		t.state = TaskStateDone
-		t.output = t.desired
+		t.output = t.desired.Resources()
 	}
 	t.runningToken = resp.Msg.ProvisioningToken
 	return nil
 }
 
-func (t *Task) constructResourceContext(r []*provisioner.Resource) []*provisioner.ResourceContext {
-	result := make([]*provisioner.ResourceContext, len(r))
-	for i, res := range r {
+func (t *Task) constructResourceContext(r *ResourceGraph) []*provisioner.ResourceContext {
+	result := make([]*provisioner.ResourceContext, len(r.Roots()))
+	for i, res := range r.Roots() {
 		result[i] = &provisioner.ResourceContext{
-			Resource: res,
-			// TODO: Collect previously constructed resources from a dependency graph here
+			Resource:     res,
+			Dependencies: r.Dependencies(res),
 		}
 	}
 	return result
@@ -77,7 +77,7 @@ func (t *Task) Progress(ctx context.Context) error {
 
 	resp, err := t.handler.Status(ctx, connect.NewRequest(&provisioner.StatusRequest{
 		ProvisioningToken: t.runningToken,
-		DesiredResources:  t.desired,
+		DesiredResources:  t.desired.Resources(),
 	}))
 	if err != nil {
 		t.state = TaskStateFailed

backend/provisioner/deployment_test.go

@@ -69,17 +69,11 @@ func TestDeployment_Progress(t *testing.T) {
 		registry.Register(&MockProvisioner{Token: "foo"}, provisioner.ResourceTypePostgres)
 		registry.Register(&MockProvisioner{Token: "bar"}, provisioner.ResourceTypeMysql)
 
-		dpl := registry.CreateDeployment(
-			"test-module",
-			[]*proto.Resource{{
-				ResourceId: "a",
-				Resource:   &proto.Resource_Mysql{},
-			}, {
-				ResourceId: "b",
-				Resource:   &proto.Resource_Postgres{},
-			}},
-			[]*proto.Resource{},
-		)
+		graph := &provisioner.ResourceGraph{}
+		graph.AddNode(&proto.Resource{ResourceId: "a", Resource: &proto.Resource_Mysql{}})
+		graph.AddNode(&proto.Resource{ResourceId: "b", Resource: &proto.Resource_Postgres{}})
+
+		dpl := registry.CreateDeployment("test-module", graph, &provisioner.ResourceGraph{})
 
 		assert.Equal(t, 2, len(dpl.State().Pending))