chore: runtime deployment key as key.Deployment (#4017)

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
block · Jan 14, 2025 · 1b9b1f4 · 1b9b1f4
1 parent 0f198b9
commit 1b9b1f4
Show file tree

Hide file tree

Showing 32 changed files with 108 additions and 53 deletions.
diff --git a/backend/admin/service.go b/backend/admin/service.go
@@ -354,7 +354,7 @@ func (s *AdminService) ResetSubscription(ctx context.Context, req *connect.Reque
 	if verb.Runtime == nil || verb.Runtime.Subscription == nil || len(verb.Runtime.Subscription.KafkaBrokers) == 0 {
 		return nil, fmt.Errorf("no Kafka brokers for subscription %q", req.Msg.Subscription)
 	}
-	if module.GetRuntime().GetDeployment().GetDeploymentKey() == "" {
+	if module.GetRuntime().GetDeployment().GetDeploymentKey().IsZero() {
 		return nil, fmt.Errorf("no deployment for module %s", req.Msg.Subscription.Module)
 	}
 	topicID := subscriber.Topic.String()

diff --git a/backend/console/console.go b/backend/console/console.go
@@ -117,7 +117,7 @@ func (s *service) GetModules(ctx context.Context, req *connect.Request[consolepb
 	nilMap := map[schema.RefKey]map[schema.RefKey]bool{}
 	var modules []*consolepb.Module
 	for _, mod := range sch.Modules {
-		if mod.GetRuntime().GetDeployment().GetDeploymentKey() == "" {
+		if mod.GetRuntime().GetDeployment().GetDeploymentKey().IsZero() {
 			continue
 		}
 		var verbs []*consolepb.Verb
@@ -149,7 +149,7 @@ func (s *service) GetModules(ctx context.Context, req *connect.Request[consolepb
 
 		modules = append(modules, &consolepb.Module{
 			Name:          mod.Name,
-			DeploymentKey: mod.Runtime.Deployment.DeploymentKey,
+			DeploymentKey: mod.Runtime.Deployment.DeploymentKey.String(),
 			Language:      mod.Runtime.Base.Language,
 			Verbs:         verbs,
 			Data:          data,
@@ -186,7 +186,7 @@ func moduleFromDeployment(deployment *schema.Module, sch *schema.Schema, refMap
 	}
 
 	module.Name = deployment.Name
-	module.DeploymentKey = deployment.Runtime.Deployment.DeploymentKey
+	module.DeploymentKey = deployment.Runtime.Deployment.DeploymentKey.String()
 	module.Language = deployment.Runtime.Base.Language
 	module.Schema = deployment.String()
 
@@ -412,7 +412,7 @@ func (s *service) sendStreamModulesResp(ctx context.Context, stream *connect.Ser
 
 	var modules []*consolepb.Module
 	for _, deployment := range deployments {
-		if deployment.GetRuntime().GetDeployment().GetDeploymentKey() == "" {
+		if deployment.GetRuntime().GetDeployment().GetDeploymentKey().IsZero() {
 			continue
 		}
 		module, err := moduleFromDeployment(deployment, sch, refMap)

diff --git a/backend/controller/controller.go b/backend/controller/controller.go
@@ -285,8 +285,8 @@ func (s *Service) Status(ctx context.Context, req *connect.Request[ftlv1.StatusR
 	routes := slices.Map(allModules.Schema().Modules, func(module *schema.Module) (out *ftlv1.StatusResponse_Route) {
 		key := ""
 		endpoint := ""
-		if module.GetRuntime().GetDeployment().GetDeploymentKey() != "" {
-			key = module.GetRuntime().GetDeployment().DeploymentKey
+		if dkey := module.GetRuntime().GetDeployment().GetDeploymentKey(); !dkey.IsZero() {
+			key = dkey.String()
 			endpoint = module.GetRuntime().GetDeployment().Endpoint
 		}
 		return &ftlv1.StatusResponse_Route{
@@ -666,7 +666,7 @@ func (s *Service) GetDeploymentContext(ctx context.Context, req *connect.Request
 				routeTable[deployment.String()] = route.String()
 			}
 		}
-		if deployment.Schema.GetRuntime().GetDeployment().GetDeploymentKey() != "" {
+		if !deployment.Schema.GetRuntime().GetDeployment().GetDeploymentKey().IsZero() {
 			routeTable[deployment.Key.String()] = deployment.Schema.Runtime.Deployment.Endpoint
 		}
 

diff --git a/backend/controller/schemaservice.go b/backend/controller/schemaservice.go
@@ -50,7 +50,10 @@ func (s *Service) UpdateDeploymentRuntime(ctx context.Context, req *connect.Requ
 	if module.Runtime == nil {
 		module.Runtime = &schema.ModuleRuntime{}
 	}
-	event := schema.ModuleRuntimeEventFromProto(req.Msg.Event)
+	event, err := schema.ModuleRuntimeEventFromProto(req.Msg.Event)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing module runtime event: %w", err)
+	}
 	module.Runtime.ApplyEvent(event)
 	err = s.schemaState.Publish(ctx, &state.DeploymentSchemaUpdatedEvent{
 		Key:    deployment,

diff --git a/backend/cron/service.go b/backend/cron/service.go
@@ -185,7 +185,7 @@ func rebuildQueue(cronJobs map[string][]cronJob) []cronJob {
 }
 
 func extractCronJobs(module *schema.Module) ([]cronJob, error) {
-	if module.GetRuntime().GetDeployment().GetDeploymentKey() == "" {
+	if module.GetRuntime().GetDeployment().GetDeploymentKey().IsZero() {
 		return nil, nil
 	}
 	cronJobs := []cronJob{}
@@ -202,13 +202,9 @@ func extractCronJobs(module *schema.Module) ([]cronJob, error) {
 		if err != nil {
 			return nil, fmt.Errorf("%s: %w", cronmd.Pos, err)
 		}
-		deploymentKey, err := key.ParseDeploymentKey(module.Runtime.Deployment.DeploymentKey)
-		if err != nil {
-			return nil, fmt.Errorf("%s: %w", cronmd.Pos, err)
-		}
 		cronJobs = append(cronJobs, cronJob{
 			module:     module.Name,
-			deployment: deploymentKey,
+			deployment: module.Runtime.Deployment.DeploymentKey,
 			verb:       verb,
 			cronmd:     cronmd,
 			pattern:    pattern,

diff --git a/backend/cron/service_test.go b/backend/cron/service_test.go
@@ -41,7 +41,7 @@ func TestCron(t *testing.T) {
 		Name: "echo",
 		Runtime: &schema.ModuleRuntime{
 			Deployment: &schema.ModuleRuntimeDeployment{
-				DeploymentKey: key.NewDeploymentKey("echo").String(),
+				DeploymentKey: key.NewDeploymentKey("echo"),
 			},
 		},
 		Decls: []schema.Decl{

diff --git a/backend/provisioner/controller_provisioner.go b/backend/provisioner/controller_provisioner.go
@@ -9,6 +9,7 @@ import (
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
 	"github.com/block/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
 	"github.com/block/ftl/common/schema"
+	"github.com/block/ftl/internal/key"
 	"github.com/block/ftl/internal/log"
 )
 
@@ -30,10 +31,14 @@ func NewControllerProvisioner(client ftlv1connect.ControllerServiceClient) *InMe
 			if err != nil {
 				return nil, fmt.Errorf("failed to create deployment: %w", err)
 			}
+			deploymentKey, err := key.ParseDeploymentKey(resp.Msg.DeploymentKey)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse deployment key: %w", err)
+			}
 
 			return &RuntimeEvent{
 				Module: &schema.ModuleRuntimeDeployment{
-					DeploymentKey: resp.Msg.DeploymentKey,
+					DeploymentKey: deploymentKey,
 				},
 			}, nil
 		},

diff --git a/backend/provisioner/deployment.go b/backend/provisioner/deployment.go
@@ -91,7 +91,10 @@ func (t *Task) Progress(ctx context.Context) error {
 			for _, event := range events {
 				switch event.Value.(type) {
 				case *provisioner.ProvisioningEvent_ModuleRuntimeEvent:
-					moduleEvent := schema.ModuleRuntimeEventFromProto(event.GetModuleRuntimeEvent())
+					moduleEvent, err := schema.ModuleRuntimeEventFromProto(event.GetModuleRuntimeEvent())
+					if err != nil {
+						return fmt.Errorf("error parsing module runtime event: %w", err)
+					}
 					module.Runtime.ApplyEvent(moduleEvent)
 
 				case *provisioner.ProvisioningEvent_DatabaseRuntimeEvent:

diff --git a/backend/provisioner/runner_scaling_provisioner.go b/backend/provisioner/runner_scaling_provisioner.go
@@ -35,7 +35,7 @@ func provisionRunner(scaling scaling.RunnerScaling) InMemResourceProvisionerFn {
 		}
 
 		deployment := module.Runtime.Deployment.DeploymentKey
-		if deployment == "" {
+		if deployment.IsZero() {
 			return nil, fmt.Errorf("failed to find deployment for runner")
 		}
 		logger.Debugf("provisioning runner: %s.%s for deployment %s", module, rc.ResourceID(), deployment)
@@ -56,11 +56,11 @@ func provisionRunner(scaling scaling.RunnerScaling) InMemResourceProvisionerFn {
 				}
 			}
 		}
-		if err := scaling.StartDeployment(ctx, module.Name, deployment, module, cron, http); err != nil {
+		if err := scaling.StartDeployment(ctx, module.Name, deployment.String(), module, cron, http); err != nil {
 			logger.Infof("failed to start deployment: %v", err)
 			return nil, fmt.Errorf("failed to start deployment: %w", err)
 		}
-		endpoint, err := scaling.GetEndpointForDeployment(ctx, module.Name, deployment)
+		endpoint, err := scaling.GetEndpointForDeployment(ctx, module.Name, deployment.String())
 		if err != nil || !endpoint.Ok() {
 			return nil, fmt.Errorf("failed to get endpoint for deployment: %w", err)
 		}
@@ -88,7 +88,7 @@ func provisionRunner(scaling scaling.RunnerScaling) InMemResourceProvisionerFn {
 		schemaClient := rpc.ClientFromContext[ftlv1connect.SchemaServiceClient](ctx)
 		controllerClient := rpc.ClientFromContext[ftlv1connect.ControllerServiceClient](ctx)
 
-		deps, err := scaling.TerminatePreviousDeployments(ctx, module.Name, deployment)
+		deps, err := scaling.TerminatePreviousDeployments(ctx, module.Name, deployment.String())
 		if err != nil {
 			logger.Errorf(err, "failed to terminate previous deployments")
 		} else {
@@ -102,7 +102,7 @@ func provisionRunner(scaling scaling.RunnerScaling) InMemResourceProvisionerFn {
 		}
 
 		logger.Debugf("updating module runtime for %s with endpoint %s", module, endpointURI)
-		_, err = schemaClient.UpdateDeploymentRuntime(ctx, connect.NewRequest(&ftlv1.UpdateDeploymentRuntimeRequest{Deployment: deployment, Event: &schemapb.ModuleRuntimeEvent{Value: &schemapb.ModuleRuntimeEvent_ModuleRuntimeDeployment{ModuleRuntimeDeployment: &schemapb.ModuleRuntimeDeployment{DeploymentKey: deployment, Endpoint: endpointURI}}}}))
+		_, err = schemaClient.UpdateDeploymentRuntime(ctx, connect.NewRequest(&ftlv1.UpdateDeploymentRuntimeRequest{Deployment: deployment.String(), Event: &schemapb.ModuleRuntimeEvent{Value: &schemapb.ModuleRuntimeEvent_ModuleRuntimeDeployment{ModuleRuntimeDeployment: &schemapb.ModuleRuntimeDeployment{DeploymentKey: deployment.String(), Endpoint: endpointURI}}}}))
 		if err != nil {
 			return nil, fmt.Errorf("failed to update module runtime: %w", err)
 		}

diff --git a/backend/provisioner/service.go b/backend/provisioner/service.go
@@ -95,7 +95,7 @@ func (s *Service) CreateDeployment(ctx context.Context, req *connect.Request[ftl
 
 	deploymentKey := deployment.Module.Runtime.Deployment.DeploymentKey
 	return connect.NewResponse(&ftlv1.CreateDeploymentResponse{
-		DeploymentKey: deploymentKey,
+		DeploymentKey: deploymentKey.String(),
 	}), nil
 }
 

diff --git a/common/schema/go2proto.to.go b/common/schema/go2proto.to.go
diff --git a/common/schema/module.go b/common/schema/module.go
@@ -262,13 +262,17 @@ func ModuleFromProtoFile(filename string) (*Module, error) {
 
 // ModuleFromProto converts a protobuf Module to a Module and validates it.
 func ModuleFromProto(s *schemapb.Module) (*Module, error) {
+	runtime, err := ModuleRuntimeFromProto(s.Runtime)
+	if err != nil {
+		return nil, err
+	}
 	module := &Module{
 		Pos:      PosFromProto(s.Pos),
 		Builtin:  s.Builtin,
 		Name:     s.Name,
 		Comments: s.Comments,
 		Decls:    declListToSchema(s.Decls),
-		Runtime:  ModuleRuntimeFromProto(s.Runtime),
+		Runtime:  runtime,
 		Metadata: metadataListToSchema(s.Metadata),
 	}
 	return module, ValidateModule(module)

diff --git a/common/schema/moduleruntime.go b/common/schema/moduleruntime.go
@@ -5,6 +5,7 @@ import (
 	"time"
 
 	schemapb "github.com/block/ftl/common/protos/xyz/block/ftl/schema/v1"
+	"github.com/block/ftl/internal/key"
 )
 
 // ModuleRuntime is runtime configuration for a module that can be dynamically updated.
@@ -26,24 +27,28 @@ func (m *ModuleRuntime) ApplyEvent(event ModuleRuntimeEvent) {
 	}
 }
 
-func ModuleRuntimeFromProto(s *schemapb.ModuleRuntime) *ModuleRuntime {
+func ModuleRuntimeFromProto(s *schemapb.ModuleRuntime) (*ModuleRuntime, error) {
 	if s == nil {
-		return nil
+		return nil, nil
+	}
+	deployment, err := ModuleRuntimeDeploymentFromProto(s.Deployment)
+	if err != nil {
+		return nil, err
 	}
 	return &ModuleRuntime{
 		Base:       *ModuleRuntimeBaseFromProto(s.Base),
 		Scaling:    ModuleRuntimeScalingFromProto(s.Scaling),
-		Deployment: ModuleRuntimeDeploymentFromProto(s.Deployment),
-	}
+		Deployment: deployment,
+	}, nil
 }
 
-func ModuleRuntimeEventFromProto(s *schemapb.ModuleRuntimeEvent) ModuleRuntimeEvent {
+func ModuleRuntimeEventFromProto(s *schemapb.ModuleRuntimeEvent) (ModuleRuntimeEvent, error) {
 	switch s.Value.(type) {
 	case *schemapb.ModuleRuntimeEvent_ModuleRuntimeBase:
-		return ModuleRuntimeBaseFromProto(s.GetModuleRuntimeBase())
+		return ModuleRuntimeBaseFromProto(s.GetModuleRuntimeBase()), nil
 
 	case *schemapb.ModuleRuntimeEvent_ModuleRuntimeScaling:
-		return ModuleRuntimeScalingFromProto(s.GetModuleRuntimeScaling())
+		return ModuleRuntimeScalingFromProto(s.GetModuleRuntimeScaling()), nil
 
 	case *schemapb.ModuleRuntimeEvent_ModuleRuntimeDeployment:
 		return ModuleRuntimeDeploymentFromProto(s.GetModuleRuntimeDeployment())
@@ -111,24 +116,32 @@ func ModuleRuntimeScalingFromProto(s *schemapb.ModuleRuntimeScaling) *ModuleRunt
 //protobuf:3 RuntimeEvent
 type ModuleRuntimeDeployment struct {
 	// Endpoint is the endpoint of the deployed module.
-	Endpoint      string    `protobuf:"1"`
-	DeploymentKey string    `protobuf:"2"`
-	CreatedAt     time.Time `protobuf:"3"`
-	ActivatedAt   time.Time `protobuf:"4"`
+	Endpoint      string         `protobuf:"1"`
+	DeploymentKey key.Deployment `protobuf:"2"`
+	CreatedAt     time.Time      `protobuf:"3"`
+	ActivatedAt   time.Time      `protobuf:"4"`
 }
 
 func (m *ModuleRuntimeDeployment) moduleRuntime() {}
 
 func (m *ModuleRuntimeDeployment) runtimeEvent() {}
 
-func ModuleRuntimeDeploymentFromProto(s *schemapb.ModuleRuntimeDeployment) *ModuleRuntimeDeployment {
+func ModuleRuntimeDeploymentFromProto(s *schemapb.ModuleRuntimeDeployment) (*ModuleRuntimeDeployment, error) {
 	if s == nil {
-		return nil
+		return nil, nil
+	}
+	deploymentKey := key.Deployment{}
+	if s.DeploymentKey != "" {
+		dk, err := key.ParseDeploymentKey(s.DeploymentKey)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing deployment key: %w", err)
+		}
+		deploymentKey = dk
 	}
 	return &ModuleRuntimeDeployment{
 		Endpoint:      s.Endpoint,
-		DeploymentKey: s.DeploymentKey,
-	}
+		DeploymentKey: deploymentKey,
+	}, nil
 }
 
 func (m *ModuleRuntime) GetScaling() *ModuleRuntimeScaling {
@@ -166,9 +179,9 @@ func (m *ModuleRuntimeDeployment) GetCreatedAt() time.Time {
 	return m.CreatedAt
 }
 
-func (m *ModuleRuntimeDeployment) GetDeploymentKey() string {
+func (m *ModuleRuntimeDeployment) GetDeploymentKey() key.Deployment {
 	if m == nil {
-		return ""
+		return key.Deployment{}
 	}
 	return m.DeploymentKey
 }
diff --git a/go-runtime/compile/testdata/go/external/go.mod b/go-runtime/compile/testdata/go/external/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/deckarep/golang-set/v2 v2.7.0 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/hashicorp/cronexpr v1.1.2 // indirect
+	github.com/multiformats/go-base36 v0.2.0 // indirect
 	github.com/swaggest/jsonschema-go v0.3.72 // indirect
 	github.com/swaggest/refl v1.3.0 // indirect
 	golang.org/x/exp v0.0.0-20241215155358-4a5509556b9e // indirect

diff --git a/go-runtime/compile/testdata/go/external/go.sum b/go-runtime/compile/testdata/go/external/go.sum
diff --git a/go-runtime/compile/testdata/go/time/go.mod b/go-runtime/compile/testdata/go/time/go.mod
@@ -12,6 +12,7 @@ require (
 	github.com/deckarep/golang-set/v2 v2.7.0 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/hashicorp/cronexpr v1.1.2 // indirect
+	github.com/multiformats/go-base36 v0.2.0 // indirect
 	github.com/swaggest/jsonschema-go v0.3.72 // indirect
 	github.com/swaggest/refl v1.3.0 // indirect
 	golang.org/x/exp v0.0.0-20241215155358-4a5509556b9e // indirect

diff --git a/go-runtime/compile/testdata/go/time/go.sum b/go-runtime/compile/testdata/go/time/go.sum
diff --git a/go-runtime/ftl/ftltest/testdata/go/outer/go.mod b/go-runtime/ftl/ftltest/testdata/go/outer/go.mod
@@ -17,6 +17,7 @@ require (
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/hashicorp/cronexpr v1.1.2 // indirect
 	github.com/hexops/gotextdiff v1.0.3 // indirect
+	github.com/multiformats/go-base36 v0.2.0 // indirect
 	github.com/swaggest/jsonschema-go v0.3.72 // indirect
 	github.com/swaggest/refl v1.3.0 // indirect
 	golang.org/x/exp v0.0.0-20241215155358-4a5509556b9e // indirect

diff --git a/go-runtime/ftl/ftltest/testdata/go/outer/go.sum b/go-runtime/ftl/ftltest/testdata/go/outer/go.sum
diff --git a/go-runtime/ftl/ftltest/testdata/go/time/go.mod b/go-runtime/ftl/ftltest/testdata/go/time/go.mod
@@ -12,6 +12,7 @@ require (
 	github.com/deckarep/golang-set/v2 v2.7.0 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/hashicorp/cronexpr v1.1.2 // indirect
+	github.com/multiformats/go-base36 v0.2.0 // indirect
 	github.com/stretchr/testify v1.9.0 // indirect
 	github.com/swaggest/jsonschema-go v0.3.72 // indirect
 	github.com/swaggest/refl v1.3.0 // indirect

diff --git a/go-runtime/ftl/ftltest/testdata/go/time/go.sum b/go-runtime/ftl/ftltest/testdata/go/time/go.sum
diff --git a/go-runtime/ftl/testdata/go/time/go.mod b/go-runtime/ftl/testdata/go/time/go.mod
@@ -12,6 +12,7 @@ require (
 	github.com/deckarep/golang-set/v2 v2.7.0 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/hashicorp/cronexpr v1.1.2 // indirect
+	github.com/multiformats/go-base36 v0.2.0 // indirect
 	github.com/swaggest/jsonschema-go v0.3.72 // indirect
 	github.com/swaggest/refl v1.3.0 // indirect
 	golang.org/x/exp v0.0.0-20241215155358-4a5509556b9e // indirect