Compare commits

..
2 Commits
Author SHA1 Message Date
Tomáš DvořákandDevin 9de586f177 feat(nodes): agent self-upgrade (#57)
POST /agents/:id/upgrade enqueues a self_upgrade command carrying the
target platform, the sha256 the server computed over its bundled binary,
and the server version. The agent downloads the replacement from the
existing /api/agents/download/:platform endpoint, verifies the checksum,
sanity-checks the new binary (--version), atomically swaps its own
executable, reports the command result, then re-execs via syscall.Exec —
no supervisor or restart plumbing required.

Agents stamp main.version via ldflags at image build, expose --version,
and now send version at register so node_agents.version refreshes on
re-register (older agents without the field keep their stored value).
The server rejects upgrades for offline nodes and for platforms with no
bundled binary (linux-amd64/linux-arm64 ship in the image).

Surfaces: nodes upgrade CLI, containr_nodes_upgrade MCP (confirm-gated),
node-detail page button, OpenAPI. Verified live: v0.0.3 agent upgraded
to v0.0.4 in place, re-registered with the new version; offline node
returns 409 NODE_OFFLINE.

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

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-09 17:15:58 +02:00
Tomáš DvořákandDevin dd5ce80ce2 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>
2026-10-09 17:03:26 +02:00
26 changed files with 771 additions and 79 deletions

No files matched your search

+22 -3
View File
@@ -289,11 +289,30 @@ and wired refs, one click or one CLI call.
(drain evicts remote containers via the agent queue + unpins services (drain evicts remote containers via the agent queue + unpins services
so they redeploy locally), `nodes cordon|uncordon|drain` CLI + so they redeploy locally), `nodes cordon|uncordon|drain` CLI +
MCP (drain confirm-gated) + node-detail page controls; `auto` and MCP (drain confirm-gated) + node-detail page controls; `auto` and
explicit pins skip cordoned nodes. Remote nodes run registry-pulled explicit pins skip cordoned nodes. Spread placement shipped:
`services.spread` distributes replicas deterministically across every
online schedulable agent (replica i → agents[i mod n]), mutually
exclusive with a pin; `auto` is now resource-aware — least memory
utilisation, then cpu, then container count, with uninstrumented
agents sorting last. Lifecycle and `GET /runtime` are
inventory-driven, so remote replicas show up and tear down wherever
they run. Remote nodes run registry-pulled
images only — git builds, Traefik domains, and remote sleep are images only — git builds, Traefik domains, and remote sleep are
documented ceilings. Remaining: `scheduling_rules` spread/affinity documented ceilings. Remaining: `scheduling_rules` affinity
policies, deployment status reported back from agent policies, deployment status reported back from agent
results, per-node resource-aware `auto` (currently count-based). results, spread weighting by real capacity.
- [x] **Agent self-upgrade** — `POST /agents/:id/upgrade` enqueues a
`self_upgrade` command carrying platform + sha256 + server version;
the agent downloads its replacement from `GET /agents/download/
:platform`, verifies the checksum, sanity-checks `--version`,
atomically swaps its own executable, reports the result, then
re-execs (`syscall.Exec`) — no supervisor needed. Agents now stamp
`main.version` via ldflags at image build and report it at
register/heartbeat (`node_agents.version` refreshes on re-register;
older agents keep their stored value). Surfaces: `nodes upgrade`
CLI, `containr_nodes_upgrade` MCP (confirm-gated), node-detail page
button (online-gated). Ceilings: linux amd64/arm64 only; agents on
other platforms get a `NO_BINARY` 409.
- [🟡] **Build-on-node** — private registry pulls shipped: the dispatch - [🟡] **Build-on-node** — private registry pulls shipped: the dispatch
payload carries the project owner's `registries` credentials for the payload carries the project owner's `registries` credentials for the
image host; the agent `docker login`s for the pull and logs out after, image host; the agent `docker login`s for the pull and logs out after,
+2 -2
View File
@@ -14,8 +14,8 @@ RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo \
-ldflags "-X containr/internal/api.Version=${VERSION}" -o main cmd/server/main.go -ldflags "-X containr/internal/api.Version=${VERSION}" -o main cmd/server/main.go
# Node-agent binaries served by the API for SSH/bootstrap installs. # Node-agent binaries served by the API for SSH/bootstrap installs.
RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /agents/containr-agent-linux-amd64 ./cmd/agent && \ RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags "-X main.version=${VERSION}" -o /agents/containr-agent-linux-amd64 ./cmd/agent && \
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o /agents/containr-agent-linux-arm64 ./cmd/agent CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -ldflags "-X main.version=${VERSION}" -o /agents/containr-agent-linux-arm64 ./cmd/agent
# Build embedded Better Auth runtime dependencies # Build embedded Better Auth runtime dependencies
FROM node:22-alpine AS auth-builder FROM node:22-alpine AS auth-builder
+88 -2
View File
@@ -3,14 +3,18 @@ package main
import ( import (
"bytes" "bytes"
"context" "context"
"crypto/sha256"
"encoding/hex"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"io"
"log" "log"
"net" "net"
"net/http" "net/http"
"os" "os"
"os/exec" "os/exec"
"path/filepath"
"runtime" "runtime"
"strconv" "strconv"
"strings" "strings"
@@ -104,7 +108,15 @@ type registerResponse struct {
Status string `json:"status"` Status string `json:"status"`
} }
// version is stamped at build time via -ldflags "-X main.version=…".
// The server reports it on upgrades so agents can show what they run.
var version = "dev"
func main() { func main() {
if len(os.Args) > 1 && (os.Args[1] == "--version" || os.Args[1] == "version") {
fmt.Printf("containr-agent %s\n", version)
return
}
cfg, err := loadConfig() cfg, err := loadConfig()
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
@@ -167,6 +179,7 @@ func register(ctx context.Context, client *http.Client, cfg agentConfig) (string
"ip_address": cfg.IP, "ip_address": cfg.IP,
"port": cfg.Port, "port": cfg.Port,
"capabilities": detectCapabilities(), "capabilities": detectCapabilities(),
"version": firstEnv("CONTAINR_AGENT_VERSION", version),
// Always send mesh so a removed interface clears stored state. // Always send mesh so a removed interface clears stored state.
"mesh": detectMeshInterfaces(), "mesh": detectMeshInterfaces(),
} }
@@ -190,7 +203,7 @@ func sendHeartbeat(ctx context.Context, client *http.Client, cfg agentConfig, ag
"container_count": dockerContainerCount(), "container_count": dockerContainerCount(),
"system_load": readSystemLoad(), "system_load": readSystemLoad(),
"uptime": readSystemUptime(), "uptime": readSystemUptime(),
"version": firstEnv("CONTAINR_AGENT_VERSION", "dev"), "version": firstEnv("CONTAINR_AGENT_VERSION", version),
} }
return postJSON(ctx, client, cfg, "/api/agents/heartbeat", payload, nil) return postJSON(ctx, client, cfg, "/api/agents/heartbeat", payload, nil)
} }
@@ -216,7 +229,13 @@ func processCommands(ctx context.Context, client *http.Client, cfg agentConfig,
return err return err
} }
for _, item := range list.Commands { for _, item := range list.Commands {
result, runErr := executeCommand(item) var result, reexecPath string
var runErr error
if item.Type == "self_upgrade" {
result, reexecPath, runErr = selfUpgrade(cfg, item.Payload)
} else {
result, runErr = executeCommand(item)
}
status := "completed" status := "completed"
errText := "" errText := ""
if runErr != nil { if runErr != nil {
@@ -232,6 +251,12 @@ func processCommands(ctx context.Context, client *http.Client, cfg agentConfig,
if err := postJSON(ctx, client, cfg, path, payload, nil); err != nil { if err := postJSON(ctx, client, cfg, path, payload, nil); err != nil {
log.Printf("report command result failed: command=%s err=%v", item.ID, err) log.Printf("report command result failed: command=%s err=%v", item.ID, err)
} }
// Re-exec AFTER the result posts — the new binary picks up the loop.
if reexecPath != "" {
if err := syscall.Exec(reexecPath, os.Args, os.Environ()); err != nil {
log.Fatalf("re-exec after upgrade failed: %v", err)
}
}
} }
return nil return nil
} }
@@ -257,6 +282,67 @@ func executeCommand(item command) (string, error) {
} }
} }
// selfUpgrade downloads the agent binary the server bundles for this
// platform, verifies it, swaps it in atomically, and returns the path to
// re-exec. On any failure the running binary is left untouched.
func selfUpgrade(cfg agentConfig, payload map[string]interface{}) (string, string, error) {
platform := firstString(stringValue(payload["platform"]), runtime.GOOS+"-"+runtime.GOARCH)
exe, err := os.Executable()
if err != nil {
return "", "", fmt.Errorf("resolve own path: %w", err)
}
if resolved, err := filepath.EvalSymlinks(exe); err == nil {
exe = resolved
}
url := fmt.Sprintf("%s/api/agents/download/%s", cfg.APIURL, platform)
req, err := http.NewRequest(http.MethodGet, url, nil)
if err != nil {
return "", "", err
}
resp, err := (&http.Client{Timeout: 2 * time.Minute}).Do(req)
if err != nil {
return "", "", fmt.Errorf("download %s: %w", platform, err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return "", "", fmt.Errorf("download %s: server returned %d", platform, resp.StatusCode)
}
newPath := exe + ".new"
out, err := os.OpenFile(newPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o755)
if err != nil {
return "", "", fmt.Errorf("write %s: %w", newPath, err)
}
h := sha256.New()
if _, err := io.Copy(io.MultiWriter(out, h), resp.Body); err != nil {
out.Close()
os.Remove(newPath)
return "", "", fmt.Errorf("download write: %w", err)
}
out.Close()
if want := stringValue(payload["sha256"]); want != "" {
if got := hex.EncodeToString(h.Sum(nil)); got != want {
os.Remove(newPath)
return "", "", fmt.Errorf("sha256 mismatch: got %s want %s", got, want)
}
}
// Sanity: the new binary must at least report its version.
if verOut, err := exec.Command(newPath, "--version").CombinedOutput(); err != nil {
os.Remove(newPath)
return "", "", fmt.Errorf("new binary failed sanity check: %s", strings.TrimSpace(string(verOut)))
}
if err := os.Rename(newPath, exe); err != nil {
os.Remove(newPath)
return "", "", fmt.Errorf("swap binary: %w", err)
}
return fmt.Sprintf("upgraded to %s (%s)", firstString(stringValue(payload["version"]), "latest"), platform), exe, nil
}
// dockerPrune runs a bounded docker system prune. Volumes are excluded // dockerPrune runs a bounded docker system prune. Volumes are excluded
// unless the payload explicitly opts in — a stray volume prune destroys // unless the payload explicitly opts in — a stray volume prune destroys
// database data. // database data.
+76 -2
View File
@@ -11,6 +11,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"io"
"log" "log"
"math" "math"
"net/http" "net/http"
@@ -353,6 +354,7 @@ func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) {
IPAddress string `json:"ip_address" binding:"required"` IPAddress string `json:"ip_address" binding:"required"`
Port int `json:"port" binding:"required"` Port int `json:"port" binding:"required"`
Capabilities AgentCapabilities `json:"capabilities" binding:"required"` Capabilities AgentCapabilities `json:"capabilities" binding:"required"`
Version string `json:"version"`
AuthToken string `json:"auth_token"` AuthToken string `json:"auth_token"`
Mesh map[string]string `json:"mesh"` Mesh map[string]string `json:"mesh"`
} }
@@ -398,7 +400,8 @@ func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) {
IpAddress: existing.IpAddress, IpAddress: existing.IpAddress,
Port: int32(req.Port), Port: int32(req.Port),
Status: sql.NullString{String: "connecting", Valid: true}, Status: sql.NullString{String: "connecting", Valid: true},
Version: existing.Version, // Older agents don't report a version — keep whatever's stored.
Version: sql.NullString{String: firstNonEmpty(req.Version, existing.Version.String), Valid: true},
Capabilities: rawJSON(req.Capabilities), Capabilities: rawJSON(req.Capabilities),
Resources: existing.Resources, Resources: existing.Resources,
LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true}, LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true},
@@ -450,7 +453,7 @@ func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) {
IpAddress: req.IPAddress, IpAddress: req.IPAddress,
Port: int32(req.Port), Port: int32(req.Port),
Status: sql.NullString{String: "connecting", Valid: true}, Status: sql.NullString{String: "connecting", Valid: true},
Version: sql.NullString{}, Version: sql.NullString{String: req.Version, Valid: req.Version != ""},
Capabilities: rawJSON(req.Capabilities), Capabilities: rawJSON(req.Capabilities),
Resources: rawJSON(resources), Resources: rawJSON(resources),
LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true}, LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true},
@@ -911,6 +914,76 @@ func (h *NodeAgentHandler) PruneAgent(c *gin.Context) {
c.JSON(http.StatusAccepted, gin.H{"command": command}) c.JSON(http.StatusAccepted, gin.H{"command": command})
} }
// UpgradeAgent enqueues a self_upgrade command: the agent downloads its
// replacement binary from this server's bundled /download endpoint,
// verifies the sha256, swaps it in, and re-execs. Fails fast when no
// bundled binary exists for the agent's architecture.
func (h *NodeAgentHandler) UpgradeAgent(c *gin.Context) {
ctx := c.Request.Context()
id := c.Param("id")
agent, err := h.q.GetAgent(ctx, id)
if err != nil {
respondError(c, http.StatusNotFound, "NOT_FOUND", "Agent not found")
return
}
if agent.Status.String != "online" {
respondError(c, http.StatusConflict, "NODE_OFFLINE", "node is offline — reconnect it before upgrading")
return
}
arch := "amd64"
var caps struct {
SupportedArchitectures []string `json:"supported_architectures"`
}
unmarshalRaw(agent.Capabilities, &caps)
if len(caps.SupportedArchitectures) > 0 && caps.SupportedArchitectures[0] != "" {
arch = caps.SupportedArchitectures[0]
}
platform := "linux-" + arch // agents only ship for linux today
dir := os.Getenv("CONTAINR_AGENT_BIN_DIR")
if dir == "" {
dir = "/agents"
}
path := filepath.Join(dir, "containr-agent-"+platform)
sum, err := sha256File(path)
if err != nil {
respondError(c, http.StatusConflict, "NO_BINARY",
fmt.Sprintf("no bundled agent binary for %s on this server", platform))
return
}
command, err := h.q.CreateCommand(ctx, sqlcdb.CreateCommandParams{
ID: uuid.New().String(),
Type: "self_upgrade",
NodeAgentID: id,
Payload: rawJSON(map[string]interface{}{
"platform": platform,
"sha256": sum,
"version": Version,
}),
})
if err != nil {
respondError(c, http.StatusInternalServerError, "INTERNAL", "failed to enqueue upgrade")
return
}
LogAuditWithRequest(c, "node", id, "upgrade", map[string]interface{}{"platform": platform, "version": Version})
c.JSON(http.StatusAccepted, gin.H{"command": command, "platform": platform, "version": Version})
}
func sha256File(path string) (string, error) {
f, err := os.Open(path)
if err != nil {
return "", err
}
defer f.Close()
h := sha256.New()
if _, err := io.Copy(h, f); err != nil {
return "", err
}
return hex.EncodeToString(h.Sum(nil)), nil
}
// ExecuteCommand executes a command on an agent // ExecuteCommand executes a command on an agent
func (h *NodeAgentHandler) ExecuteCommand(c *gin.Context) { func (h *NodeAgentHandler) ExecuteCommand(c *gin.Context) {
agentID := c.Param("id") agentID := c.Param("id")
@@ -1390,6 +1463,7 @@ func (h *NodeAgentHandler) SetupAdminRoutes(router *gin.RouterGroup) {
agents.POST("/:id/cordon", h.CordonAgent) agents.POST("/:id/cordon", h.CordonAgent)
agents.POST("/:id/uncordon", h.UncordonAgent) agents.POST("/:id/uncordon", h.UncordonAgent)
agents.POST("/:id/drain", h.DrainAgent) agents.POST("/:id/drain", h.DrainAgent)
agents.POST("/:id/upgrade", h.UpgradeAgent)
agents.POST("/:id/prune", h.PruneAgent) agents.POST("/:id/prune", h.PruneAgent)
agents.POST("/:id/commands", h.ExecuteCommand) agents.POST("/:id/commands", h.ExecuteCommand)
agents.GET("/:id/commands", h.GetAgentCommands) agents.GET("/:id/commands", h.GetAgentCommands)
+4 -2
View File
@@ -425,9 +425,10 @@ func runDeploymentAndSyncWithImage(
// on the local host, so remote placement only works for registry-pulled // on the local host, so remote placement only works for registry-pulled
// images — git-sourced and rollback deploys fail fast with that message. // images — git-sourced and rollback deploys fail fast with that message.
nodeID := serviceNodeID(db, service.ID) nodeID := serviceNodeID(db, service.ID)
if nodeID != "" && (service.GitRepo != "" || imageOverride != "") { spread := serviceSpread(db, service.ID)
if (nodeID != "" || spread) && (service.GitRepo != "" || imageOverride != "") {
failedAt := time.Now() 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( _, _ = db.Exec(
`UPDATE deployments SET status = 'failed', error = $1, completed_at = $2, updated_at = $2 WHERE id = $3`, `UPDATE deployments SET status = 'failed', error = $1, completed_at = $2, updated_at = $2 WHERE id = $3`,
msg, failedAt, dbDeployment.ID, msg, failedAt, dbDeployment.ID,
@@ -462,6 +463,7 @@ func runDeploymentAndSyncWithImage(
MaintenanceURL: maintURL, MaintenanceURL: maintURL,
BasicAuthUsers: basicAuthUsers, BasicAuthUsers: basicAuthUsers,
NodeID: nodeID, NodeID: nodeID,
Spread: spread,
}, },
Trigger: deployment.TriggerConfig{ Trigger: deployment.TriggerConfig{
Type: req.Trigger, Type: req.Trigger,
+107 -42
View File
@@ -48,6 +48,65 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
if !agent.Status.Valid || agent.Status.String != "online" { 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 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 != "" { 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") return nil, fmt.Errorf("domains are routed by the local Traefik — remove the service's domains or run it on the local node")
} }
@@ -57,31 +116,68 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
replicas = 1 replicas = 1
} }
// Retire replicas above the desired count using inventory rows.
prefix := fmt.Sprintf("containr-%s-", spec.ServiceID) 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 { for _, row := range existing {
if row.ServiceID != spec.ServiceID || !strings.HasPrefix(row.Name, prefix) { if !strings.HasPrefix(row.Name, prefix) {
continue continue
} }
replica, err := strconv.Atoi(strings.TrimPrefix(row.Name, prefix)) replica, err := strconv.Atoi(strings.TrimPrefix(row.Name, prefix))
if err == nil && replica >= replicas { keep := err == nil && replica < replicas && row.NodeAgentID == targetFor(replica)
if cmdErr := r.enqueueAndWait(ctx, spec.NodeID, "", "remove_container", if keep {
continue
}
if cmdErr := r.enqueueAndWait(ctx, row.NodeAgentID, "", "remove_container",
map[string]interface{}{"container_name": row.Name}); cmdErr != nil { map[string]interface{}{"container_name": row.Name}); cmdErr != nil {
return nil, fmt.Errorf("remove stale replica %s: %w", row.Name, cmdErr) return nil, fmt.Errorf("remove stale replica %s on node %s: %w", row.Name, row.NodeAgentID, cmdErr)
} }
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{ _ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: row.ID, ID: row.ID,
Status: rawJSON(map[string]interface{}{"state": "removed"}), Status: rawJSON(map[string]interface{}{"state": "removed"}),
}) })
} }
}
state := &deployment.RuntimeState{Desired: replicas, Status: "running"} state := &deployment.RuntimeState{Desired: replicas, Status: "running"}
for i := 0; i < replicas; i++ { for i := 0; i < replicas; i++ {
agentID := targetFor(i)
name := remoteContainerName(spec.ServiceID, i) name := remoteContainerName(spec.ServiceID, i)
containerID := remoteContainerID(spec.NodeID, spec.ServiceID, i) 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,
Name: name,
Image: spec.Image,
ProjectID: spec.ProjectID,
ServiceID: spec.ServiceID,
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),
Environment: rawJSON(env),
Volumes: rawJSON(volumes),
Networks: rawJSON([]string{}),
RestartPolicy: rawJSON(map[string]interface{}{"name": restart}),
HealthCheck: rawJSON(map[string]interface{}{"path": spec.HealthPath}),
})
if err := r.enqueueAndWait(ctx, agentID, containerID, "create_container",
map[string]interface{}{"container": container}); err != nil {
return nil, fmt.Errorf("replica %d on node %s: %w", i, agentID, err)
}
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: containerID,
Status: rawJSON(map[string]interface{}{"state": "running", "health": "none"}),
})
state.Containers = append(state.Containers, deployment.RuntimeContainer{
ID: containerID, Name: name, State: "running", Replica: i,
})
}
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{}{} ports := []interface{}{}
if spec.Port > 0 { if spec.Port > 0 {
hostPort := spec.PublishedPort // 0 → ephemeral on the node hostPort := spec.PublishedPort // 0 → ephemeral on the node
@@ -113,7 +209,7 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
for _, arg := range spec.Command { for _, arg := range spec.Command {
cmd = append(cmd, arg) cmd = append(cmd, arg)
} }
container := map[string]interface{}{ return map[string]interface{}{
"name": name, "name": name,
"image": spec.Image, "image": spec.Image,
"command": cmd, "command": cmd,
@@ -129,40 +225,9 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
"containr.project": spec.ProjectID, "containr.project": spec.ProjectID,
"containr.service": spec.ServiceID, "containr.service": spec.ServiceID,
"containr.replica": fmt.Sprint(i), "containr.replica": fmt.Sprint(i),
"containr.node": spec.NodeID, "containr.node": agentID,
}, },
} }, env, ports, volumes, restart
_ = r.q.UpsertServiceContainer(ctx, sqlcdb.UpsertServiceContainerParams{
ID: containerID,
Name: name,
Image: spec.Image,
ProjectID: spec.ProjectID,
ServiceID: spec.ServiceID,
NodeAgentID: spec.NodeID,
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),
Environment: rawJSON(env),
Volumes: rawJSON(volumes),
Networks: rawJSON([]string{}),
RestartPolicy: rawJSON(map[string]interface{}{"name": restart}),
HealthCheck: rawJSON(map[string]interface{}{"path": spec.HealthPath}),
})
if err := r.enqueueAndWait(ctx, spec.NodeID, containerID, "create_container",
map[string]interface{}{"container": container}); err != nil {
return nil, fmt.Errorf("replica %d on %s: %w", i, agent.Name, err)
}
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: containerID,
Status: rawJSON(map[string]interface{}{"state": "running", "health": "none"}),
})
state.Containers = append(state.Containers, deployment.RuntimeContainer{
ID: containerID, Name: name, State: "running", Replica: i,
})
}
return state, nil
} }
// pullCredentials resolves the project owner's registry auth for the image // pullCredentials resolves the project owner's registry auth for the image
+72 -12
View File
@@ -18,13 +18,37 @@ import (
"github.com/google/uuid" "github.com/google/uuid"
) )
// serviceLifecycle routes a start/stop/restart action to the node agent when // remoteAgentsForService lists every agent holding inventory rows for the
// the service is pinned, or to the local Docker engine otherwise. Remote // service — covers pinned placements, spread replicas, and leftovers from a
// actions block on the agent's poll loop; that matches the synchronous // since-cleared pin.
// behavior the local path already has. 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 { func serviceLifecycle(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, serviceID uuid.UUID, action string) error {
if nodeID := serviceNodeID(db, serviceID); nodeID != "" { for _, agentID := range remoteAgentsForService(ctx, db, serviceID) {
return engine.ControlServiceOnNode(ctx, serviceID.String(), nodeID, action) if err := engine.ControlServiceOnNode(ctx, serviceID.String(), agentID, action); err != nil {
return err
}
} }
switch action { switch action {
case "start": 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) return fmt.Errorf("unsupported lifecycle action %q", action)
} }
// removeServiceRuntime tears down replicas wherever they run — remote node // removeServiceRuntime tears down replicas wherever they run — remote nodes
// when pinned, local Docker otherwise. // (pinned or spread) and the local Docker host.
func removeServiceRuntime(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, serviceID uuid.UUID) { func removeServiceRuntime(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, serviceID uuid.UUID) {
if nodeID := serviceNodeID(db, serviceID); nodeID != "" { for _, agentID := range remoteAgentsForService(ctx, db, serviceID) {
_ = engine.RemoveServiceContainersOnNode(ctx, serviceID.String(), nodeID) _ = engine.RemoveServiceContainersOnNode(ctx, serviceID.String(), agentID)
return
} }
_ = engine.RemoveServiceContainers(ctx, serviceID.String()) _ = engine.RemoveServiceContainers(ctx, serviceID.String())
} }
@@ -70,6 +93,7 @@ func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpe
HealthPath: service.HealthCheckPath, HealthPath: service.HealthCheckPath,
RestartPolicy: service.RestartPolicy, RestartPolicy: service.RestartPolicy,
NodeID: serviceNodeID(db, service.ID), NodeID: serviceNodeID(db, service.ID),
Spread: serviceSpread(db, service.ID),
} }
// Best effort: reuse the host port from the last live deployment so the // Best effort: reuse the host port from the last live deployment so the
// public URL survives restarts. Column exists post-migration; older DBs // 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()}) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return 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) engine.ProbeServiceHealth(c.Request.Context(), state, service.ProjectID.String(), service.Name, service.Port, service.HealthCheckPath)
persistPublishedPort(db, service.ID, state.Ports) persistPublishedPort(db, service.ID, state.Ports)
if host := requestHostname(c); host != "" { if host := requestHostname(c); host != "" {
@@ -564,6 +589,37 @@ func enqueueServiceRedeploy(c *gin.Context, db *database.DB, service Service) bo
return true 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 // persistPublishedPort records the host port Docker actually bound so later
// redeploys can reuse it. Idempotent — only writes when the value changed. // redeploys can reuse it. Idempotent — only writes when the value changed.
func persistPublishedPort(db *database.DB, serviceID uuid.UUID, ports []uint16) { 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) ctx, cancel := context.WithTimeout(c.Request.Context(), 3*time.Second)
defer cancel() defer cancel()
state, err := engine.RuntimeState(ctx, service.ID.String(), service.Replicas) 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 return
} }
if state.Status != service.Status { if state.Status != service.Status {
+50 -5
View File
@@ -53,6 +53,9 @@ type Service struct {
// resolved "auto" picks the least-loaded online agent at write time). // resolved "auto" picks the least-loaded online agent at write time).
NodeID string `json:"node_id,omitempty" db:"-"` NodeID string `json:"node_id,omitempty" db:"-"`
NodeName string `json:"node_name,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"` CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"` UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
} }
@@ -79,10 +82,21 @@ func serviceNodeID(db *database.DB, serviceID uuid.UUID) string {
return "" 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) { func loadServiceNode(db *database.DB, s *Service) {
var name sql.NullString var name sql.NullString
s.NodeID = serviceNodeID(db, s.ID) s.NodeID = serviceNodeID(db, s.ID)
s.Spread = serviceSpread(db, s.ID)
if s.NodeID == "" { if s.NodeID == "" {
return return
} }
@@ -171,6 +185,9 @@ type CreateServiceRequest struct {
// NodeID pins placement to a node agent; "auto" picks the least-loaded // NodeID pins placement to a node agent; "auto" picks the least-loaded
// online agent, "local"/"" runs on the Containr host. // online agent, "local"/"" runs on the Containr host.
NodeID string `json:"node_id"` 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 // UpdateServiceRequest represents a request to update a service
@@ -202,6 +219,9 @@ type UpdateServiceRequest struct {
SleepIdleMinutes *int `json:"sleep_idle_minutes"` SleepIdleMinutes *int `json:"sleep_idle_minutes"`
// NodeID pins placement; pointer so "local"/"" explicitly clears the pin. // NodeID pins placement; pointer so "local"/"" explicitly clears the pin.
NodeID *string `json:"node_id"` 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 // handleGetServices retrieves all services for a project
@@ -442,14 +462,23 @@ func handleCreateService(c *gin.Context) {
sourceType := inferServiceSourceType(service) sourceType := inferServiceSourceType(service)
if req.Spread != nil {
service.Spread = *req.Spread
}
if req.NodeID != "" { 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) nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), req.NodeID)
if nodeErr != nil { if nodeErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"}) c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
return return
} }
if !service.Spread {
service.NodeID = nodeID service.NodeID = nodeID
} }
}
// Insert service into database // Insert service into database
_, err = db.(*database.DB).Exec( _, err = db.(*database.DB).Exec(
@@ -459,9 +488,9 @@ func handleCreateService(c *gin.Context) {
git_repo, git_branch, build_path, cpu, memory, replicas, port, domain, git_repo, git_branch, build_path, cpu, memory, replicas, port, domain,
healthcheck_path, restart_policy, volumes, healthcheck_path, restart_policy, volumes,
builder, cpu_reserve, memory_reserve, static_build_cmd, static_dir, 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, 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, service.ID, service.ProjectID, service.Name, environmentID, service.Type,
sourceType, firstNonEmpty(service.GitRepo, service.Image), service.Image, sourceType, firstNonEmpty(service.GitRepo, service.Image), service.Image,
"", service.Command, service.Type, service.Status, service.Image, service.Command, "", 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, volumesJSON, service.Builder, service.CPUReserve, service.MemoryReserve,
service.StaticBuildCmd, service.StaticDir, service.StaticBuildCmd, service.StaticDir,
sql.NullString{String: service.NodeID, Valid: service.NodeID != ""}, sql.NullString{String: service.NodeID, Valid: service.NodeID != ""},
service.Spread,
service.CreatedAt, service.UpdatedAt, service.CreatedAt, service.UpdatedAt,
) )
@@ -772,8 +802,18 @@ func handleUpdateService(c *gin.Context) {
volumesArg = volumesJSON volumesArg = volumesJSON
} }
if req.Spread != nil {
existingService.Spread = *req.Spread
} else {
existingService.Spread = serviceSpread(db.(*database.DB), serviceID)
}
if req.NodeID != nil { 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 { if nodeErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"}) c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
return return
@@ -782,6 +822,10 @@ func handleUpdateService(c *gin.Context) {
} else { } else {
existingService.NodeID = serviceNodeID(db.(*database.DB), serviceID) 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() existingService.UpdatedAt = time.Now()
@@ -794,7 +838,7 @@ func handleUpdateService(c *gin.Context) {
restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16, restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16,
builder = $19, cpu_reserve = $20, memory_reserve = $21, builder = $19, cpu_reserve = $20, memory_reserve = $21,
static_build_cmd = $22, static_dir = $23, 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`, WHERE id = $18`,
existingService.Name, existingService.Type, existingService.Image, existingService.Command, existingService.Name, existingService.Type, existingService.Image, existingService.Command,
existingService.Environment, existingService.GitRepo, existingService.GitBranch, existingService.Environment, existingService.GitRepo, existingService.GitBranch,
@@ -806,6 +850,7 @@ func handleUpdateService(c *gin.Context) {
existingService.StaticBuildCmd, existingService.StaticDir, existingService.StaticBuildCmd, existingService.StaticDir,
existingService.SleepEnabled, existingService.SleepIdleMinutes, existingService.SleepEnabled, existingService.SleepIdleMinutes,
sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""}, sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""},
existingService.Spread,
) )
if err != nil { if err != nil {
+19 -1
View File
@@ -190,6 +190,24 @@ var nodesDrainCmd = &cobra.Command{
}, },
} }
var nodesUpgradeCmd = &cobra.Command{
Use: "upgrade <id>",
Short: "Upgrade the node agent to the server's bundled binary",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
c, err := client()
if err != nil {
return err
}
data, err := c.Do("POST", "/agents/"+args[0]+"/upgrade", nil)
if err != nil {
return err
}
PrintRaw(data)
return nil
},
}
var nodesCommandsCmd = &cobra.Command{ var nodesCommandsCmd = &cobra.Command{
Use: "commands <id>", Use: "commands <id>",
Short: "List recent commands sent to a node agent", Short: "List recent commands sent to a node agent",
@@ -277,5 +295,5 @@ func init() {
nodesPruneCmd.Flags().String("until", "", "only prune objects older than this duration (e.g. 168h)") nodesPruneCmd.Flags().String("until", "", "only prune objects older than this duration (e.g. 168h)")
nodesPruneCmd.Flags().Bool("volumes", false, "also prune volumes (destructive — asks twice)") nodesPruneCmd.Flags().Bool("volumes", false, "also prune volumes (destructive — asks twice)")
nodeTokensCmd.AddCommand(nodeTokensIssueCmd, nodeTokensListCmd, nodeTokensRevokeCmd) nodeTokensCmd.AddCommand(nodeTokensIssueCmd, nodeTokensListCmd, nodeTokensRevokeCmd)
NodesCmd.AddCommand(nodesListCmd, nodesGetCmd, nodesUpdateCmd, nodesCordonCmd, nodesUncordonCmd, nodesDrainCmd, nodesPruneCmd, nodesCommandsCmd, nodesDeleteCmd, nodeTokensCmd) NodesCmd.AddCommand(nodesListCmd, nodesGetCmd, nodesUpdateCmd, nodesCordonCmd, nodesUncordonCmd, nodesDrainCmd, nodesUpgradeCmd, nodesPruneCmd, nodesCommandsCmd, nodesDeleteCmd, nodeTokensCmd)
} }
@@ -114,6 +114,9 @@ var servicesCreateCmd = &cobra.Command{
if v, _ := cmd.Flags().GetInt("replicas"); v > 0 { if v, _ := cmd.Flags().GetInt("replicas"); v > 0 {
body["replicas"] = v body["replicas"] = v
} }
if v, _ := cmd.Flags().GetBool("spread"); v {
body["spread"] = v
}
data, err := c.Do("POST", "/projects/"+args[0]+"/services", body) data, err := c.Do("POST", "/projects/"+args[0]+"/services", body)
if err != nil { if err != nil {
return err return err
@@ -203,6 +206,10 @@ var servicesUpdateCmd = &cobra.Command{
v, _ := cmd.Flags().GetBool("sleep") v, _ := cmd.Flags().GetBool("sleep")
body["sleep_enabled"] = v body["sleep_enabled"] = v
} }
if cmd.Flags().Changed("spread") {
v, _ := cmd.Flags().GetBool("spread")
body["spread"] = v
}
if cmd.Flags().Changed("sleep-idle") { if cmd.Flags().Changed("sleep-idle") {
v, _ := cmd.Flags().GetInt("sleep-idle") v, _ := cmd.Flags().GetInt("sleep-idle")
body["sleep_idle_minutes"] = v body["sleep_idle_minutes"] = v
@@ -495,6 +502,7 @@ func init() {
f.String("static-cmd", "", "build command for the static builder") f.String("static-cmd", "", "build command for the static builder")
f.String("static-dir", "", "output dir for the static builder (default dist)") f.String("static-dir", "", "output dir for the static builder (default dist)")
f.String("node", "", "pin to a node agent id, or 'auto' for least-loaded (image services only)") f.String("node", "", "pin to a node agent id, or 'auto' for least-loaded (image services only)")
f.Bool("spread", false, "distribute replicas across all online schedulable nodes")
uf := servicesUpdateCmd.Flags() uf := servicesUpdateCmd.Flags()
uf.String("name", "", "service name") uf.String("name", "", "service name")
@@ -520,6 +528,7 @@ func init() {
uf.Bool("sleep", false, "enable scale-to-zero on idle (sleep mode)") uf.Bool("sleep", false, "enable scale-to-zero on idle (sleep mode)")
uf.Int("sleep-idle", 0, "idle minutes before sleeping (1-1440)") uf.Int("sleep-idle", 0, "idle minutes before sleeping (1-1440)")
uf.String("node", "", "pin to a node agent id, 'auto' for least-loaded, 'local' to clear") uf.String("node", "", "pin to a node agent id, 'auto' for least-loaded, 'local' to clear")
uf.Bool("spread", false, "spread replicas across all online schedulable nodes (--spread=false to disable)")
for _, a := range []string{"start", "stop", "restart", "redeploy", "sleep", "wake"} { for _, a := range []string{"start", "stop", "restart", "redeploy", "sleep", "wake"} {
verb := a verb := a
@@ -691,6 +691,80 @@ func (q *Queries) ListPendingCommands(ctx context.Context, nodeAgentID string) (
return items, nil return items, nil
} }
const listSchedulableAgents = `-- name: ListSchedulableAgents :many
SELECT id, name FROM node_agents
WHERE status = 'online' AND schedulable
ORDER BY id
`
type ListSchedulableAgentsRow struct {
ID string `json:"id"`
Name string `json:"name"`
}
func (q *Queries) ListSchedulableAgents(ctx context.Context) ([]ListSchedulableAgentsRow, error) {
rows, err := q.db.QueryContext(ctx, listSchedulableAgents)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListSchedulableAgentsRow{}
for rows.Next() {
var i ListSchedulableAgentsRow
if err := rows.Scan(&i.ID, &i.Name); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listServiceContainers = `-- name: ListServiceContainers :many
SELECT id, name, node_agent_id, status FROM container_instances
WHERE service_id = $1
`
type ListServiceContainersRow struct {
ID string `json:"id"`
Name string `json:"name"`
NodeAgentID string `json:"node_agent_id"`
Status pqtype.NullRawMessage `json:"status"`
}
func (q *Queries) ListServiceContainers(ctx context.Context, serviceID string) ([]ListServiceContainersRow, error) {
rows, err := q.db.QueryContext(ctx, listServiceContainers, serviceID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListServiceContainersRow{}
for rows.Next() {
var i ListServiceContainersRow
if err := rows.Scan(
&i.ID,
&i.Name,
&i.NodeAgentID,
&i.Status,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const pickLeastLoadedAgent = `-- name: PickLeastLoadedAgent :one const pickLeastLoadedAgent = `-- name: PickLeastLoadedAgent :one
SELECT a.id FROM node_agents a SELECT a.id FROM node_agents a
LEFT JOIN LATERAL ( LEFT JOIN LATERAL (
@@ -699,10 +773,20 @@ LEFT JOIN LATERAL (
ORDER BY h.timestamp DESC LIMIT 1 ORDER BY h.timestamp DESC LIMIT 1
) h ON true ) h ON true
WHERE a.status = 'online' AND a.schedulable WHERE a.status = 'online' AND a.schedulable
ORDER BY COALESCE(h.container_count, 0) ASC, a.created_at ASC ORDER BY
COALESCE(
(a.resources->'memory'->>'used')::double precision
/ NULLIF((a.resources->'memory'->>'total')::double precision, 0),
1.0
) ASC,
COALESCE((a.resources->'cpu'->>'usage')::double precision, 100.0) ASC,
COALESCE(h.container_count, 0) ASC,
a.created_at ASC
LIMIT 1 LIMIT 1
` `
// Resource-aware: lowest memory utilisation first, then cpu, then container
// count. Agents without telemetry sort last (NULL → worst score).
func (q *Queries) PickLeastLoadedAgent(ctx context.Context) (string, error) { func (q *Queries) PickLeastLoadedAgent(ctx context.Context) (string, error) {
row := q.db.QueryRowContext(ctx, pickLeastLoadedAgent) row := q.db.QueryRowContext(ctx, pickLeastLoadedAgent)
var id string var id string
@@ -276,6 +276,7 @@ type Service struct {
StaticBuildCmd string `json:"static_build_cmd"` StaticBuildCmd string `json:"static_build_cmd"`
StaticDir string `json:"static_dir"` StaticDir string `json:"static_dir"`
NodeID sql.NullString `json:"node_id"` NodeID sql.NullString `json:"node_id"`
Spread bool `json:"spread"`
} }
type ServiceDomain struct { type ServiceDomain struct {
@@ -94,6 +94,8 @@ type Querier interface {
// display, when the caller owns or is a member of it, or when the caller is a // display, when the caller owns or is a member of it, or when the caller is a
// platform admin. Anonymous callers pass user_id = uuid.Nil, is_admin = false. // platform admin. Anonymous callers pass user_id = uuid.Nil, is_admin = false.
ListProjectsWithStatsByUser(ctx context.Context, arg ListProjectsWithStatsByUserParams) ([]ListProjectsWithStatsByUserRow, error) ListProjectsWithStatsByUser(ctx context.Context, arg ListProjectsWithStatsByUserParams) ([]ListProjectsWithStatsByUserRow, error)
ListSchedulableAgents(ctx context.Context) ([]ListSchedulableAgentsRow, error)
ListServiceContainers(ctx context.Context, serviceID string) ([]ListServiceContainersRow, error)
ListServiceTemplatesByCategoryForUser(ctx context.Context, arg ListServiceTemplatesByCategoryForUserParams) ([]ServiceTemplate, error) ListServiceTemplatesByCategoryForUser(ctx context.Context, arg ListServiceTemplatesByCategoryForUserParams) ([]ServiceTemplate, error)
ListServiceTemplatesForUser(ctx context.Context, ownerID uuid.NullUUID) ([]ServiceTemplate, error) ListServiceTemplatesForUser(ctx context.Context, ownerID uuid.NullUUID) ([]ServiceTemplate, error)
ListUserInvites(ctx context.Context) ([]UserInvite, error) ListUserInvites(ctx context.Context) ([]UserInvite, error)
@@ -102,6 +104,8 @@ type Querier interface {
MarkAllNotificationsReadByUser(ctx context.Context, arg MarkAllNotificationsReadByUserParams) error MarkAllNotificationsReadByUser(ctx context.Context, arg MarkAllNotificationsReadByUserParams) error
MarkNotificationReadByIDAndUser(ctx context.Context, arg MarkNotificationReadByIDAndUserParams) error MarkNotificationReadByIDAndUser(ctx context.Context, arg MarkNotificationReadByIDAndUserParams) error
MarkUserInviteUsed(ctx context.Context, arg MarkUserInviteUsedParams) error MarkUserInviteUsed(ctx context.Context, arg MarkUserInviteUsedParams) error
// Resource-aware: lowest memory utilisation first, then cpu, then container
// count. Agents without telemetry sort last (NULL → worst score).
PickLeastLoadedAgent(ctx context.Context) (string, error) PickLeastLoadedAgent(ctx context.Context) (string, error)
RevokeAgentAuthToken(ctx context.Context, id uuid.UUID) (AgentAuthToken, error) RevokeAgentAuthToken(ctx context.Context, id uuid.UUID) (AgentAuthToken, error)
RevokeUserToken(ctx context.Context, arg RevokeUserTokenParams) (int64, error) RevokeUserToken(ctx context.Context, arg RevokeUserTokenParams) (int64, error)
+10
View File
@@ -17,11 +17,17 @@ type NodeRunner interface {
// ReconcileOnNode brings the node's containers in line with the spec and // ReconcileOnNode brings the node's containers in line with the spec and
// returns the resulting runtime state. // returns the resulting runtime state.
ReconcileOnNode(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error) ReconcileOnNode(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error)
// ReconcileSpread distributes the service's replicas across every online,
// schedulable agent — replica i lands on agents[i % len].
ReconcileSpread(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error)
// ControlService fans a lifecycle action ("start"|"stop"|"restart"|"remove") // ControlService fans a lifecycle action ("start"|"stop"|"restart"|"remove")
// out to every replica of the service on the agent. // out to every replica of the service on the agent.
ControlService(ctx context.Context, serviceID, agentID, action string) error ControlService(ctx context.Context, serviceID, agentID, action string) error
// RemoveService tears down every replica of the service on the agent. // RemoveService tears down every replica of the service on the agent.
RemoveService(ctx context.Context, serviceID, agentID string) error RemoveService(ctx context.Context, serviceID, agentID string) error
// RemoteRuntimeState reads the service's remote replicas from inventory
// across all agents. nil/empty means nothing runs remotely.
RemoteRuntimeState(ctx context.Context, serviceID string) (*RuntimeState, error)
} }
type DeploymentEngine struct { type DeploymentEngine struct {
@@ -85,6 +91,9 @@ type ServiceConfig struct {
// host. Remote placement requires a registry-pullable image — builds stay // host. Remote placement requires a registry-pullable image — builds stay
// on the local host (build-on-node is not yet supported). // on the local host (build-on-node is not yet supported).
NodeID string `json:"node_id,omitempty"` NodeID string `json:"node_id,omitempty"`
// Spread distributes replicas across every online, schedulable agent.
// Same registry-image constraint as NodeID.
Spread bool `json:"spread,omitempty"`
} }
type PortMapping struct { type PortMapping struct {
@@ -376,6 +385,7 @@ func (de *DeploymentEngine) deployService(ctx context.Context, deployment *Deplo
MaintenanceURL: cfg.MaintenanceURL, MaintenanceURL: cfg.MaintenanceURL,
BasicAuthUsers: cfg.BasicAuthUsers, BasicAuthUsers: cfg.BasicAuthUsers,
NodeID: cfg.NodeID, NodeID: cfg.NodeID,
Spread: cfg.Spread,
} }
for _, pm := range cfg.PortMappings { for _, pm := range cfg.PortMappings {
if spec.Port == 0 { if spec.Port == 0 {
@@ -57,6 +57,9 @@ type RuntimeSpec struct {
BasicAuthUsers string // htpasswd-format user:hash pairs, comma-separated BasicAuthUsers string // htpasswd-format user:hash pairs, comma-separated
// NodeID pins the service to a node agent; "" runs on the local host. // NodeID pins the service to a node agent; "" runs on the local host.
NodeID string NodeID string
// Spread distributes replicas across all online, schedulable agents
// instead of a single node. Mutually exclusive with NodeID.
Spread bool
} }
// RuntimeContainer describes one live replica. // RuntimeContainer describes one live replica.
@@ -156,6 +159,12 @@ func (de *DeploymentEngine) ListServiceContainers(ctx context.Context, serviceID
// otherwise only the replica delta is applied. A pinned NodeID routes the // otherwise only the replica delta is applied. A pinned NodeID routes the
// whole reconcile to that agent's command queue instead of local Docker. // whole reconcile to that agent's command queue instead of local Docker.
func (de *DeploymentEngine) ReconcileService(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error) { func (de *DeploymentEngine) ReconcileService(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error) {
if spec.Spread && spec.NodeID == "" {
if de.nodeRunner == nil {
return nil, fmt.Errorf("service has spread placement but remote dispatch is unavailable")
}
return de.nodeRunner.ReconcileSpread(ctx, spec)
}
if spec.NodeID != "" { if spec.NodeID != "" {
if de.nodeRunner == nil { if de.nodeRunner == nil {
return nil, fmt.Errorf("service is pinned to node %s but remote dispatch is unavailable", spec.NodeID) return nil, fmt.Errorf("service is pinned to node %s but remote dispatch is unavailable", spec.NodeID)
@@ -526,6 +535,15 @@ func (de *DeploymentEngine) RemoveServiceContainersOnNode(ctx context.Context, s
return de.nodeRunner.RemoveService(ctx, serviceID, nodeID) return de.nodeRunner.RemoveService(ctx, serviceID, nodeID)
} }
// RemoteRuntimeState exposes inventory-backed remote container state; nil
// runner means no node support was wired at boot.
func (de *DeploymentEngine) RemoteRuntimeState(ctx context.Context, serviceID string) (*RuntimeState, error) {
if de.nodeRunner == nil {
return nil, nil
}
return de.nodeRunner.RemoteRuntimeState(ctx, serviceID)
}
// RemoveProjectContainers deletes every managed container in a project. // RemoveProjectContainers deletes every managed container in a project.
func (de *DeploymentEngine) RemoveProjectContainers(ctx context.Context, projectID string) error { func (de *DeploymentEngine) RemoveProjectContainers(ctx context.Context, projectID string) error {
containers, err := de.dockerClient.ListContainersFiltered(ctx, filters.NewArgs( containers, err := de.dockerClient.ListContainersFiltered(ctx, filters.NewArgs(
+3 -2
View File
@@ -53,7 +53,7 @@ func init() {
// Services // Services
register(spec{"containr_services_list", "List services in a project", "GET", "/projects/{project_id}/services", []string{"project_id"}, nil, false}) register(spec{"containr_services_list", "List services in a project", "GET", "/projects/{project_id}/services", []string{"project_id"}, nil, false})
register(spec{"containr_services_get", "Get a service by id", "GET", "/services/{id}", []string{"id"}, nil, false}) register(spec{"containr_services_get", "Get a service by id", "GET", "/services/{id}", []string{"id"}, nil, false})
register(spec{"containr_services_create", "Create a service in a project. Args: name (required), type, image, git_repo, git_branch, environment, port, domain, replicas, volumes = [{type: volume|bind, source, target, read_only}], builder = auto|railpack|nixpacks|dockerfile|static, cpu, memory, cpu_reserve, memory_reserve, static_build_cmd, static_dir, node_id (agent id | 'auto' = least-loaded online node; image services only — remote nodes can't run git builds).", "POST", "/projects/{project_id}/services", []string{"project_id"}, nil, false}) register(spec{"containr_services_create", "Create a service in a project. Args: name (required), type, image, git_repo, git_branch, environment, port, domain, replicas, volumes = [{type: volume|bind, source, target, read_only}], builder = auto|railpack|nixpacks|dockerfile|static, cpu, memory, cpu_reserve, memory_reserve, static_build_cmd, static_dir, node_id (agent id | 'auto' = least-loaded online node; image services only — remote nodes can't run git builds), spread (bool — distribute replicas across every online schedulable node; mutually exclusive with a node pin).", "POST", "/projects/{project_id}/services", []string{"project_id"}, nil, false})
register(spec{"containr_services_delete", "Delete a service. Requires confirm:true.", "DELETE", "/services/{id}", []string{"id"}, nil, true}) register(spec{"containr_services_delete", "Delete a service. Requires confirm:true.", "DELETE", "/services/{id}", []string{"id"}, nil, true})
register(spec{"containr_services_start", "Start a service", "POST", "/services/{id}/start", []string{"id"}, nil, false}) register(spec{"containr_services_start", "Start a service", "POST", "/services/{id}/start", []string{"id"}, nil, false})
register(spec{"containr_services_stop", "Stop a service", "POST", "/services/{id}/stop", []string{"id"}, nil, false}) register(spec{"containr_services_stop", "Stop a service", "POST", "/services/{id}/stop", []string{"id"}, nil, false})
@@ -66,7 +66,7 @@ func init() {
register(spec{"containr_services_move", "Move a service to another project the caller owns. Args: id, project_id (required).", "POST", "/services/{id}/move", []string{"id"}, []string{"project_id"}, false}) register(spec{"containr_services_move", "Move a service to another project the caller owns. Args: id, project_id (required).", "POST", "/services/{id}/move", []string{"id"}, []string{"project_id"}, false})
register(spec{"containr_services_logs", "Get service runtime logs. Args: id, tail (default 100).", "GET", "/services/{id}/logs", []string{"id"}, nil, false}) register(spec{"containr_services_logs", "Get service runtime logs. Args: id, tail (default 100).", "GET", "/services/{id}/logs", []string{"id"}, nil, false})
register(spec{"containr_services_exec", "Run a one-off command in the service container (30s ceiling). Args: id, command (required).", "POST", "/services/{id}/exec", []string{"id"}, []string{"command"}, false}) register(spec{"containr_services_exec", "Run a one-off command in the service container (30s ceiling). Args: id, command (required).", "POST", "/services/{id}/exec", []string{"id"}, []string{"command"}, false})
register(spec{"containr_services_update", "Update a service. Args: id, name, image, command, domain, restart_policy, healthcheck_path, cpu, memory, replicas, port, volumes = [{type: volume|bind, source, target, read_only}] — volumes replaces the whole mount list; pass [] to clear. maintenance_mode (bool), basic_auth = [{username, password}] — replaces all creds; [] clears. builder = auto|railpack|nixpacks|dockerfile|static, cpu_reserve, memory_reserve, static_build_cmd, static_dir (empty string clears reserves/static fields). sleep_enabled (bool) + sleep_idle_minutes (1-1440) enable scale-to-zero on idle; a wake placeholder answers traffic while the service resumes. node_id pins placement to a node agent ('auto' = least-loaded, 'local' = run on the Containr host).", "PUT", "/services/{id}", []string{"id"}, nil, false}) register(spec{"containr_services_update", "Update a service. Args: id, name, image, command, domain, restart_policy, healthcheck_path, cpu, memory, replicas, port, volumes = [{type: volume|bind, source, target, read_only}] — volumes replaces the whole mount list; pass [] to clear. maintenance_mode (bool), basic_auth = [{username, password}] — replaces all creds; [] clears. builder = auto|railpack|nixpacks|dockerfile|static, cpu_reserve, memory_reserve, static_build_cmd, static_dir (empty string clears reserves/static fields). sleep_enabled (bool) + sleep_idle_minutes (1-1440) enable scale-to-zero on idle; a wake placeholder answers traffic while the service resumes. node_id pins placement to a node agent ('auto' = least-loaded, 'local' = run on the Containr host). spread (bool) distributes replicas across every online schedulable node; mutually exclusive with a pin — clear node_id to 'local' first if pinned.", "PUT", "/services/{id}", []string{"id"}, nil, false})
// Domains // Domains
register(spec{"containr_domains_list", "List domains attached to a service", "GET", "/services/{id}/domains", []string{"id"}, nil, false}) register(spec{"containr_domains_list", "List domains attached to a service", "GET", "/services/{id}/domains", []string{"id"}, nil, false})
@@ -178,6 +178,7 @@ func init() {
register(spec{"containr_nodes_uncordon", "Re-open a node for placement (admin). Args: id.", "POST", "/agents/{id}/uncordon", []string{"id"}, nil, false}) register(spec{"containr_nodes_uncordon", "Re-open a node for placement (admin). Args: id.", "POST", "/agents/{id}/uncordon", []string{"id"}, nil, false})
register(spec{"containr_nodes_drain", "Drain a node: cordon + remove all service containers via the agent + unpin affected services so they redeploy locally. Requires the node online (admin). Args: id.", "POST", "/agents/{id}/drain", []string{"id"}, nil, true}) register(spec{"containr_nodes_drain", "Drain a node: cordon + remove all service containers via the agent + unpin affected services so they redeploy locally. Requires the node online (admin). Args: id.", "POST", "/agents/{id}/drain", []string{"id"}, nil, true})
register(spec{"containr_nodes_prune", "Enqueue a bounded docker system prune on a node (admin). Args: id; optional until (e.g. 168h), volumes (danger, off by default).", "POST", "/agents/{id}/prune", []string{"id"}, []string{"until", "volumes"}, true}) register(spec{"containr_nodes_prune", "Enqueue a bounded docker system prune on a node (admin). Args: id; optional until (e.g. 168h), volumes (danger, off by default).", "POST", "/agents/{id}/prune", []string{"id"}, []string{"until", "volumes"}, true})
register(spec{"containr_nodes_upgrade", "Upgrade a node's agent to the server's bundled binary: downloads over the API, verifies sha256, swaps atomically, re-execs. Node must be online (admin). Args: id.", "POST", "/agents/{id}/upgrade", []string{"id"}, nil, true})
register(spec{"containr_nodes_commands", "List recent commands sent to a node agent (admin). Args: id.", "GET", "/agents/{id}/commands", []string{"id"}, nil, false}) register(spec{"containr_nodes_commands", "List recent commands sent to a node agent (admin). Args: id.", "GET", "/agents/{id}/commands", []string{"id"}, nil, false})
// Scaling / HA / security // Scaling / HA / security
@@ -0,0 +1,5 @@
-- +goose Up
ALTER TABLE services ADD COLUMN IF NOT EXISTS spread BOOLEAN NOT NULL DEFAULT false;
-- +goose Down
ALTER TABLE services DROP COLUMN IF EXISTS spread;
+20 -1
View File
@@ -116,6 +116,8 @@ WHERE node_agent_id = $1 AND type = $2
ORDER BY created_at DESC LIMIT 1; ORDER BY created_at DESC LIMIT 1;
-- name: PickLeastLoadedAgent :one -- name: PickLeastLoadedAgent :one
-- Resource-aware: lowest memory utilisation first, then cpu, then container
-- count. Agents without telemetry sort last (NULL → worst score).
SELECT a.id FROM node_agents a SELECT a.id FROM node_agents a
LEFT JOIN LATERAL ( LEFT JOIN LATERAL (
SELECT h.container_count FROM agent_heartbeats h SELECT h.container_count FROM agent_heartbeats h
@@ -123,9 +125,26 @@ LEFT JOIN LATERAL (
ORDER BY h.timestamp DESC LIMIT 1 ORDER BY h.timestamp DESC LIMIT 1
) h ON true ) h ON true
WHERE a.status = 'online' AND a.schedulable WHERE a.status = 'online' AND a.schedulable
ORDER BY COALESCE(h.container_count, 0) ASC, a.created_at ASC ORDER BY
COALESCE(
(a.resources->'memory'->>'used')::double precision
/ NULLIF((a.resources->'memory'->>'total')::double precision, 0),
1.0
) ASC,
COALESCE((a.resources->'cpu'->>'usage')::double precision, 100.0) ASC,
COALESCE(h.container_count, 0) ASC,
a.created_at ASC
LIMIT 1; LIMIT 1;
-- name: ListSchedulableAgents :many
SELECT id, name FROM node_agents
WHERE status = 'online' AND schedulable
ORDER BY id;
-- name: ListServiceContainers :many
SELECT id, name, node_agent_id, status FROM container_instances
WHERE service_id = $1;
-- name: SetAgentSchedulable :exec -- name: SetAgentSchedulable :exec
UPDATE node_agents SET schedulable = $2, updated_at = NOW() WHERE id = $1; UPDATE node_agents SET schedulable = $2, updated_at = NOW() WHERE id = $1;
+2 -1
View File
@@ -67,7 +67,8 @@ CREATE TABLE services (
memory_reserve VARCHAR(20) NOT NULL DEFAULT '', memory_reserve VARCHAR(20) NOT NULL DEFAULT '',
static_build_cmd VARCHAR(255) NOT NULL DEFAULT '', static_build_cmd VARCHAR(255) NOT NULL DEFAULT '',
static_dir VARCHAR(255) NOT NULL DEFAULT '', static_dir VARCHAR(255) NOT NULL DEFAULT '',
node_id VARCHAR(255) node_id VARCHAR(255),
spread BOOLEAN NOT NULL DEFAULT false
); );
CREATE TABLE registries ( CREATE TABLE registries (
@@ -11,6 +11,7 @@ import {
listAgentContainers, listAgentContainers,
pruneAgent, pruneAgent,
updateAgent, updateAgent,
upgradeAgent,
type AgentMetricPoint, type AgentMetricPoint,
} from '@/lib/api-client'; } from '@/lib/api-client';
import { getCurrentUserProfile } from '@/lib/api-client'; import { getCurrentUserProfile } from '@/lib/api-client';
@@ -97,6 +98,10 @@ export function NodeDetailPage() {
mutationFn: () => drainAgent(id), mutationFn: () => drainAgent(id),
onSuccess: () => { setConfirmDrain(false); invalidate(); }, onSuccess: () => { setConfirmDrain(false); invalidate(); },
}); });
const upgradeMutation = useMutation({
mutationFn: () => upgradeAgent(id),
onSuccess: invalidate,
});
const agent = agentQuery.data; const agent = agentQuery.data;
const metrics = useMemo(() => metricsQuery.data ?? [], [metricsQuery.data]); const metrics = useMemo(() => metricsQuery.data ?? [], [metricsQuery.data]);
@@ -215,6 +220,16 @@ export function NodeDetailPage() {
> >
drain drain
</button> </button>
<button
onClick={() => upgradeMutation.mutate()}
disabled={upgradeMutation.isPending || !online}
title={online
? `Upgrade agent to the server's bundled binary (server ${upgradeMutation.data?.version ?? 'build'}) — the agent restarts itself`
: 'Node must be online to upgrade'}
className="v-mono text-[11px] px-3 py-1.5 rounded-md border border-[var(--border-subtle)] text-[var(--text-secondary)] hover:text-[var(--text-primary)] disabled:opacity-50"
>
{upgradeMutation.isPending ? 'upgrading…' : 'upgrade agent'}
</button>
<button <button
onClick={() => setConfirmVolumes(true)} onClick={() => setConfirmVolumes(true)}
disabled={pruneMutation.isPending} disabled={pruneMutation.isPending}
@@ -2404,10 +2404,12 @@ export function ServiceDetailPage() {
</div> </div>
<div className="flex items-center gap-3 text-xs"> <div className="flex items-center gap-3 text-xs">
<select <select
value={nodeDraft ?? (service.nodeId || 'local')} value={service.spread ? '__spread__' : (nodeDraft ?? (service.nodeId || 'local'))}
disabled={service.spread}
onChange={(e) => setNodeDraft(e.target.value)} onChange={(e) => setNodeDraft(e.target.value)}
className="px-2 py-1.5 rounded-[var(--radius-md)] bg-[var(--surface-muted)] border border-[var(--border-subtle)] text-sm text-[var(--text-primary)]" className="px-2 py-1.5 rounded-[var(--radius-md)] bg-[var(--surface-muted)] border border-[var(--border-subtle)] text-sm text-[var(--text-primary)] disabled:opacity-50"
> >
{service.spread && <option value="__spread__">All nodes (spread)</option>}
<option value="local">This server (local Docker)</option> <option value="local">This server (local Docker)</option>
<option value="auto">Auto — least-loaded online node</option> <option value="auto">Auto — least-loaded online node</option>
{(nodesQuery.data ?? []).map((n) => ( {(nodesQuery.data ?? []).map((n) => (
@@ -2416,6 +2418,17 @@ export function ServiceDetailPage() {
</option> </option>
))} ))}
</select> </select>
<label className="flex items-center gap-1.5 text-[var(--text-secondary)]">
<input
type="checkbox"
checked={service.spread ?? false}
disabled={updateServiceMutation.isPending || (!!service.nodeId && !service.spread)}
title={service.nodeId && !service.spread ? 'Clear the node pin first (choose "This server")' : undefined}
onChange={(e) => updateServiceMutation.mutate({ spread: e.target.checked })}
className="accent-[var(--accent-primary)]"
/>
spread replicas across nodes
</label>
{nodeDraft !== null && ( {nodeDraft !== null && (
<> <>
<button <button
+70
View File
@@ -8732,6 +8732,70 @@ export interface paths {
patch?: never; patch?: never;
trace?: never; trace?: never;
}; };
"/agents/{id}/upgrade": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Upgrade the node agent
* @description Enqueues a `self_upgrade` command. The agent downloads its replacement binary from this server's bundled `/agents/download` endpoint, verifies the sha256, atomically swaps the executable, and re-execs itself — the command result reports before the process swaps. The node must be online, and the server must bundle a binary for the node's architecture (`linux-amd64`/`linux-arm64`).
*/
post: {
parameters: {
query?: never;
header?: never;
path: {
/** @description Node agent ID */
id: components["parameters"]["AgentIdParam"];
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Upgrade command enqueued */
202: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": {
command?: Record<string, never>;
platform?: string;
version?: string;
};
};
};
/** @description Agent not found */
404: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["Error"];
};
};
/** @description Node offline or no bundled binary for its architecture */
409: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["Error"];
};
};
};
};
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/agents/{id}/metrics": { "/agents/{id}/metrics": {
parameters: { parameters: {
query?: never; query?: never;
@@ -10878,6 +10942,8 @@ export interface components {
node_id?: string; node_id?: string;
/** @description Display name of the pinned node agent (read-only) */ /** @description Display name of the pinned node agent (read-only) */
node_name?: string; node_name?: string;
/** @description Replicas are distributed across every online, schedulable node agent (replica i → agents[i mod n]). Mutually exclusive with an explicit node_id pin. */
spread?: boolean;
/** @description Computed public URL (published port or domain) */ /** @description Computed public URL (published port or domain) */
public_url?: string; public_url?: string;
/** /**
@@ -10950,6 +11016,8 @@ export interface components {
static_dir?: string; static_dir?: string;
/** @description Node agent id to pin to, 'auto' = least-loaded online agent; omit/empty = local host */ /** @description Node agent id to pin to, 'auto' = least-loaded online agent; omit/empty = local host */
node_id?: string; node_id?: string;
/** @description Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin */
spread?: boolean;
}; };
UpdateServiceRequest: { UpdateServiceRequest: {
/** @description Service name */ /** @description Service name */
@@ -11013,6 +11081,8 @@ export interface components {
sleep_idle_minutes?: number; sleep_idle_minutes?: number;
/** @description Node agent id to pin to, 'auto' = least-loaded online agent, 'local' = run on the Containr host (clears the pin) */ /** @description Node agent id to pin to, 'auto' = least-loaded online agent, 'local' = run on the Containr host (clears the pin) */
node_id?: string; node_id?: string;
/** @description Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin — send 'local' first to clear a pin */
spread?: boolean;
}; };
ServiceDomain: { ServiceDomain: {
/** Format: uuid */ /** Format: uuid */
+1 -1
View File
@@ -40,7 +40,7 @@ export const bundledDocs: BundledDoc[] = [
"path": "guides/CLI.md", "path": "guides/CLI.md",
"section": "guides", "section": "guides",
"title": "Containr CLI", "title": "Containr CLI",
"markdown": "# Containr CLI\n\nThe CLI controls every part of a Containr instance — the same API surface\nthe web UI uses. Build it with `scripts/build-cli.sh` or\n`go build ./cmd/cli` from `app/backend`.\n\n## Authentication\n\nCreate a personal access token in the web UI under\n**Settings → Personal Access Tokens** (or `containr tokens create`), then:\n\n```bash\ncontainr auth login cnp_... # verify + store\ncontainr auth login cnp_... --url https://containr.example.com\ncontainr auth login cnp_... --name production # named profile\ncontainr auth status # show active profile + identity\ncontainr auth logout\n```\n\nResolution order for every command: `--token`/`--api-url` flags >\n`CONTAINR_TOKEN`/`CONTAINR_API_URL` env > active profile >\n`http://localhost:8080/api/v1`. The active profile is `--profile` >\n`CONTAINR_PROFILE` > `current_profile` in `~/.containr.yaml`.\n\n## Output and scripting\n\n- `--json` prints the raw API response — safe to pipe and parse.\n- `--yes` / `-y` skips confirmation on destructive commands (delete,\n revoke, restore, failover). Non-interactive shells refuse destructive\n commands without it.\n- Exit codes are stable: `0` ok, `1` error, `2` auth/permission,\n `3` not found, `4` validation, `5` unreachable.\n\n## Commands\n\n```\nauth login|logout|status token auth and profiles\ntokens list|create|revoke personal access tokens\nprojects list|get|create|delete\nservices list|get|create|update|delete|start|stop|restart|redeploy\nservices clone|move duplicate config or re-parent a service\nservices domains list|add|remove|default|check multi-domain + DNS preflight\nregistries list|add|update|remove private-image pull credentials\nvolumes list|delete docker volume inventory + cleanup (admin)\ndeploy <service-id> trigger a deployment (--no-cache, --commit-hash, --branch)\ndeployments list|get|logs|rollback|cancel\nlogs <service-id> [--follow] runtime logs (SSE follow)\nexec <service-id> -- <cmd> one-off container command\nvariables list|set|unset env vars (secrets masked; --redeploy applies them)\nvariables shared list|set|unset project shared vars — referenced as ${{shared.KEY}}\ndatabases list|get|create|update|delete|action|backup|restore|download-backup|import-backup\n import-backup <id> <file.tar.gz|-> uploads an archive as a restore\n point; ships offsite too when a backup target is assigned\ndatabases list|get|create|update|delete|action|backup|restore|download-backup\n create --public exposes the port on all interfaces;\n update <id> --public/--public=false toggles it (container recreates, data persists)\ndatabases register <name> register an external database (--type --host --port --database --username --password --ssl; probed before storing)\ndatabases test-connection probe a connection — by id, or ad-hoc flags\nbackup-targets list|add|update|remove|test\n S3-compatible archive destinations; add/update probe the bucket\n first, empty --access-key/--secret-key on update keep stored keys;\n update <db-id> --backup-target assigns a target (\"\" clears it)\ncron list|get|create|delete|trigger|executions\ntemplates list|get|create|delete|deploy|plan|import-compose|deploy-graph\n deploy --name --var K=V; plan dry-resolves expressions;\n import-compose <compose.yml|-> converts a compose file into\n a v2 graph config; deploy-graph <project-id> <config.json|->\n deploys an ad-hoc service graph\nimport compose <project-id> <compose.yml|->\n deploys a docker-compose file straight into a project —\n no template row. DB images become managed databases\n attached under their service name; --var K=V overrides\n template variables\nnodes list|get|delete|tokens node agents + onboarding tokens (admin)\nnodes add <user@host> SSH bootstrap — one session installs the agent, then never needed again (--print for manual runs)\nnodes update|prune|commands rename/auto-prune, bounded docker prune, command history\nmigrate railway pull a Railway project (services, vars, domains) via GraphQL and recreate it — --dry-run prints the plan\nmigrate dflow import a dflow project (services, vars, volumes, domains) via its Payload REST API — needs --url + --api-key, --dry-run prints the plan\nscaling status|policies|services|scale\nha status|alerts|health|policies|enable|disable|failover\nsecurity scan|vulnerabilities|metrics\ngateway services|keys|anLine truncated "markdown": "# Containr CLI\n\nThe CLI controls every part of a Containr instance — the same API surface\nthe web UI uses. Build it with `scripts/build-cli.sh` or\n`go build ./cmd/cli` from `app/backend`.\n\n## Authentication\n\nCreate a personal access token in the web UI under\n**Settings → Personal Access Tokens** (or `containr tokens create`), then:\n\n```bash\ncontainr auth login cnp_... # verify + store\ncontainr auth login cnp_... --url https://containr.example.com\ncontainr auth login cnp_... --name production # named profile\ncontainr auth status # show active profile + identity\ncontainr auth logout\n```\n\nResolution order for every command: `--token`/`--api-url` flags >\n`CONTAINR_TOKEN`/`CONTAINR_API_URL` env > active profile >\n`http://localhost:8080/api/v1`. The active profile is `--profile` >\n`CONTAINR_PROFILE` > `current_profile` in `~/.containr.yaml`.\n\n## Output and scripting\n\n- `--json` prints the raw API response — safe to pipe and parse.\n- `--yes` / `-y` skips confirmation on destructive commands (delete,\n revoke, restore, failover). Non-interactive shells refuse destructive\n commands without it.\n- Exit codes are stable: `0` ok, `1` error, `2` auth/permission,\n `3` not found, `4` validation, `5` unreachable.\n\n## Commands\n\n```\nauth login|logout|status token auth and profiles\ntokens list|create|revoke personal access tokens\nprojects list|get|create|delete\nservices list|get|create|update|delete|start|stop|restart|redeploy\nservices clone|move duplicate config or re-parent a service\nservices domains list|add|remove|default|check multi-domain + DNS preflight\nregistries list|add|update|remove private-image pull credentials\nvolumes list|delete docker volume inventory + cleanup (admin)\ndeploy <service-id> trigger a deployment (--no-cache, --commit-hash, --branch)\ndeployments list|get|logs|rollback|cancel\nlogs <service-id> [--follow] runtime logs (SSE follow)\nexec <service-id> -- <cmd> one-off container command\nvariables list|set|unset env vars (secrets masked; --redeploy applies them)\nvariables shared list|set|unset project shared vars — referenced as ${{shared.KEY}}\ndatabases list|get|create|update|delete|action|backup|restore|download-backup|import-backup\n import-backup <id> <file.tar.gz|-> uploads an archive as a restore\n point; ships offsite too when a backup target is assigned\ndatabases list|get|create|update|delete|action|backup|restore|download-backup\n create --public exposes the port on all interfaces;\n update <id> --public/--public=false toggles it (container recreates, data persists)\ndatabases register <name> register an external database (--type --host --port --database --username --password --ssl; probed before storing)\ndatabases test-connection probe a connection — by id, or ad-hoc flags\nbackup-targets list|add|update|remove|test\n S3-compatible archive destinations; add/update probe the bucket\n first, empty --access-key/--secret-key on update keep stored keys;\n update <db-id> --backup-target assigns a target (\"\" clears it)\ncron list|get|create|delete|trigger|executions\ntemplates list|get|create|delete|deploy|plan|import-compose|deploy-graph\n deploy --name --var K=V; plan dry-resolves expressions;\n import-compose <compose.yml|-> converts a compose file into\n a v2 graph config; deploy-graph <project-id> <config.json|->\n deploys an ad-hoc service graph\nimport compose <project-id> <compose.yml|->\n deploys a docker-compose file straight into a project —\n no template row. DB images become managed databases\n attached under their service name; --var K=V overrides\n template variables\nnodes list|get|delete|tokens node agents + onboarding tokens (admin)\nnodes add <user@host> SSH bootstrap — one session installs the agent, then never needed again (--print for manual runs)\nnodes update|prune|commands rename/auto-prune, bounded docker prune, command history\nmigrate railway pull a Railway project (services, vars, domains) via GraphQL and recreate it — --dry-run prints the plan\nmigrate dflow import a dflow project (services, vars, volumes, domains) via its Payload REST API — needs --url + --api-key, --dry-run prints the plan\nscaling status|policies|services|scale\nha status|alerts|health|policies|enable|disable|failover\nsecurity scan|vulnerabilities|metrics\ngateway services|keys|anLine truncated
}, },
{ {
"path": "guides/CLOUDFLARE_SETUP.md", "path": "guides/CLOUDFLARE_SETUP.md",
+8
View File
@@ -86,6 +86,7 @@ export type ServiceEntity = {
sleepIdleMinutes?: number; sleepIdleMinutes?: number;
nodeId?: string; nodeId?: string;
nodeName?: string; nodeName?: string;
spread?: boolean;
publicUrl?: string; publicUrl?: string;
}; };
@@ -696,6 +697,7 @@ function normalizeService(service: RawService): ServiceEntity | null {
sleepIdleMinutes: service.sleep_idle_minutes, sleepIdleMinutes: service.sleep_idle_minutes,
nodeId: service.node_id, nodeId: service.node_id,
nodeName: service.node_name, nodeName: service.node_name,
spread: service.spread,
publicUrl: service.public_url, publicUrl: service.public_url,
}; };
} }
@@ -1162,6 +1164,12 @@ export async function drainAgent(id: string): Promise<DrainResult> {
return requestJson<DrainResult>(`/agents/${id}/drain`, { method: 'POST' }); return requestJson<DrainResult>(`/agents/${id}/drain`, { method: 'POST' });
} }
export type UpgradeResult = { platform: string; version: string };
export async function upgradeAgent(id: string): Promise<UpgradeResult> {
return requestJson<UpgradeResult>(`/agents/${id}/upgrade`, { method: 'POST' });
}
export async function getAgent(id: string): Promise<NodeAgentEntity | null> { export async function getAgent(id: string): Promise<NodeAgentEntity | null> {
const payload = await requestJson<{ agent?: RawNodeAgent }>(`/agents/${id}`); const payload = await requestJson<{ agent?: RawNodeAgent }>(`/agents/${id}`);
return payload.agent ? normalizeAgent(payload.agent) : null; return payload.agent ? normalizeAgent(payload.agent) : null;
+49
View File
@@ -5503,6 +5503,46 @@ paths:
schema: schema:
$ref: '#/components/schemas/Error' $ref: '#/components/schemas/Error'
/agents/{id}/upgrade:
post:
tags:
- Agents
summary: Upgrade the node agent
description: >-
Enqueues a `self_upgrade` command. The agent downloads its
replacement binary from this server's bundled `/agents/download`
endpoint, verifies the sha256, atomically swaps the executable, and
re-execs itself — the command result reports before the process
swaps. The node must be online, and the server must bundle a binary
for the node's architecture (`linux-amd64`/`linux-arm64`).
security:
- bearerAuth: []
parameters:
- $ref: '#/components/parameters/AgentIdParam'
responses:
'202':
description: Upgrade command enqueued
content:
application/json:
schema:
type: object
properties:
command: { type: object }
platform: { type: string }
version: { type: string }
'404':
description: Agent not found
content:
application/json:
schema:
$ref: '#/components/schemas/Error'
'409':
description: Node offline or no bundled binary for its architecture
content:
application/json:
schema:
$ref: '#/components/schemas/Error'
/agents/{id}/metrics: /agents/{id}/metrics:
get: get:
tags: tags:
@@ -6974,6 +7014,9 @@ components:
node_name: node_name:
type: string type: string
description: Display name of the pinned node agent (read-only) description: Display name of the pinned node agent (read-only)
spread:
type: boolean
description: Replicas are distributed across every online, schedulable node agent (replica i → agents[i mod n]). Mutually exclusive with an explicit node_id pin.
public_url: public_url:
type: string type: string
description: Computed public URL (published port or domain) description: Computed public URL (published port or domain)
@@ -7068,6 +7111,9 @@ components:
node_id: node_id:
type: string type: string
description: Node agent id to pin to, 'auto' = least-loaded online agent; omit/empty = local host description: Node agent id to pin to, 'auto' = least-loaded online agent; omit/empty = local host
spread:
type: boolean
description: Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin
UpdateServiceRequest: UpdateServiceRequest:
type: object type: object
@@ -7157,6 +7203,9 @@ components:
node_id: node_id:
type: string type: string
description: Node agent id to pin to, 'auto' = least-loaded online agent, 'local' = run on the Containr host (clears the pin) description: Node agent id to pin to, 'auto' = least-loaded online agent, 'local' = run on the Containr host (clears the pin)
spread:
type: boolean
description: Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin — send 'local' first to clear a pin
ServiceDomain: ServiceDomain:
type: object type: object
+16
View File
@@ -143,6 +143,15 @@ containr services update <id> --node auto # least-loaded online node
containr services update <id> --node local # move back to this server containr services update <id> --node local # move back to this server
``` ```
`auto` scores nodes by real headroom — lowest memory utilisation, then
cpu, then container count. Or spread replicas across every online,
schedulable node (mutually exclusive with a pin):
```bash
containr services update <id> --replicas 4 --spread # replica i → node[i % n]
containr services update <id> --spread=false # back to single placement
```
Remote nodes run registry-pulled images only — git builds, domains Remote nodes run registry-pulled images only — git builds, domains
(Traefik lives on the host), and sleep detection stay local. Start, (Traefik lives on the host), and sleep detection stay local. Start,
stop, restart, redeploy, and delete all fan out to the pinned node's stop, restart, redeploy, and delete all fan out to the pinned node's
@@ -154,8 +163,15 @@ Take a node out of rotation or empty it entirely (admin):
containr nodes cordon <id> # no new pins; running replicas stay containr nodes cordon <id> # no new pins; running replicas stay
containr nodes uncordon <id> containr nodes uncordon <id>
containr nodes drain <id> # cordon + remove its containers + unpin services containr nodes drain <id> # cordon + remove its containers + unpin services
containr nodes upgrade <id> # agent pulls its replacement from the server, re-execs
``` ```
`nodes upgrade` swaps the agent binary for the one bundled in the running
server image — the agent downloads it over the API, verifies the sha256 the
server computed, sanity-checks it, atomically replaces its own executable,
and re-execs. The node must be online and the server must ship a binary for
its architecture (`linux-amd64`/`linux-arm64`).
## Service domains and access ## Service domains and access
A service can carry multiple hostnames; the default is mirrored to A service can carry multiple hostnames; the default is mirrored to