chore: simple in memory provisioning scheduler (#2815)

Rough domain model for provisioning deployments. 
- `Task` is a single executable task to bring the infra from one state
to another
- `Deployment` is a list of sequentially executable tasks
- `Provisioner` is an interface for calling provisioners, wrapping calls
to provisioner plugins

This is a WIP, and there are a lot of `TODO`s around, but I wanted to
get this out to keep PRs small.

---------

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

backend/provisioner/deployment/deployment.go

@@ -0,0 +1,133 @@
+package deployment
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1beta1/provisioner"
+)
+
+type TaskState string
+
+const (
+ TaskStatePending TaskState = ""
+ TaskStateRunning TaskState = "running"
+ TaskStateDone TaskState = "done"
+ TaskStateFailed TaskState = "failed"
+)
+
+// Task is a unit of work for a deployment
+type Task struct {
+ Handler Provisioner
+ Module string
+ State TaskState
+ Desired []*provisioner.Resource
+ Existing []*provisioner.Resource
+ // populated only when the task is done
+ Output []*provisioner.Resource
+
+ // set if the task is currently running
+ RunningToken string
+}
+
+func (t *Task) Start(ctx context.Context) error {
+ if t.State != TaskStatePending {
+ return fmt.Errorf("task state is not pending: %s", t.State)
+ }
+ t.State = TaskStateRunning
+ token, err := t.Handler.Provision(ctx, t.Module, t.constructResourceContext(t.Desired), t.Existing)
+ if err != nil {
+ t.State = TaskStateFailed
+ return fmt.Errorf("error provisioning resources: %w", err)
+ }
+ if token == "" {
+ // no changes
+ t.State = TaskStateDone
+ t.Output = t.Desired
+ }
+ t.RunningToken = token
+ return nil
+}
+
+func (t *Task) constructResourceContext(r []*provisioner.Resource) []*provisioner.ResourceContext {
+ result := make([]*provisioner.ResourceContext, len(r))
+ for i, res := range r {
+ result[i] = &provisioner.ResourceContext{
+ Resource: res,
+ // TODO: Collect previously constructed resources from a dependency graph here
+ }
+ }
+ return result
+}
+
+func (t *Task) Progress(ctx context.Context) error {
+ if t.State != TaskStateRunning {
+ return fmt.Errorf("task state is not running: %s", t.State)
+ }
+ state, output, err := t.Handler.State(ctx, t.RunningToken, t.Desired)
+ if err != nil {
+ return fmt.Errorf("error getting state: %w", err)
+ }
+ if state == TaskStateDone {
+ t.State = TaskStateDone
+ t.Output = output
+ }
+ return nil
+}
+
+// Deployment is a single deployment of resources for a single module
+type Deployment struct {
+ Module string
+ Tasks []*Task
+}
+
+// next running or pending task. Nil if all tasks are done.
+func (d *Deployment) next() *Task {
+ for _, t := range d.Tasks {
+ if t.State == TaskStatePending || t.State == TaskStateRunning || t.State == TaskStateFailed {
+ return t
+ }
+ }
+ return nil
+}
+
+// Progress the deployment. Returns true if there are still tasks running or pending.
+func (d *Deployment) Progress(ctx context.Context) (bool, error) {
+ next := d.next()
+ if next == nil {
+ return false, nil
+ }
+
+ if next.State == TaskStatePending {
+ err := next.Start(ctx)
+ if err != nil {
+ return true, err
+ }
+ }
+ err := next.Progress(ctx)
+ return d.next() != nil, err
+}
+
+type DeploymentState struct {
+ Pending []*Task
+ Running *Task
+ Failed *Task
+ Done []*Task
+}
+
+func (d *Deployment) State() *DeploymentState {
+ result := &DeploymentState{}
+ for _, t := range d.Tasks {
+ switch t.State {
+ case TaskStatePending:
+ result.Pending = append(result.Pending, t)
+ case TaskStateRunning:
+ result.Running = t
+ case TaskStateFailed:
+ result.Failed = t
+ case TaskStateDone:
+ result.Done = append(result.Done, t)
+ }
+ }
+ return result
+}

backend/provisioner/deployment/deployment_test.go

