feat: keys include host-port-suffix

backend/controller/controller.go

@@ -22,7 +22,6 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/jellydator/ttlcache/v3"
 	"github.com/jpillora/backoff"
-	"github.com/oklog/ulid/v2"
 	"golang.org/x/exp/maps"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/structpb"
@@ -55,7 +54,7 @@ type Config struct {
 	ConsoleURL                   *url.URL            `help:"The public URL of the console (for CORS)." env:"FTL_CONTROLLER_CONSOLE_URL"`
 	AllowOrigins                 []*url.URL          `help:"Allow CORS requests to ingress endpoints from these origins." env:"FTL_CONTROLLER_ALLOW_ORIGIN"`
 	ContentTime                  time.Time           `help:"Time to use for console resource timestamps." default:"${timestamp=1970-01-01T00:00:00Z}"`
-	Key                          model.ControllerKey `help:"Controller key (auto)." placeholder:"C<ULID>" default:"C00000000000000000000000000"`
+	Key                          model.ControllerKey `help:"Controller key (auto)."`
 	DSN                          string              `help:"DAL DSN." default:"postgres://localhost:54320/ftl?sslmode=disable&user=postgres&password=secret" env:"FTL_CONTROLLER_DSN"`
 	RunnerTimeout                time.Duration       `help:"Runner heartbeat timeout." default:"10s"`
 	DeploymentReservationTimeout time.Duration       `help:"Deployment reservation timeout." default:"120s"`
@@ -152,9 +151,10 @@ type Service struct {
 }
 
 func New(ctx context.Context, db *dal.DAL, config Config, runnerScaling scaling.RunnerScaling) (*Service, error) {
+	var zero model.ControllerKey
 	key := config.Key
-	if config.Key.ULID() == (ulid.ULID{}) {
-		key = model.NewControllerKey()
+	if config.Key == zero {
+		key = model.NewControllerKey(config.Bind.Hostname(), config.Bind.Port())
 	}
 	config.SetDefaults()
 	svc := &Service{

backend/controller/dal/dal.go

@@ -91,8 +91,9 @@ func runnerFromDB(row sql.GetRunnerRow) Runner {
 	if err := json.Unmarshal(row.Labels, &attrs); err != nil {
 		return Runner{}
 	}
+
 	return Runner{
-		Key:        model.RunnerKey(row.RunnerKey),
+		Key:        row.RunnerKey,
 		Endpoint:   row.Endpoint,
 		State:      RunnerState(row.State),
 		Deployment: deployment,
@@ -308,8 +309,9 @@ func (d *DAL) GetStatus(
 		if err := json.Unmarshal(in.Labels, &attrs); err != nil {
 			return Runner{}, fmt.Errorf("invalid attributes JSON for runner %s: %w", in.RunnerKey, err)
 		}
+
 		return Runner{
-			Key:        model.RunnerKey(in.RunnerKey),
+			Key:        in.RunnerKey,
 			Endpoint:   in.Endpoint,
 			State:      RunnerState(in.State),
 			Deployment: deployment,
@@ -335,7 +337,7 @@ func (d *DAL) GetStatus(
 		Routes: slices.Map(routes, func(row sql.GetRoutingTableRow) Route {
 			return Route{
 				Module:     row.ModuleName.MustGet(),
-				Runner:     model.RunnerKey(row.RunnerKey),
+				Runner:     row.RunnerKey,
 				Deployment: row.DeploymentName,
 				Endpoint:   row.Endpoint,
 			}
@@ -354,8 +356,9 @@ func (d *DAL) GetRunnersForDeployment(ctx context.Context, deployment model.Depl
 		if err := json.Unmarshal(row.Labels, &attrs); err != nil {
 			return nil, fmt.Errorf("invalid attributes JSON for runner %d: %w", row.ID, err)
 		}
+
 		runners = append(runners, Runner{
-			Key:        model.RunnerKey(row.Key),
+			Key:        row.Key,
 			Endpoint:   row.Endpoint,
 			State:      RunnerState(row.State),
 			Deployment: optional.Some(deployment),
@@ -502,7 +505,7 @@ func (d *DAL) UpsertRunner(ctx context.Context, runner Runner) error {
 		return fmt.Errorf("%s: %w", "failed to JSON encode runner labels", err)
 	}
 	deploymentID, err := d.db.UpsertRunner(ctx, sql.UpsertRunnerParams{
-		Key:            sql.Key(runner.Key),
+		Key:            runner.Key,
 		Endpoint:       runner.Endpoint,
 		State:          sql.RunnerState(runner.State),
 		DeploymentName: pgDeploymentName,
@@ -511,9 +514,6 @@ func (d *DAL) UpsertRunner(ctx context.Context, runner Runner) error {
 	if err != nil {
 		return translatePGError(err)
 	}
-	if err != nil {
-		return translatePGError(err)
-	}
 	if runner.Deployment.Ok() && !deploymentID.Ok() {
 		return fmt.Errorf("deployment %s not found", runner.Deployment)
 	}
@@ -534,7 +534,7 @@ func (d *DAL) KillStaleControllers(ctx context.Context, age time.Duration) (int6
 
 // DeregisterRunner deregisters the given runner.
 func (d *DAL) DeregisterRunner(ctx context.Context, key model.RunnerKey) error {
-	count, err := d.db.DeregisterRunner(ctx, sql.Key(key))
+	count, err := d.db.DeregisterRunner(ctx, key)
 	if err != nil {
 		return translatePGError(err)
 	}
@@ -574,11 +574,12 @@ func (d *DAL) ReserveRunnerForDeployment(ctx context.Context, deployment model.D
 		cancel()
 		return nil, fmt.Errorf("failed to JSON decode labels for runner %d: %w", runner.ID, err)
 	}
+
 	return &postgresClaim{
 		cancel: cancel,
 		tx:     tx,
 		runner: Runner{
-			Key:        model.RunnerKey(runner.Key),
+			Key:        runner.Key,
 			Endpoint:   runner.Endpoint,
 			State:      RunnerState(runner.State),
 			Deployment: optional.Some(deployment),
@@ -628,7 +629,7 @@ func (d *DAL) SetDeploymentReplicas(ctx context.Context, key model.DeploymentNam
 	}
 
 	err = tx.InsertDeploymentUpdatedEvent(ctx, sql.InsertDeploymentUpdatedEventParams{
-		DeploymentName:  key.String(),
+		DeploymentName:  string(key),
 		MinReplicas:     int32(minReplicas),
 		PrevMinReplicas: deployment.MinReplicas,
 	})
@@ -766,8 +767,9 @@ func (d *DAL) GetProcessList(ctx context.Context) ([]Process, error) {
 			if err := json.Unmarshal(row.RunnerLabels, &labels); err != nil {
 				return Process{}, fmt.Errorf("invalid labels JSON for runner %s: %w", row.RunnerKey, err)
 			}
+
 			runner = optional.Some(ProcessRunner{
-				Key:      model.RunnerKey(row.RunnerKey.MustGet()),
+				Key:      row.RunnerKey.MustGet(),
 				Endpoint: endpoint,
 				Labels:   labels,
 			})
@@ -812,8 +814,9 @@ func (d *DAL) GetIdleRunners(ctx context.Context, limit int, labels model.Labels
 		if err != nil {
 			return Runner{}, fmt.Errorf("%s: %w", "could not unmarshal labels", err)
 		}
+
 		return Runner{
-			Key:      model.RunnerKey(row.Key),
+			Key:      row.Key,
 			Endpoint: row.Endpoint,
 			State:    RunnerState(row.State),
 			Labels:   labels,
@@ -840,23 +843,23 @@ func (d *DAL) GetRoutingTable(ctx context.Context, modules []string) (map[string
 		out[moduleName] = append(out[moduleName], Route{
 			Module:     moduleName,
 			Deployment: route.DeploymentName,
-			Runner:     model.RunnerKey(route.RunnerKey),
+			Runner:     route.RunnerKey,
 			Endpoint:   route.Endpoint,
 		})
 	}
 	return out, nil
 }
 
 func (d *DAL) GetRunnerState(ctx context.Context, runnerKey model.RunnerKey) (RunnerState, error) {
-	state, err := d.db.GetRunnerState(ctx, sql.Key(runnerKey))
+	state, err := d.db.GetRunnerState(ctx, runnerKey)
 	if err != nil {
 		return "", translatePGError(err)
 	}
 	return RunnerState(state), nil
 }
 
 func (d *DAL) GetRunner(ctx context.Context, runnerKey model.RunnerKey) (Runner, error) {
-	row, err := d.db.GetRunner(ctx, sql.Key(runnerKey))
+	row, err := d.db.GetRunner(ctx, runnerKey)
 	if err != nil {
 		return Runner{}, translatePGError(err)
 	}
@@ -927,7 +930,7 @@ func (d *DAL) GetIngressRoutes(ctx context.Context, method string) ([]IngressRou
 	}
 	return slices.Map(routes, func(row sql.GetIngressRoutesRow) IngressRoute {
 		return IngressRoute{
-			Runner:     model.RunnerKey(row.RunnerKey),
+			Runner:     row.RunnerKey,
 			Deployment: row.DeploymentName,
 			Endpoint:   row.Endpoint,
 			Path:       row.Path,

backend/controller/dal/dal_test.go

@@ -86,7 +86,7 @@ func TestDAL(t *testing.T) {
 		assert.Equal(t, []sha256.SHA256{misshingSHA}, missing)
 	})
 
-	runnerID := model.NewRunnerKey()
+	runnerID := model.NewRunnerKey("localhost", "8080")
 	labels := map[string]any{"languages": []any{"go"}}
 
 	t.Run("RegisterRunner", func(t *testing.T) {
@@ -101,7 +101,7 @@ func TestDAL(t *testing.T) {
 
 	t.Run("RegisterRunnerFailsOnDuplicate", func(t *testing.T) {
 		err = dal.UpsertRunner(ctx, Runner{
-			Key:      model.NewRunnerKey(),
+			Key:      model.NewRunnerKey("localhost", "8080"),
 			Labels:   labels,
 			Endpoint: "http://localhost:8080",
 			State:    RunnerStateIdle,
@@ -333,7 +333,7 @@ func TestDAL(t *testing.T) {
 	})
 
 	t.Run("DeregisterRunnerFailsOnMissing", func(t *testing.T) {
-		err = dal.DeregisterRunner(ctx, model.NewRunnerKey())
+		err = dal.DeregisterRunner(ctx, model.NewRunnerKey("localhost", "8080"))
 		assert.IsError(t, err, ErrNotFound)
 	})
 }

backend/controller/scaling/localscaling/local_scaling.go

@@ -27,6 +27,8 @@ type LocalScaling struct {
 
 	portAllocator       *bind.BindAllocator
 	controllerAddresses []*url.URL
+
+	prevRunnerSuffix int
 }
 
 func NewLocalScaling(portAllocator *bind.BindAllocator, controllerAddresses []*url.URL) (*LocalScaling, error) {
@@ -40,6 +42,7 @@ func NewLocalScaling(portAllocator *bind.BindAllocator, controllerAddresses []*u
 		runners:             map[model.RunnerKey]context.CancelFunc{},
 		portAllocator:       portAllocator,
 		controllerAddresses: controllerAddresses,
+		prevRunnerSuffix:    -1,
 	}, nil
 }
 
@@ -72,25 +75,28 @@ func (l *LocalScaling) SetReplicas(ctx context.Context, replicas int, idleRunner
 
 	logger.Debugf("Adding %d replicas", replicasToAdd)
 	for i := 0; i < replicasToAdd; i++ {
-		i := i
-
 		controllerEndpoint := l.controllerAddresses[len(l.runners)%len(l.controllerAddresses)]
+
+		bind := l.portAllocator.Next()
+		keySuffix := l.prevRunnerSuffix + 1
+		l.prevRunnerSuffix = keySuffix
+
 		config := runner.Config{
-			Bind:               l.portAllocator.Next(),
+			Bind:               bind,
 			ControllerEndpoint: controllerEndpoint,
 			TemplateDir:        templateDir(ctx),
-			Key:                model.NewRunnerKey(),
+			Key:                model.NewLocalRunnerKey(keySuffix),
 		}
 
-		name := fmt.Sprintf("runner%d", i)
+		simpleName := fmt.Sprintf("runner%d", keySuffix)
 		if err := kong.ApplyDefaults(&config, kong.Vars{
-			"deploymentdir": filepath.Join(l.cacheDir, "ftl-runner", name, "deployments"),
+			"deploymentdir": filepath.Join(l.cacheDir, "ftl-runner", simpleName, "deployments"),
 			"language":      "go,kotlin",
 		}); err != nil {
 			return err
 		}
 
-		runnerCtx := log.ContextWithLogger(ctx, logger.Scope(name))
+		runnerCtx := log.ContextWithLogger(ctx, logger.Scope(simpleName))
 
 		runnerCtx, cancel := context.WithCancel(runnerCtx)
 		l.runners[config.Key] = cancel

backend/controller/scheduledtask/scheduledtask_test.go

@@ -31,10 +31,10 @@ func TestCron(t *testing.T) {
 	}
 
 	controllers := []*controller{
-		{controller: dal.Controller{Key: model.NewControllerKey()}},
-		{controller: dal.Controller{Key: model.NewControllerKey()}},
-		{controller: dal.Controller{Key: model.NewControllerKey()}},
-		{controller: dal.Controller{Key: model.NewControllerKey()}},
+		{controller: dal.Controller{Key: model.NewControllerKey("localhost", "8080")}},
+		{controller: dal.Controller{Key: model.NewControllerKey("localhost", "8081")}},
+		{controller: dal.Controller{Key: model.NewControllerKey("localhost", "8082")}},
+		{controller: dal.Controller{Key: model.NewControllerKey("localhost", "8083")}},
 	}
 
 	clock := clock.NewMock()

backend/controller/sql/queries.sql

@@ -122,7 +122,7 @@ WITH matches AS (
     UPDATE runners
         SET state = 'dead',
             deployment_id = NULL
-        WHERE key = $1
+        WHERE key = sqlc.arg('key')::runner_key
         RETURNING 1)
 SELECT COUNT(*)
 FROM matches;
@@ -222,7 +222,7 @@ RETURNING runners.*;
 -- name: GetRunnerState :one
 SELECT state
 FROM runners
-WHERE key = $1;
+WHERE key = sqlc.arg('key')::runner_key;
 
 -- name: GetRunner :one
 SELECT DISTINCT ON (r.key) r.key                                   AS runner_key,
@@ -236,7 +236,7 @@ SELECT DISTINCT ON (r.key) r.key                                   AS runner_key
                                             THEN d.name END, NULL) AS deployment_name
 FROM runners r
          LEFT JOIN deployments d on d.id = r.deployment_id OR r.deployment_id IS NULL
-WHERE r.key = $1;
+WHERE r.key = sqlc.arg('key')::runner_key;
 
 -- name: GetRoutingTable :many
 SELECT endpoint, r.key AS runner_key, r.module_name, d.name deployment_name
@@ -251,7 +251,7 @@ WHERE state = 'assigned'
 SELECT endpoint, r.key AS runner_key, r.module_name, d.name deployment_name, r.state
 FROM runners r
          LEFT JOIN deployments d on r.deployment_id = d.id
-WHERE r.key = $1;
+WHERE r.key = sqlc.arg('key')::runner_key;
 
 -- name: GetRunnersForDeployment :many
 SELECT *