feat(scheduling): replica spread + resource-aware auto placement (#56)

services.spread distributes replicas across every online, schedulable
agent (replica i → agents[i mod n], deterministic). auto placement now
orders by real headroom — memory utilisation, cpu usage, then container
count — instead of count alone.

Lifecycle (start/stop/restart/delete) and GET /runtime are now driven
by container_instances inventory, so remote replicas are visible and
tear down wherever they run — covers spread, pins, and stale pin
leftovers. Spread and explicit pins are mutually exclusive.

Generated with [Devin](https://devin.ai)

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Tomáš DvořákandDevin authored and GitHub committed 2026-10-09 17:03:26 +02:00
1 parent bc93c83cf9
commit dd5ce80ce2
21 files changed
+469 -100

No files matched your search

+4 -2
View File
@@ -425,9 +425,10 @@ func runDeploymentAndSyncWithImage(
// on the local host, so remote placement only works for registry-pulled
// images — git-sourced and rollback deploys fail fast with that message.
nodeID := serviceNodeID(db, service.ID)
if nodeID != "" && (service.GitRepo != "" || imageOverride != "") {
spread := serviceSpread(db, service.ID)
if (nodeID != "" || spread) && (service.GitRepo != "" || imageOverride != "") {
failedAt := time.Now()
msg := "service is pinned to a remote node but remote nodes only run registry-pulled images (git builds and rollbacks stay local — build-on-node is not yet supported)"
msg := "service uses remote node placement but remote nodes only run registry-pulled images (git builds and rollbacks stay local — build-on-node is not yet supported)"
_, _ = db.Exec(
`UPDATE deployments SET status = 'failed', error = $1, completed_at = $2, updated_at = $2 WHERE id = $3`,
msg, failedAt, dbDeployment.ID,
@@ -462,6 +463,7 @@ func runDeploymentAndSyncWithImage(
MaintenanceURL: maintURL,
BasicAuthUsers: basicAuthUsers,
NodeID: nodeID,
Spread: spread,
},
Trigger: deployment.TriggerConfig{
Type: req.Trigger,
+132 -67
View File
@@ -48,6 +48,65 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
if !agent.Status.Valid || agent.Status.String != "online" {
return nil, fmt.Errorf("node %s is %s — heartbeats resume scheduling when it reconnects", agent.Name, agent.Status.String)
}
return r.reconcileOnAgents(ctx, spec, func(int) string { return spec.NodeID })
}
// ReconcileSpread distributes replicas deterministically across online,
// schedulable agents: replica i lands on agents[i % len(agents)], agents
// sorted by id. Replicas wrap when they outnumber nodes. The local host is
// not a spread target — pin "local" or leave unset for local execution.
func (r *agentNodeRunner) ReconcileSpread(ctx context.Context, spec deployment.RuntimeSpec) (*deployment.RuntimeState, error) {
agents, err := r.q.ListSchedulableAgents(ctx)
if err != nil {
return nil, fmt.Errorf("list schedulable nodes: %w", err)
}
if len(agents) == 0 {
return nil, fmt.Errorf("no online schedulable nodes — connect an agent or unset spread")
}
return r.reconcileOnAgents(ctx, spec, func(i int) string {
return agents[i%len(agents)].ID
})
}
// RemoteRuntimeState reads the service's remote replicas from inventory.
// Rows only carry the last dispatched state — the agent doesn't stream
// container events — so "running" means "last create succeeded".
func (r *agentNodeRunner) RemoteRuntimeState(ctx context.Context, serviceID string) (*deployment.RuntimeState, error) {
rows, err := r.q.ListServiceContainers(ctx, serviceID)
if err != nil {
return nil, err
}
state := &deployment.RuntimeState{Status: "stopped"}
running := 0
for _, row := range rows {
var status struct {
State string `json:"state"`
}
unmarshalRaw(row.Status, &status)
if status.State == "" || status.State == "removed" {
continue
}
replica := -1
if idx, err := strconv.Atoi(strings.TrimPrefix(row.Name, fmt.Sprintf("containr-%s-", serviceID))); err == nil {
replica = idx
}
state.Containers = append(state.Containers, deployment.RuntimeContainer{
ID: row.ID, Name: row.Name, State: status.State, Replica: replica,
})
if status.State == "running" {
running++
}
}
if running > 0 {
state.Status = "running"
}
return state, nil
}
// reconcileOnAgents creates the service's replicas on the agents chosen by
// targetFor and retires stale replicas on any agent — including leftovers
// from a previous pin or a different spread assignment.
func (r *agentNodeRunner) reconcileOnAgents(ctx context.Context, spec deployment.RuntimeSpec, targetFor func(replica int) string) (*deployment.RuntimeState, error) {
if len(spec.Domains) > 0 || spec.Domain != "" {
return nil, fmt.Errorf("domains are routed by the local Traefik — remove the service's domains or run it on the local node")
}
@@ -57,81 +116,33 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
replicas = 1
}
// Retire replicas above the desired count using inventory rows.
prefix := fmt.Sprintf("containr-%s-", spec.ServiceID)
existing, _ := r.q.ListContainersForAgent(ctx, spec.NodeID)
existing, _ := r.q.ListServiceContainers(ctx, spec.ServiceID)
for _, row := range existing {
if row.ServiceID != spec.ServiceID || !strings.HasPrefix(row.Name, prefix) {
if !strings.HasPrefix(row.Name, prefix) {
continue
}
replica, err := strconv.Atoi(strings.TrimPrefix(row.Name, prefix))
if err == nil && replica >= replicas {
if cmdErr := r.enqueueAndWait(ctx, spec.NodeID, "", "remove_container",
map[string]interface{}{"container_name": row.Name}); cmdErr != nil {
return nil, fmt.Errorf("remove stale replica %s: %w", row.Name, cmdErr)
}
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: row.ID,
Status: rawJSON(map[string]interface{}{"state": "removed"}),
})
keep := err == nil && replica < replicas && row.NodeAgentID == targetFor(replica)
if keep {
continue
}
if cmdErr := r.enqueueAndWait(ctx, row.NodeAgentID, "", "remove_container",
map[string]interface{}{"container_name": row.Name}); cmdErr != nil {
return nil, fmt.Errorf("remove stale replica %s on node %s: %w", row.Name, row.NodeAgentID, cmdErr)
}
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: row.ID,
Status: rawJSON(map[string]interface{}{"state": "removed"}),
})
}
state := &deployment.RuntimeState{Desired: replicas, Status: "running"}
for i := 0; i < replicas; i++ {
agentID := targetFor(i)
name := remoteContainerName(spec.ServiceID, i)
containerID := remoteContainerID(spec.NodeID, spec.ServiceID, i)
ports := []interface{}{}
if spec.Port > 0 {
hostPort := spec.PublishedPort // 0 → ephemeral on the node
if i > 0 {
hostPort = 0 // only replica 0 claims the preferred host port
}
ports = append(ports, map[string]interface{}{
"published": true,
"host_port": hostPort,
"container_port": spec.Port,
"protocol": "tcp",
})
}
volumes := make([]interface{}, 0, len(spec.Volumes))
for _, v := range spec.Volumes {
volumes = append(volumes, map[string]interface{}{
"source": v.Source, "target": v.Destination, "read_only": v.ReadOnly,
})
}
env := map[string]interface{}{}
for k, v := range spec.Env {
env[k] = v
}
restart := spec.RestartPolicy
if restart == "" {
restart = "unless-stopped"
}
cmd := make([]interface{}, 0, len(spec.Command))
for _, arg := range spec.Command {
cmd = append(cmd, arg)
}
container := map[string]interface{}{
"name": name,
"image": spec.Image,
"command": cmd,
"environment": env,
"registry": r.pullCredentials(spec.ProjectID, spec.Image),
"ports": ports,
"volumes": volumes,
"restart_policy": restart,
"memory": spec.MemoryBytes,
"cpus": float64(spec.NanoCPUs) / 1e9,
"labels": map[string]interface{}{
"containr.managed": "true",
"containr.project": spec.ProjectID,
"containr.service": spec.ServiceID,
"containr.replica": fmt.Sprint(i),
"containr.node": spec.NodeID,
},
}
containerID := remoteContainerID(agentID, spec.ServiceID, i)
container, env, ports, volumes, restart := r.replicaPayload(spec, agentID, i, name)
_ = r.q.UpsertServiceContainer(ctx, sqlcdb.UpsertServiceContainerParams{
ID: containerID,
@@ -139,7 +150,7 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
Image: spec.Image,
ProjectID: spec.ProjectID,
ServiceID: spec.ServiceID,
NodeAgentID: spec.NodeID,
NodeAgentID: agentID,
Status: rawJSON(map[string]interface{}{"state": "created", "health": "none"}),
Resources: rawJSON(map[string]interface{}{"memory": spec.MemoryBytes, "cpus": float64(spec.NanoCPUs) / 1e9}),
Ports: rawJSON(ports),
@@ -150,9 +161,9 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
HealthCheck: rawJSON(map[string]interface{}{"path": spec.HealthPath}),
})
if err := r.enqueueAndWait(ctx, spec.NodeID, containerID, "create_container",
if err := r.enqueueAndWait(ctx, agentID, containerID, "create_container",
map[string]interface{}{"container": container}); err != nil {
return nil, fmt.Errorf("replica %d on %s: %w", i, agent.Name, err)
return nil, fmt.Errorf("replica %d on node %s: %w", i, agentID, err)
}
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: containerID,
@@ -165,6 +176,60 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
return state, nil
}
// replicaPayload builds the agent create_container payload for one replica.
func (r *agentNodeRunner) replicaPayload(spec deployment.RuntimeSpec, agentID string, i int, name string) (map[string]interface{}, map[string]interface{}, []interface{}, []interface{}, string) {
ports := []interface{}{}
if spec.Port > 0 {
hostPort := spec.PublishedPort // 0 → ephemeral on the node
if i > 0 {
hostPort = 0 // only replica 0 claims the preferred host port
}
ports = append(ports, map[string]interface{}{
"published": true,
"host_port": hostPort,
"container_port": spec.Port,
"protocol": "tcp",
})
}
volumes := make([]interface{}, 0, len(spec.Volumes))
for _, v := range spec.Volumes {
volumes = append(volumes, map[string]interface{}{
"source": v.Source, "target": v.Destination, "read_only": v.ReadOnly,
})
}
env := map[string]interface{}{}
for k, v := range spec.Env {
env[k] = v
}
restart := spec.RestartPolicy
if restart == "" {
restart = "unless-stopped"
}
cmd := make([]interface{}, 0, len(spec.Command))
for _, arg := range spec.Command {
cmd = append(cmd, arg)
}
return map[string]interface{}{
"name": name,
"image": spec.Image,
"command": cmd,
"environment": env,
"registry": r.pullCredentials(spec.ProjectID, spec.Image),
"ports": ports,
"volumes": volumes,
"restart_policy": restart,
"memory": spec.MemoryBytes,
"cpus": float64(spec.NanoCPUs) / 1e9,
"labels": map[string]interface{}{
"containr.managed": "true",
"containr.project": spec.ProjectID,
"containr.service": spec.ServiceID,
"containr.replica": fmt.Sprint(i),
"containr.node": agentID,
},
}, env, ports, volumes, restart
}
// pullCredentials resolves the project owner's registry auth for the image
// host so private images pull on remote nodes. Empty map → anonymous pull.
func (r *agentNodeRunner) pullCredentials(projectID, imageRef string) map[string]interface{} {
+72 -12
View File
@@ -18,13 +18,37 @@ import (
"github.com/google/uuid"
)
// serviceLifecycle routes a start/stop/restart action to the node agent when
// the service is pinned, or to the local Docker engine otherwise. Remote
// actions block on the agent's poll loop; that matches the synchronous
// behavior the local path already has.
// remoteAgentsForService lists every agent holding inventory rows for the
// service — covers pinned placements, spread replicas, and leftovers from a
// since-cleared pin.
func remoteAgentsForService(ctx context.Context, db *database.DB, serviceID uuid.UUID) []string {
rows, err := db.Query(
`SELECT DISTINCT node_agent_id FROM container_instances WHERE service_id = $1`,
serviceID,
)
if err != nil {
return nil
}
defer rows.Close()
var ids []string
for rows.Next() {
var id string
if rows.Scan(&id) == nil && id != "" {
ids = append(ids, id)
}
}
return ids
}
// serviceLifecycle routes a start/stop/restart action to every agent holding
// the service's replicas, then to the local Docker engine. Both run — a
// service can't be pinned and local at once, but stale containers from a
// cleared pin shouldn't survive a stop.
func serviceLifecycle(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, serviceID uuid.UUID, action string) error {
if nodeID := serviceNodeID(db, serviceID); nodeID != "" {
return engine.ControlServiceOnNode(ctx, serviceID.String(), nodeID, action)
for _, agentID := range remoteAgentsForService(ctx, db, serviceID) {
if err := engine.ControlServiceOnNode(ctx, serviceID.String(), agentID, action); err != nil {
return err
}
}
switch action {
case "start":
@@ -40,12 +64,11 @@ func serviceLifecycle(ctx context.Context, db *database.DB, engine *deployment.D
return fmt.Errorf("unsupported lifecycle action %q", action)
}
// removeServiceRuntime tears down replicas wherever they run — remote node
// when pinned, local Docker otherwise.
// removeServiceRuntime tears down replicas wherever they run — remote nodes
// (pinned or spread) and the local Docker host.
func removeServiceRuntime(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, serviceID uuid.UUID) {
if nodeID := serviceNodeID(db, serviceID); nodeID != "" {
_ = engine.RemoveServiceContainersOnNode(ctx, serviceID.String(), nodeID)
return
for _, agentID := range remoteAgentsForService(ctx, db, serviceID) {
_ = engine.RemoveServiceContainersOnNode(ctx, serviceID.String(), agentID)
}
_ = engine.RemoveServiceContainers(ctx, serviceID.String())
}
@@ -70,6 +93,7 @@ func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpe
HealthPath: service.HealthCheckPath,
RestartPolicy: service.RestartPolicy,
NodeID: serviceNodeID(db, service.ID),
Spread: serviceSpread(db, service.ID),
}
// Best effort: reuse the host port from the last live deployment so the
// public URL survives restarts. Column exists post-migration; older DBs
@@ -374,6 +398,7 @@ func handleGetServiceRuntime(c *gin.Context) {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
mergeRemoteRuntime(c.Request.Context(), engine, service.ID.String(), state)
engine.ProbeServiceHealth(c.Request.Context(), state, service.ProjectID.String(), service.Name, service.Port, service.HealthCheckPath)
persistPublishedPort(db, service.ID, state.Ports)
if host := requestHostname(c); host != "" {
@@ -564,6 +589,37 @@ func enqueueServiceRedeploy(c *gin.Context, db *database.DB, service Service) bo
return true
}
// mergeRemoteRuntime folds inventory-backed remote replica state into a
// local runtime view so pinned and spread services report their containers.
// Remote rows carry deterministic UUIDs, local ones Docker IDs — no overlap.
func mergeRemoteRuntime(ctx context.Context, engine *deployment.DeploymentEngine, serviceID string, state *deployment.RuntimeState) {
remote, err := engine.RemoteRuntimeState(ctx, serviceID)
if err != nil || remote == nil {
return
}
running := 0
for _, c := range state.Containers {
if c.State == "running" {
running++
}
}
for _, rc := range remote.Containers {
state.Containers = append(state.Containers, rc)
if rc.State == "running" {
running++
}
}
switch {
case len(state.Containers) == 0:
case running == 0:
state.Status = "stopped"
case running < state.Desired:
state.Status = "degraded"
default:
state.Status = "running"
}
}
// persistPublishedPort records the host port Docker actually bound so later
// redeploys can reuse it. Idempotent — only writes when the value changed.
func persistPublishedPort(db *database.DB, serviceID uuid.UUID, ports []uint16) {
@@ -587,7 +643,11 @@ func liveServiceStatus(c *gin.Context, db *database.DB, service *Service) {
ctx, cancel := context.WithTimeout(c.Request.Context(), 3*time.Second)
defer cancel()
state, err := engine.RuntimeState(ctx, service.ID.String(), service.Replicas)
if err != nil || state == nil || len(state.Containers) == 0 {
if err != nil || state == nil {
return
}
mergeRemoteRuntime(ctx, engine, service.ID.String(), state)
if len(state.Containers) == 0 {
return
}
if state.Status != service.Status {
+53 -8
View File
@@ -51,8 +51,11 @@ type Service struct {
SleepIdleMinutes int `json:"sleep_idle_minutes" db:"-"`
// NodeID pins the service to a node agent ("" = local Docker host,
// resolved "auto" picks the least-loaded online agent at write time).
NodeID string `json:"node_id,omitempty" db:"-"`
NodeName string `json:"node_name,omitempty" db:"-"`
NodeID string `json:"node_id,omitempty" db:"-"`
NodeName string `json:"node_name,omitempty" db:"-"`
// Spread distributes replicas across every online, schedulable agent.
// Mutually exclusive with an explicit NodeID pin.
Spread bool `json:"spread" db:"-"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
}
@@ -79,10 +82,21 @@ func serviceNodeID(db *database.DB, serviceID uuid.UUID) string {
return ""
}
// loadServiceNode fills the pin + the agent's display name for responses.
// serviceSpread reads the spread column with the same lazy pattern.
func serviceSpread(db *database.DB, serviceID uuid.UUID) bool {
var spread bool
if err := db.QueryRow(`SELECT COALESCE(spread, false) FROM services WHERE id = $1`, serviceID).Scan(&spread); err == nil {
return spread
}
return false
}
// loadServiceNode fills the pin + spread + the agent's display name for
// responses.
func loadServiceNode(db *database.DB, s *Service) {
var name sql.NullString
s.NodeID = serviceNodeID(db, s.ID)
s.Spread = serviceSpread(db, s.ID)
if s.NodeID == "" {
return
}
@@ -171,6 +185,9 @@ type CreateServiceRequest struct {
// NodeID pins placement to a node agent; "auto" picks the least-loaded
// online agent, "local"/"" runs on the Containr host.
NodeID string `json:"node_id"`
// Spread distributes replicas across every online, schedulable agent.
// Mutually exclusive with an explicit node pin.
Spread *bool `json:"spread"`
}
// UpdateServiceRequest represents a request to update a service
@@ -202,6 +219,9 @@ type UpdateServiceRequest struct {
SleepIdleMinutes *int `json:"sleep_idle_minutes"`
// NodeID pins placement; pointer so "local"/"" explicitly clears the pin.
NodeID *string `json:"node_id"`
// Spread toggles multi-node replica distribution; pointer so false is
// a real update, not a missing field.
Spread *bool `json:"spread"`
}
// handleGetServices retrieves all services for a project
@@ -442,13 +462,22 @@ func handleCreateService(c *gin.Context) {
sourceType := inferServiceSourceType(service)
if req.Spread != nil {
service.Spread = *req.Spread
}
if req.NodeID != "" {
if service.Spread && req.NodeID != "auto" {
c.JSON(http.StatusBadRequest, gin.H{"error": "spread and an explicit node pin are mutually exclusive", "code": "VALIDATION"})
return
}
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), req.NodeID)
if nodeErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
return
}
service.NodeID = nodeID
if !service.Spread {
service.NodeID = nodeID
}
}
// Insert service into database
@@ -459,9 +488,9 @@ func handleCreateService(c *gin.Context) {
git_repo, git_branch, build_path, cpu, memory, replicas, port, domain,
healthcheck_path, restart_policy, volumes,
builder, cpu_reserve, memory_reserve, static_build_cmd, static_dir,
node_id, created_at, updated_at)
node_id, spread, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26,
$27, $28, $29, $30, $31, $32, $33, $34)`,
$27, $28, $29, $30, $31, $32, $33, $34, $35)`,
service.ID, service.ProjectID, service.Name, environmentID, service.Type,
sourceType, firstNonEmpty(service.GitRepo, service.Image), service.Image,
"", service.Command, service.Type, service.Status, service.Image, service.Command,
@@ -470,6 +499,7 @@ func handleCreateService(c *gin.Context) {
volumesJSON, service.Builder, service.CPUReserve, service.MemoryReserve,
service.StaticBuildCmd, service.StaticDir,
sql.NullString{String: service.NodeID, Valid: service.NodeID != ""},
service.Spread,
service.CreatedAt, service.UpdatedAt,
)
@@ -772,8 +802,18 @@ func handleUpdateService(c *gin.Context) {
volumesArg = volumesJSON
}
if req.Spread != nil {
existingService.Spread = *req.Spread
} else {
existingService.Spread = serviceSpread(db.(*database.DB), serviceID)
}
if req.NodeID != nil {
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), *req.NodeID)
nodeValue := strings.TrimSpace(*req.NodeID)
if existingService.Spread && nodeValue != "" && nodeValue != "auto" && nodeValue != "local" {
c.JSON(http.StatusBadRequest, gin.H{"error": "spread and an explicit node pin are mutually exclusive — set spread=false first", "code": "VALIDATION"})
return
}
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), nodeValue)
if nodeErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
return
@@ -782,6 +822,10 @@ func handleUpdateService(c *gin.Context) {
} else {
existingService.NodeID = serviceNodeID(db.(*database.DB), serviceID)
}
if existingService.Spread && existingService.NodeID != "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "service already has an explicit node pin — clear it (node_id=\"local\") before enabling spread", "code": "VALIDATION"})
return
}
existingService.UpdatedAt = time.Now()
@@ -794,7 +838,7 @@ func handleUpdateService(c *gin.Context) {
restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16,
builder = $19, cpu_reserve = $20, memory_reserve = $21,
static_build_cmd = $22, static_dir = $23,
sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26
sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26, spread = $27
WHERE id = $18`,
existingService.Name, existingService.Type, existingService.Image, existingService.Command,
existingService.Environment, existingService.GitRepo, existingService.GitBranch,
@@ -806,6 +850,7 @@ func handleUpdateService(c *gin.Context) {
existingService.StaticBuildCmd, existingService.StaticDir,
existingService.SleepEnabled, existingService.SleepIdleMinutes,
sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""},
existingService.Spread,
)
if err != nil {