@@ -0,0 +1,92 @@
+package deployment_test
+
+import (
+ "context"
+ "testing"
+
+ "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1beta1/provisioner"
+ "github.com/TBD54566975/ftl/backend/provisioner/deployment"
+ "github.com/alecthomas/assert/v2"
+)
+
+// MockProvisioner is a mock implementation of the Provisioner interface
+type MockProvisioner struct {
+ Token string
+ stateCalls int
+}
+
+var _ deployment.Provisioner = (*MockProvisioner)(nil)
+
+func (m *MockProvisioner) Provision(
+ ctx context.Context,
+ module string,
+ desired []*provisioner.ResourceContext,
+ existing []*provisioner.Resource,
+) (string, error) {
+ return m.Token, nil
+}
+
+func (m *MockProvisioner) State(
+ ctx context.Context,
+ token string,
+ desired []*provisioner.Resource,
+) (deployment.TaskState, []*provisioner.Resource, error) {
+ m.stateCalls++
+ if m.stateCalls <= 1 {
+ return deployment.TaskStateRunning, nil, nil
+ }
+ return deployment.TaskStateDone, desired, nil
+}
+
+func TestDeployment_Progress(t *testing.T) {
+ ctx := context.Background()
+
+ t.Run("no tasks", func(t *testing.T) {
+ deployment := &deployment.Deployment{}
+ progress, err := deployment.Progress(ctx)
+ assert.NoError(t, err)
+ assert.False(t, progress)
+ })
+
+ t.Run("progresses each provisioner in order", func(t *testing.T) {
+ registry := deployment.ProvisionerRegistry{}
+ registry.Register(&MockProvisioner{Token: "foo"}, deployment.ResourceTypePostgres)
+ registry.Register(&MockProvisioner{Token: "bar"}, deployment.ResourceTypeMysql)
+
+ dpl := registry.CreateDeployment(
+ "test-module",
+ []*provisioner.Resource{{
+ ResourceId: "a",
+ Resource: &provisioner.Resource_Mysql{},
+ }, {
+ ResourceId: "b",
+ Resource: &provisioner.Resource_Postgres{},
+ }},
+ []*provisioner.Resource{},
+ )
+
+ assert.Equal(t, 2, len(dpl.State().Pending))
+
+ _, err := dpl.Progress(ctx)
+ assert.NoError(t, err)
+ assert.Equal(t, 1, len(dpl.State().Pending))
+ assert.NotZero(t, dpl.State().Running)
+
+ _, err = dpl.Progress(ctx)
+ assert.NoError(t, err)
+ assert.Equal(t, 1, len(dpl.State().Pending))
+ assert.Zero(t, dpl.State().Running)
+ assert.Equal(t, 1, len(dpl.State().Done))
+
+ _, err = dpl.Progress(ctx)
+ assert.NoError(t, err)
+ assert.Equal(t, 0, len(dpl.State().Pending))
+ assert.NotZero(t, dpl.State().Running)
+ assert.Equal(t, 1, len(dpl.State().Done))
+
+ running, err := dpl.Progress(ctx)
+ assert.NoError(t, err)
+ assert.Equal(t, 2, len(dpl.State().Done))
+ assert.False(t, running)
+ })
+}

backend/provisioner/deployment/provisioner.go

@@ -0,0 +1,171 @@
+package deployment
+
+import (
+ "context"
+ "fmt"
+
+ "connectrpc.com/connect"
+
+ "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1beta1/provisioner"
+ "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1beta1/provisioner/provisionerconnect"
+ "github.com/TBD54566975/ftl/backend/schema"
+ "github.com/TBD54566975/ftl/common/plugin"
+ "github.com/TBD54566975/ftl/internal/log"
+)
+
+// ResourceType is a type of resource used to configure provisioners
+type ResourceType string
+
+const (
+ ResourceTypeUnknown ResourceType = "unknown"
+ ResourceTypePostgres ResourceType = "postgres"
+ ResourceTypeMysql ResourceType = "mysql"
+)
+
+// Provisioner is a runnable process to provision resources
+type Provisioner interface {
+ Provision(ctx context.Context, module string, desired []*provisioner.ResourceContext, existing []*provisioner.Resource) (string, error)
+ State(ctx context.Context, token string, desired []*provisioner.Resource) (TaskState, []*provisioner.Resource, error)
+}
+
+type provisionerConfig struct {
+ provisioner Provisioner
+ types []ResourceType
+}
+
+// ProvisionerRegistry contains all known resource handlers in the order they should be executed
+type ProvisionerRegistry struct {
+ Provisioners []*provisionerConfig
+}
+
+// Register to the registry, to be executed after all the previously added handlers
+func (reg *ProvisionerRegistry) Register(handler Provisioner, types ...ResourceType) {
+ reg.Provisioners = append(reg.Provisioners, &provisionerConfig{
+ provisioner: handler,
+ types: types,
+ })
+}
+
+// CreateDeployment to take the system to the desired state
+func (reg *ProvisionerRegistry) CreateDeployment(module string, desiredResources, existingResources []*provisioner.Resource) *Deployment {
+ var result []*Task
+
+ existingByHandler := reg.groupByProvisioner(existingResources)
+ desiredByHandler := reg.groupByProvisioner(desiredResources)
+
+ for handler, desired := range desiredByHandler {
+ existing := existingByHandler[handler]
+ result = append(result, &Task{
+ Handler: handler,
+ Desired: desired,
+ Existing: existing,
+ })
+ }
+ return &Deployment{Tasks: result, Module: module}
+}
+
+// ExtractResources from a module schema
+func ExtractResources(sch *schema.Module) ([]*provisioner.Resource, error) {
+ var result []*provisioner.Resource
+ for _, decl := range sch.Decls {
+ if db, ok := decl.(*schema.Database); ok {
+ if db.Type == "postgres" {
+ result = append(result, &provisioner.Resource{
+ ResourceId: decl.GetName(),
+ Resource: &provisioner.Resource_Postgres{},
+ })
+ } else if db.Type == "mysql" {
+ result = append(result, &provisioner.Resource{
+ ResourceId: decl.GetName(),
+ Resource: &provisioner.Resource_Mysql{},
+ })
+ } else {
+ return nil, fmt.Errorf("unknown db type: %s", db.Type)
+ }
+ }
+ }
+ return result, nil
+}
+
+func (reg *ProvisionerRegistry) groupByProvisioner(resources []*provisioner.Resource) map[Provisioner][]*provisioner.Resource {
+ result := map[Provisioner][]*provisioner.Resource{}
+ for _, r := range resources {
+ for _, cfg := range reg.Provisioners {
+ for _, t := range cfg.types {
+ typed := typeOf(r)
+ if t == typed {
+ result[cfg.provisioner] = append(result[cfg.provisioner], r)
+ break
+ }
+ }
+ }
+ }
+ return result
+}
+
+func typeOf(r *provisioner.Resource) ResourceType {
+ if _, ok := r.Resource.(*provisioner.Resource_Mysql); ok {
+ return ResourceTypeMysql
+ } else if _, ok := r.Resource.(*provisioner.Resource_Postgres); ok {
+ return ResourceTypePostgres
+ }
+ return ResourceTypeUnknown
+}
+
+// PluginProvisioner delegates provisioning to an external plugin
+type PluginProvisioner struct {
+ cmdCtx context.Context
+ client *plugin.Plugin[provisionerconnect.ProvisionerPluginServiceClient]
+}
+
+func NewPluginProvisioner(ctx context.Context, name, dir, exe string) (*PluginProvisioner, error) {
+ client, cmdCtx, err := plugin.Spawn(
+ ctx,
+ log.Debug,
+ name,
+ dir,
+ exe,
+ provisionerconnect.NewProvisionerPluginServiceClient,
+ )
+ if err != nil {
+ return nil, fmt.Errorf("error spawning plugin: %w", err)
+ }
+
+ return &PluginProvisioner{
+ cmdCtx: cmdCtx,
+ client: client,
+ }, nil
+}
+
+func (p *PluginProvisioner) Provision(ctx context.Context, module string, desired []*provisioner.ResourceContext, existing []*provisioner.Resource) (string, error) {
+ resp, err := p.client.Client.Provision(ctx, connect.NewRequest(&provisioner.ProvisionRequest{
+ DesiredResources: desired,
+ ExistingResources: existing,
+ FtlClusterId: "ftl",
+ Module: module,
+ }))
+ if err != nil {
+ return "", fmt.Errorf("error calling plugin: %w", err)
+ }
+ if resp.Msg.Status != provisioner.ProvisionResponse_SUBMITTED {
+ return resp.Msg.ProvisioningToken, nil
+ }
+ return "", nil
+}
+
+func (p *PluginProvisioner) State(ctx context.Context, token string, desired []*provisioner.Resource) (TaskState, []*provisioner.Resource, error) {
+ resp, err := p.client.Client.Status(ctx, connect.NewRequest(&provisioner.StatusRequest{
+ ProvisioningToken: token,
+ }))
+ if err != nil {
+ return "", nil, fmt.Errorf("error getting status from plugin: %w", err)
+ }
+ if failed, ok := resp.Msg.Status.(*provisioner.StatusResponse_Failed); ok {
+ return TaskStateFailed, nil, fmt.Errorf("provisioning failed: %s", failed.Failed.ErrorMessage)
+ } else if success, ok := resp.Msg.Status.(*provisioner.StatusResponse_Success); ok {
+ return TaskStateDone, success.Success.UpdatedResources, nil
+ }
+ return TaskStateRunning, nil, nil
+}
+
+var _ Provisioner = (*PluginProvisioner)(nil)