From bc93c83cf9fac5f1d8eca0b2ee0d1e4c1a89385a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1=C5=A1=20Dvo=C5=99=C3=A1k?= <150935816+Dvorinka@users.noreply.github.com> Date: Fri, 9 Oct 2026 16:19:12 +0200 Subject: [PATCH] feat(scheduling): multi-node service placement + remote lifecycle (#54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(scheduling): multi-node service placement + remote container lifecycle services.node_id pins a service to a node agent: explicit id, "auto" (least-loaded online node), or "local"/empty for the local Docker host. - DeploymentEngine gains a NodeRunner seam; ReconcileService routes pinned services to the agent's command queue (create_container with image/env/command/ports/volumes/labels/resource limits), tracked in container_instances with deterministic per-replica ids. - Stop/start/restart/delete fan out to the pinned node — no more leaked remote containers on service delete. - Agent create_container is idempotent (rm -f first) and honors restart policy, labels, memory/cpus, published+ephemeral ports, volumes, env. - GET /nodes/options exposes id/name/status to authed users; full agent management stays admin-only. - CLI services create/update --node, MCP node_id docs, OpenAPI node_id/node_name + /nodes/options, service-detail Placement section. Ceilings: remote nodes run registry-pulled images only — git builds, Traefik domains, and remote sleep detection stay local until build-on-node and per-node ingress land. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * feat(scheduling): node cordon/drain + schedulable gate - node_agents.schedulable column; "auto" placement and explicit pins skip cordoned nodes; already-pinned services keep running (reconcile is unaffected — cordon is not eviction). - POST /agents/:id/cordon|uncordon|drain (admin). Drain cordons, removes every service container via the agent queue, and clears the services' pins so they redeploy on the local host. Requires the node online — an offline node can't acknowledge removals. - CLI nodes cordon|uncordon|drain (drain confirm-gated), MCP containr_nodes_cordon|uncordon|drain, OpenAPI paths + schedulable field, node-detail page cordon badge + drain/cordon controls. - Fix: re-registering an agent no longer resets auto_prune. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * feat(scheduling): remote registry pulls + post-run health check - Dispatch payload carries the project owner's registry credentials for the image host; the agent docker-logins for the pull and logs out after, so private images deploy to remote nodes. - Completed command payloads get their registry password scrubbed — secrets don't linger in agent_commands rows. - Agent verifies the container is still running ~1.5s after docker run; crash-loops now fail the deployment with the container's last log lines instead of reporting "running" optimistically. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- ROADMAP2.md | 31 +- app/backend/cmd/agent/main.go | 101 +++++- app/backend/internal/api/agents.go | 161 +++++++++- app/backend/internal/api/deployments.go | 15 + app/backend/internal/api/noderunner.go | 304 ++++++++++++++++++ app/backend/internal/api/routes.go | 4 + app/backend/internal/api/runtime.go | 43 ++- app/backend/internal/api/services.go | 74 ++++- app/backend/internal/api/templategraph.go | 2 +- app/backend/internal/cli/commands/nodes.go | 59 +++- app/backend/internal/cli/commands/services.go | 12 +- .../internal/database/sqlcdb/agents.sql.go | 140 +++++++- .../internal/database/sqlcdb/models.go | 2 + .../internal/database/sqlcdb/querier.go | 6 + app/backend/internal/deployment/engine.go | 24 ++ app/backend/internal/deployment/runtime.go | 30 +- app/backend/internal/mcpserver/server.go | 7 +- .../20261108000000_service_node_pin.sql | 19 ++ .../20261109000000_node_schedulable.sql | 7 + app/backend/sqlc/queries/agents.sql | 35 ++ app/backend/sqlc/schema.sql | 6 +- .../src/features/nodes/NodeDetailPage.tsx | 47 +++ .../service/pages/ServiceDetailPage.tsx | 61 ++++ app/frontend/src/generated/api-types.ts | 228 +++++++++++++ app/frontend/src/generated/docs-snapshot.ts | 2 +- app/frontend/src/lib/api-client.ts | 27 ++ docs/api/openapi.yaml | 142 ++++++++ docs/guides/CLI.md | 24 ++ 28 files changed, 1565 insertions(+), 48 deletions(-) create mode 100644 app/backend/internal/api/noderunner.go create mode 100644 app/backend/migrations_goose/20261108000000_service_node_pin.sql create mode 100644 app/backend/migrations_goose/20261109000000_node_schedulable.sql diff --git a/ROADMAP2.md b/ROADMAP2.md index 4b28200..788f078 100644 --- a/ROADMAP2.md +++ b/ROADMAP2.md @@ -277,12 +277,31 @@ and wired refs, one click or one CLI call. `nodes commands`, MCP `containr_nodes_update|prune|commands` (confirm-gated), Usage-page per-node Prune button + auto toggle. `cmd/agent/main.go`, `notification_producers.go:scheduleAutoPrune`. -- [ ] **Multi-node scheduling** — wire `scheduling_rules` (table exists): - pin/spread/affinity, prefer-lowest-usage; deploy engine routes to - `services.node_id` via agent commands (agent already runs containers). -- [ ] **Build-on-node** — agents currently `docker run` prebuilt images - only; extend agent to pull registry images (build stays central) — - private registry creds flow through Phase C. +- [🟡] **Multi-node scheduling** — phase 1 shipped: `services.node_id` + pin (explicit / `auto` = least-loaded online / `local` = clear), + `NodeRunner` seam in the deploy engine routes reconcile to the agent's + command queue (`create_container` with image/env/cmd/ports/volumes/ + labels/limits), `container_instances` inventory per (agent, service, + replica), remote stop/start/restart/delete fan-out, `GET + /nodes/options`, `--node` on services create/update + MCP + OpenAPI + + service-detail Placement section. Drain/cordon shipped: + `node_agents.schedulable`, `POST /agents/:id/cordon|uncordon|drain` + (drain evicts remote containers via the agent queue + unpins services + so they redeploy locally), `nodes cordon|uncordon|drain` CLI + + MCP (drain confirm-gated) + node-detail page controls; `auto` and + explicit pins skip cordoned nodes. Remote nodes run registry-pulled + images only — git builds, Traefik domains, and remote sleep are + documented ceilings. Remaining: `scheduling_rules` spread/affinity + policies, deployment status reported back from agent + results, per-node resource-aware `auto` (currently count-based). +- [🟡] **Build-on-node** — private registry pulls shipped: the dispatch + payload carries the project owner's `registries` credentials for the + image host; the agent `docker login`s for the pull and logs out after, + and command payloads are scrubbed of passwords on completion. Agent + also verifies the container is still running ~1.5s after `run` so + crash-loops fail the deploy instead of reporting healthy. Remaining: + real builds on the node (source checkout + build daemon) — pulls + already work. - [ ] **IaC provisioning (ROADMAP §2.2, unchanged)** — Terraform/ OpenTofu, `infra_connections`, Proxmox + AWS modules, + Hetzner module (dflow's second provider), cloud-init → auto-enroll. diff --git a/app/backend/cmd/agent/main.go b/app/backend/cmd/agent/main.go index 1dd53e2..9fc79a8 100644 --- a/app/backend/cmd/agent/main.go +++ b/app/backend/cmd/agent/main.go @@ -283,7 +283,38 @@ func createContainer(payload map[string]interface{}) (string, error) { return "", errors.New("container name and image are required") } - args := []string{"run", "-d", "--name", name, "--restart", "unless-stopped"} + // Private registry: log in for the pull, then log back out so the + // credential doesn't linger in the node's docker config. + if reg, ok := container["registry"].(map[string]interface{}); ok { + if user := stringValue(reg["username"]); user != "" { + server := firstString(stringValue(reg["server"]), registryHostOf(image)) + login := exec.Command("docker", "login", server, "-u", user, "--password-stdin") + login.Stdin = strings.NewReader(stringValue(reg["password"])) + if out, err := login.CombinedOutput(); err != nil { + return "", fmt.Errorf("registry login %s: %s", server, strings.TrimSpace(string(out))) + } + defer func() { _, _ = runDocker("logout", server) }() + } + } + + // Redeploys reuse the deterministic name — clear the stale container + // first so `run` never fails on a name conflict. + _, _ = runDocker("rm", "-f", name) + + args := []string{"run", "-d", "--name", name} + restart := firstString(stringValue(container["restart_policy"]), "unless-stopped") + args = append(args, "--restart", restart) + if labels, ok := container["labels"].(map[string]interface{}); ok { + for key, value := range labels { + args = append(args, "--label", fmt.Sprintf("%s=%v", key, value)) + } + } + if mem := int64Value(container["memory"]); mem > 0 { + args = append(args, "--memory", fmt.Sprintf("%d", mem)) + } + if cpus := floatValue(container["cpus"]); cpus > 0 { + args = append(args, "--cpus", fmt.Sprintf("%g", cpus)) + } if env, ok := container["environment"].(map[string]interface{}); ok { for key, value := range env { args = append(args, "-e", fmt.Sprintf("%s=%v", key, value)) @@ -298,8 +329,14 @@ func createContainer(payload map[string]interface{}) (string, error) { hostPort := intValue(port["host_port"]) containerPort := intValue(port["container_port"]) protocol := firstString(stringValue(port["protocol"]), "tcp") - if hostPort > 0 && containerPort > 0 { + if containerPort <= 0 { + continue + } + if hostPort > 0 { args = append(args, "-p", fmt.Sprintf("%d:%d/%s", hostPort, containerPort, protocol)) + } else { + // Ephemeral host port — docker picks the binding. + args = append(args, "-p", fmt.Sprintf("%d/%s", containerPort, protocol)) } } } @@ -321,7 +358,33 @@ func createContainer(payload map[string]interface{}) (string, error) { } } args = append(args, image) - return runDocker(args...) + if cmd, ok := container["command"].([]interface{}); ok { + for _, arg := range cmd { + args = append(args, stringValue(arg)) + } + } + if _, err := runDocker(args...); err != nil { + return "", err + } + // docker run -d returns before the entrypoint fails — a crash-looping + // service would be reported healthy. Give it a moment, then verify. + time.Sleep(1500 * time.Millisecond) + state, _ := runDocker("inspect", "-f", "{{.State.Status}}", name) + if state != "running" { + logs, _ := runDocker("logs", "--tail", "5", name) + return "", fmt.Errorf("container %s is %q after start: %s", name, state, logs) + } + return name, nil +} + +// registryHostOf mirrors the server's registryHost: first path segment with +// a "." or ":" is a registry host; bare names mean Docker Hub. +func registryHostOf(imageRef string) string { + parts := strings.Split(imageRef, "/") + if len(parts) > 1 && (strings.Contains(parts[0], ".") || strings.Contains(parts[0], ":")) { + return parts[0] + } + return "docker.io" } func commandTarget(payload map[string]interface{}) string { @@ -620,3 +683,35 @@ func boolValue(value interface{}) bool { typed, _ := value.(bool) return typed } + +func int64Value(value interface{}) int64 { + switch typed := value.(type) { + case int64: + return typed + case int: + return int64(typed) + case float64: + return int64(typed) + case json.Number: + out, _ := typed.Int64() + return out + default: + return 0 + } +} + +func floatValue(value interface{}) float64 { + switch typed := value.(type) { + case float64: + return typed + case int: + return float64(typed) + case int64: + return float64(typed) + case json.Number: + out, _ := typed.Float64() + return out + default: + return 0 + } +} diff --git a/app/backend/internal/api/agents.go b/app/backend/internal/api/agents.go index 25e994f..b3bfdfb 100644 --- a/app/backend/internal/api/agents.go +++ b/app/backend/internal/api/agents.go @@ -11,6 +11,7 @@ import ( "encoding/json" "errors" "fmt" + "log" "math" "net/http" "os" @@ -39,6 +40,7 @@ type NodeAgent struct { Resources NodeResources `json:"resources"` LastHeartbeat time.Time `json:"last_heartbeat"` AutoPrune bool `json:"auto_prune"` + Schedulable bool `json:"schedulable"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` Metadata map[string]interface{} `json:"metadata"` @@ -228,17 +230,38 @@ func unmarshalRaw(raw pqtype.NullRawMessage, dst interface{}) { } } +// ListNodeOptions returns the minimal id/name/status list used by the service +// placement picker — available to any authenticated user (no hostnames, IPs, +// or telemetry). +func (h *NodeAgentHandler) ListNodeOptions(c *gin.Context) { + rows, err := h.q.ListAgents(c.Request.Context()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list nodes"}) + return + } + options := make([]gin.H, 0, len(rows)) + for _, row := range rows { + options = append(options, gin.H{ + "id": row.ID, + "name": row.Name, + "status": row.Status.String, + }) + } + c.JSON(http.StatusOK, gin.H{"nodes": options}) +} + func agentFromRow(row sqlcdb.NodeAgent) NodeAgent { agent := NodeAgent{ - ID: row.ID, - Name: row.Name, - Hostname: row.Hostname, - IPAddress: row.IpAddress, - Port: int(row.Port), - Status: row.Status.String, - Version: row.Version.String, - AutoPrune: row.AutoPrune, - Metadata: map[string]interface{}{}, + ID: row.ID, + Name: row.Name, + Hostname: row.Hostname, + IPAddress: row.IpAddress, + Port: int(row.Port), + Status: row.Status.String, + Version: row.Version.String, + AutoPrune: row.AutoPrune, + Schedulable: row.Schedulable, + Metadata: map[string]interface{}{}, } if row.LastHeartbeat.Valid { agent.LastHeartbeat = row.LastHeartbeat.Time @@ -380,6 +403,8 @@ func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) { Resources: existing.Resources, LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true}, Metadata: rawJSON(metadata), + AutoPrune: existing.AutoPrune, + Schedulable: existing.Schedulable, }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent"}) @@ -511,6 +536,7 @@ func (h *NodeAgentHandler) UpdateAgent(c *gin.Context) { LastHeartbeat: sql.NullTime{Time: agent.LastHeartbeat, Valid: !agent.LastHeartbeat.IsZero()}, Metadata: rawJSON(agent.Metadata), AutoPrune: agent.AutoPrune, + Schedulable: agent.Schedulable, }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent"}) @@ -539,6 +565,7 @@ func applyAgentUpdates(agent *NodeAgent, updates map[string]interface{}) { remarshal("resources", &agent.Resources) remarshal("metadata", &agent.Metadata) remarshal("auto_prune", &agent.AutoPrune) + remarshal("schedulable", &agent.Schedulable) } // DeleteAgent removes an agent @@ -673,9 +700,40 @@ func (h *NodeAgentHandler) CompleteCommand(c *gin.Context) { h.updateContainerStatusAfterCommand(ctx, out) } + // Command payloads can carry registry credentials — scrub them once the + // command is settled so secrets don't linger in the queue rows. + scrubCommandSecrets(ctx, h.q, commandID, command.Payload) + c.JSON(http.StatusOK, gin.H{"command": out}) } +// scrubCommandSecrets rewrites a completed command's payload with any +// registry password blanked. Best-effort — a failed scrub leaves the row. +func scrubCommandSecrets(ctx context.Context, q *sqlcdb.Queries, commandID string, payload pqtype.NullRawMessage) { + if !payload.Valid || len(payload.RawMessage) == 0 { + return + } + var doc map[string]interface{} + if err := json.Unmarshal(payload.RawMessage, &doc); err != nil { + return + } + container, ok := doc["container"].(map[string]interface{}) + if !ok { + return + } + reg, ok := container["registry"].(map[string]interface{}) + if !ok || reg["password"] == nil { + return + } + reg["password"] = "***" + if err := q.ScrubCommandPayload(ctx, sqlcdb.ScrubCommandPayloadParams{ + ID: commandID, + Payload: rawJSON(doc), + }); err != nil { + log.Printf("agents: scrub command %s payload: %v", commandID, err) + } +} + func (h *NodeAgentHandler) updateContainerStatusAfterCommand(ctx context.Context, command AgentCommand) { row, err := h.q.GetContainer(ctx, *command.ContainerID) if err != nil { @@ -1329,9 +1387,94 @@ func (h *NodeAgentHandler) SetupAdminRoutes(router *gin.RouterGroup) { agents.DELETE("/:id/containers/:containerId", h.ContainerAction) agents.GET("/:id/metrics", h.GetAgentMetrics) + agents.POST("/:id/cordon", h.CordonAgent) + agents.POST("/:id/uncordon", h.UncordonAgent) + agents.POST("/:id/drain", h.DrainAgent) agents.POST("/:id/prune", h.PruneAgent) agents.POST("/:id/commands", h.ExecuteCommand) agents.GET("/:id/commands", h.GetAgentCommands) agents.GET("/:id/commands/:commandId", h.GetCommandStatus) } } + +// setSchedulable flips the node's scheduling gate. Cordoned nodes keep +// running their placed replicas but take no new pins or `auto` placements. +func (h *NodeAgentHandler) setSchedulable(c *gin.Context, schedulable bool) { + id := c.Param("id") + if _, err := h.q.GetAgent(c.Request.Context(), id); err != nil { + respondError(c, http.StatusNotFound, "NOT_FOUND", "Agent not found") + return + } + if err := h.q.SetAgentSchedulable(c.Request.Context(), sqlcdb.SetAgentSchedulableParams{ + ID: id, Schedulable: schedulable, + }); err != nil { + respondError(c, http.StatusInternalServerError, "INTERNAL", "failed to update node") + return + } + LogAuditWithRequest(c, "node", id, map[bool]string{true: "uncordon", false: "cordon"}[schedulable], nil) + c.JSON(http.StatusOK, gin.H{"id": id, "schedulable": schedulable}) +} + +// CordonAgent blocks new placements on the node (explicit pins and `auto`). +func (h *NodeAgentHandler) CordonAgent(c *gin.Context) { h.setSchedulable(c, false) } + +// UncordonAgent re-opens the node for placement. +func (h *NodeAgentHandler) UncordonAgent(c *gin.Context) { h.setSchedulable(c, true) } + +// DrainAgent cordons the node, removes every service container via the agent +// command queue, and clears the affected services' pins — they fall back to +// the local host on next deploy and can be re-pinned to another node. +// Requires the node to be online; an offline node can't acknowledge the +// removals. +func (h *NodeAgentHandler) DrainAgent(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 not online — reconnect it, or remove the containers manually") + return + } + if err := h.q.SetAgentSchedulable(ctx, sqlcdb.SetAgentSchedulableParams{ + ID: id, Schedulable: false, + }); err != nil { + respondError(c, http.StatusInternalServerError, "INTERNAL", "failed to cordon node") + return + } + + runner := &agentNodeRunner{q: h.q} + rows, err := h.q.ListContainersForAgent(ctx, id) + if err != nil { + respondError(c, http.StatusInternalServerError, "INTERNAL", "failed to list node containers") + return + } + services := map[string]bool{} + for _, row := range rows { + services[row.ServiceID] = true + } + var drained int + var firstErr error + for serviceID := range services { + if err := runner.RemoveService(ctx, serviceID, id); err != nil { + if firstErr == nil { + firstErr = err + } + continue + } + drained += 1 + } + unpinned, _ := h.q.ClearServiceNodePins(ctx, sql.NullString{String: id, Valid: true}) + if firstErr != nil { + respondError(c, http.StatusBadGateway, "DRAIN_PARTIAL", + fmt.Sprintf("drained %d service(s), unpinned %d — one failed: %v", drained, unpinned, firstErr)) + return + } + LogAuditWithRequest(c, "node", id, "drain", map[string]interface{}{ + "services": drained, "unpinned": unpinned, + }) + c.JSON(http.StatusOK, gin.H{"id": id, "schedulable": false, "services_drained": drained, "services_unpinned": unpinned}) +} diff --git a/app/backend/internal/api/deployments.go b/app/backend/internal/api/deployments.go index d0e3ce0..2a8ac91 100644 --- a/app/backend/internal/api/deployments.go +++ b/app/backend/internal/api/deployments.go @@ -421,6 +421,20 @@ func runDeploymentAndSyncWithImage( maintURL = maintenanceURL(db) } + // Node-pinned services dispatch to the agent's command queue. Builds run + // 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 != "") { + 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)" + _, _ = db.Exec( + `UPDATE deployments SET status = 'failed', error = $1, completed_at = $2, updated_at = $2 WHERE id = $3`, + msg, failedAt, dbDeployment.ID, + ) + return + } + deployReq := &deployment.DeploymentRequest{ ProjectID: service.ProjectID.String(), ServiceID: service.ID.String(), @@ -447,6 +461,7 @@ func runDeploymentAndSyncWithImage( Maintenance: maintenanceMode, MaintenanceURL: maintURL, BasicAuthUsers: basicAuthUsers, + NodeID: nodeID, }, Trigger: deployment.TriggerConfig{ Type: req.Trigger, diff --git a/app/backend/internal/api/noderunner.go b/app/backend/internal/api/noderunner.go new file mode 100644 index 0000000..6258619 --- /dev/null +++ b/app/backend/internal/api/noderunner.go @@ -0,0 +1,304 @@ +package api + +import ( + "context" + "crypto/sha256" + "database/sql" + "fmt" + "strconv" + "strings" + "time" + + "github.com/google/uuid" + + "containr/internal/database" + "containr/internal/database/sqlcdb" + "containr/internal/deployment" +) + +// agentNodeRunner implements deployment.NodeRunner by dispatching container +// lifecycle commands to a node agent's command queue and waiting for the +// reported result. Remote nodes have no Traefik — domains stay local-only +// until per-node ingress lands. +type agentNodeRunner struct { + q *sqlcdb.Queries + db *database.DB +} + +func newAgentNodeRunner(db *database.DB) *agentNodeRunner { + return &agentNodeRunner{q: sqlcdb.New(db.DB), db: db} +} + +// remoteContainerID is deterministic per (agent, service, replica) so redeploys +// update the same inventory row instead of accumulating stale ones. +func remoteContainerID(agentID, serviceID string, replica int) string { + sum := sha256.Sum256([]byte(agentID + "|" + serviceID + "|" + fmt.Sprint(replica))) + return fmt.Sprintf("%x-%x-%x-%x-%x", sum[0:4], sum[4:6], sum[6:8], sum[8:10], sum[10:16]) +} + +func remoteContainerName(serviceID string, replica int) string { + return fmt.Sprintf("containr-%s-%d", serviceID, replica) +} + +func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.RuntimeSpec) (*deployment.RuntimeState, error) { + agent, err := r.q.GetAgent(ctx, spec.NodeID) + if err != nil { + return nil, fmt.Errorf("node %s not found", spec.NodeID) + } + 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) + } + 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") + } + + replicas := spec.Replicas + if replicas < 1 { + 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) + for _, row := range existing { + if row.ServiceID != spec.ServiceID || !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"}), + }) + } + } + + state := &deployment.RuntimeState{Desired: replicas, Status: "running"} + for i := 0; i < replicas; 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, + }, + } + + _ = 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 +// host so private images pull on remote nodes. Empty map → anonymous pull. +func (r *agentNodeRunner) pullCredentials(projectID, imageRef string) map[string]interface{} { + var ownerID string + if err := r.db.QueryRow( + `SELECT owner_id FROM projects WHERE id = $1`, projectID, + ).Scan(&ownerID); err != nil { + return map[string]interface{}{} + } + auth := registryAuthFor(r.db, ownerID, imageRef) + if auth.Username == "" { + return map[string]interface{}{} + } + return map[string]interface{}{ + "server": auth.ServerAddress, + "username": auth.Username, + "password": auth.Password, + } +} + +// ControlService fans a lifecycle action out to every replica row the +// service has on the agent. Actions map onto the agent's container commands; +// "remove" also tombstones the inventory row. +func (r *agentNodeRunner) ControlService(ctx context.Context, serviceID, agentID, action string) error { + cmdType := map[string]string{ + "start": "start_container", + "stop": "stop_container", + "restart": "restart_container", + "remove": "remove_container", + }[action] + if cmdType == "" { + return fmt.Errorf("unsupported action %q", action) + } + rows, err := r.q.ListContainersForAgent(ctx, agentID) + if err != nil { + return err + } + removed := map[string]interface{}{"state": "removed"} + for _, row := range rows { + if row.ServiceID != serviceID { + continue + } + if err := r.enqueueAndWait(ctx, agentID, row.ID, cmdType, + map[string]interface{}{"container_name": row.Name}); err != nil { + return fmt.Errorf("%s %s on node: %w", action, row.Name, err) + } + if action == "remove" { + _ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{ + ID: row.ID, Status: rawJSON(removed)}) + } + } + return nil +} + +// RemoveService tears down every replica of the service on the agent and +// clears the inventory rows. +func (r *agentNodeRunner) RemoveService(ctx context.Context, serviceID, agentID string) error { + if err := r.ControlService(ctx, serviceID, agentID, "remove"); err != nil { + return err + } + return r.q.DeleteServiceContainersOnAgent(ctx, sqlcdb.DeleteServiceContainersOnAgentParams{ + NodeAgentID: agentID, + ServiceID: serviceID, + }) +} + +// enqueueAndWait pushes a command to the agent's queue and polls for the +// reported result. Agents poll on their own interval (≥5s), so this blocks — +// it runs inside the deploy queue worker, never on a request path. +func (r *agentNodeRunner) enqueueAndWait(ctx context.Context, agentID, containerID, cmdType string, payload map[string]interface{}) error { + var cid sql.NullString + if containerID != "" { + cid = sql.NullString{String: containerID, Valid: true} + } + cmd, err := r.q.CreateCommand(ctx, sqlcdb.CreateCommandParams{ + ID: uuid.New().String(), + Type: cmdType, + NodeAgentID: agentID, + ContainerID: cid, + Payload: rawJSON(payload), + }) + if err != nil { + return err + } + + deadline := time.Now().Add(8 * time.Minute) // image pulls on a cold node + for { + row, err := r.q.GetCommandForAgent(ctx, sqlcdb.GetCommandForAgentParams{ + ID: cmd.ID, + NodeAgentID: agentID, + }) + if err == nil { + switch row.Status.String { + case "completed": + return nil + case "failed": + if row.Error.Valid && row.Error.String != "" { + return fmt.Errorf("agent: %s", row.Error.String) + } + return fmt.Errorf("agent reported failure") + } + } + if time.Now().After(deadline) { + return fmt.Errorf("command %s timed out waiting for the agent", cmdType) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(2 * time.Second): + } + } +} + +// resolveNodePin validates or resolves a node_id request value. "auto" picks +// the least-loaded online agent; "" / "local" clears the pin. +func resolveNodePin(ctx context.Context, db *database.DB, requested string) (string, error) { + requested = strings.TrimSpace(requested) + if requested == "" || requested == "local" { + return "", nil + } + q := sqlcdb.New(db.DB) + if requested == "auto" { + id, err := q.PickLeastLoadedAgent(ctx) + if err != nil { + return "", fmt.Errorf("no online nodes available") + } + return id, nil + } + agent, err := q.GetAgent(ctx, requested) + if err != nil { + return "", fmt.Errorf("node not found") + } + if !agent.Schedulable { + return "", fmt.Errorf("node %s is cordoned — uncordon it before pinning", agent.Name) + } + return requested, nil +} diff --git a/app/backend/internal/api/routes.go b/app/backend/internal/api/routes.go index f0b0236..f7ff060 100644 --- a/app/backend/internal/api/routes.go +++ b/app/backend/internal/api/routes.go @@ -43,6 +43,9 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg dockerClient = client buildManager = build.NewBuildManager("/tmp/containr-builds", dockerClient) deploymentEngine = deployment.NewDeploymentEngine(buildManager, dockerClient) + // Node-pinned services dispatch container lifecycle to the agent's + // command queue instead of the local Docker socket. + deploymentEngine.SetNodeRunner(newAgentNodeRunner(db)) // Restore managed sidecars (e.g. cloudflared) from saved settings. syncManagedContainers(dockerClient, db) } @@ -355,6 +358,7 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg // Cross-cutting ops view — active/queued/failed work authed.GET("/operations", handleGetOperations) + authed.GET("/nodes/options", agentHandler.ListNodeOptions) authed.GET("/activity", handleGetActivity) authed.GET("/projects/:id/activity", handleGetProjectActivity) diff --git a/app/backend/internal/api/runtime.go b/app/backend/internal/api/runtime.go index 9749135..c311678 100644 --- a/app/backend/internal/api/runtime.go +++ b/app/backend/internal/api/runtime.go @@ -18,6 +18,38 @@ 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. +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) + } + switch action { + case "start": + return engine.StartService(ctx, serviceID.String()) + case "stop": + return engine.StopService(ctx, serviceID.String()) + case "restart": + if err := engine.StopService(ctx, serviceID.String()); err != nil { + return err + } + return engine.StartService(ctx, serviceID.String()) + } + return fmt.Errorf("unsupported lifecycle action %q", action) +} + +// removeServiceRuntime tears down replicas wherever they run — remote node +// when pinned, local Docker otherwise. +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 + } + _ = engine.RemoveServiceContainers(ctx, serviceID.String()) +} + // serviceRuntimeSpec builds the container spec for a service: stored env vars // merged with ${{Service.KEY}} references resolved across the project. func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpec, error) { @@ -37,6 +69,7 @@ func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpe Domain: service.Domain, HealthPath: service.HealthCheckPath, RestartPolicy: service.RestartPolicy, + NodeID: serviceNodeID(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 @@ -384,7 +417,7 @@ func handleServiceStart(c *gin.Context) { if !ok { return } - if err := engine.StartService(c.Request.Context(), service.ID.String()); err != nil { + if err := serviceLifecycle(c.Request.Context(), db, engine, service.ID, "start"); err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } @@ -402,7 +435,7 @@ func handleServiceStop(c *gin.Context) { if !ok { return } - if err := engine.StopService(c.Request.Context(), service.ID.String()); err != nil { + if err := serviceLifecycle(c.Request.Context(), db, engine, service.ID, "stop"); err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } @@ -421,11 +454,7 @@ func handleServiceRestart(c *gin.Context) { return } ctx := c.Request.Context() - if err := engine.StopService(ctx, service.ID.String()); err != nil { - c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) - return - } - if err := engine.StartService(ctx, service.ID.String()); err != nil { + if err := serviceLifecycle(ctx, db, engine, service.ID, "restart"); err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) return } diff --git a/app/backend/internal/api/services.go b/app/backend/internal/api/services.go index 482685b..6c90239 100644 --- a/app/backend/internal/api/services.go +++ b/app/backend/internal/api/services.go @@ -5,6 +5,7 @@ import ( "containr/internal/deployment" "containr/internal/deployqueue" "context" + "database/sql" "encoding/json" "fmt" "net/http" @@ -48,8 +49,12 @@ type Service struct { StaticDir string `json:"static_dir,omitempty" db:"static_dir"` SleepEnabled bool `json:"sleep_enabled" db:"-"` SleepIdleMinutes int `json:"sleep_idle_minutes" db:"-"` - CreatedAt time.Time `json:"created_at" db:"created_at"` - UpdatedAt time.Time `json:"updated_at" db:"updated_at"` + // 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:"-"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` } // loadServiceSleep fills the sleep-mode columns, which live outside the @@ -64,6 +69,28 @@ func loadServiceSleep(db *database.DB, s *Service) { } } +// serviceNodeID reads the pin column directly — kept out of the wide service +// SELECTs so pre-migration schemas keep working. +func serviceNodeID(db *database.DB, serviceID uuid.UUID) string { + var id sql.NullString + if err := db.QueryRow(`SELECT node_id FROM services WHERE id = $1`, serviceID).Scan(&id); err == nil { + return id.String + } + return "" +} + +// loadServiceNode fills the pin + the agent's display name for responses. +func loadServiceNode(db *database.DB, s *Service) { + var name sql.NullString + s.NodeID = serviceNodeID(db, s.ID) + if s.NodeID == "" { + return + } + if err := db.QueryRow(`SELECT name FROM node_agents WHERE id = $1`, s.NodeID).Scan(&name); err == nil { + s.NodeName = name.String + } +} + // ServiceVolume is a volume/bind mount applied to every replica. type ServiceVolume struct { Type string `json:"type"` // volume | bind (empty = volume) @@ -141,6 +168,9 @@ type CreateServiceRequest struct { MemoryReserve string `json:"memory_reserve"` StaticBuildCmd string `json:"static_build_cmd"` StaticDir string `json:"static_dir"` + // 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"` } // UpdateServiceRequest represents a request to update a service @@ -170,6 +200,8 @@ type UpdateServiceRequest struct { StaticDir *string `json:"static_dir"` SleepEnabled *bool `json:"sleep_enabled"` SleepIdleMinutes *int `json:"sleep_idle_minutes"` + // NodeID pins placement; pointer so "local"/"" explicitly clears the pin. + NodeID *string `json:"node_id"` } // handleGetServices retrieves all services for a project @@ -253,6 +285,7 @@ func handleGetServices(c *gin.Context) { // Reconcile stored status with live container state (no-op without Docker). for i := range services { liveServiceStatus(c, db.(*database.DB), &services[i]) + loadServiceNode(db.(*database.DB), &services[i]) } c.JSON(http.StatusOK, gin.H{"services": services}) @@ -409,6 +442,15 @@ func handleCreateService(c *gin.Context) { sourceType := inferServiceSourceType(service) + if req.NodeID != "" { + 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 + } + // Insert service into database _, err = db.(*database.DB).Exec( `INSERT INTO services @@ -417,16 +459,18 @@ 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, - created_at, updated_at) + node_id, 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)`, + $27, $28, $29, $30, $31, $32, $33, $34)`, 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, service.Environment, service.GitRepo, service.GitBranch, service.BuildPath, service.CPU, service.Memory, service.Replicas, service.Port, service.Domain, service.HealthCheckPath, service.RestartPolicy, volumesJSON, service.Builder, service.CPUReserve, service.MemoryReserve, - service.StaticBuildCmd, service.StaticDir, service.CreatedAt, service.UpdatedAt, + service.StaticBuildCmd, service.StaticDir, + sql.NullString{String: service.NodeID, Valid: service.NodeID != ""}, + service.CreatedAt, service.UpdatedAt, ) if err != nil { @@ -456,7 +500,9 @@ func handleCreateService(c *gin.Context) { LogAuditWithRequest(c, "service", service.ID.String(), "create", map[string]interface{}{ "name": service.Name, "type": service.Type, "project_id": service.ProjectID, + "node_id": service.NodeID, }) + loadServiceNode(db.(*database.DB), &service) c.JSON(http.StatusCreated, gin.H{"service": service}) } @@ -547,6 +593,7 @@ func handleGetService(c *gin.Context) { } service.Domains = loadServiceDomains(db.(*database.DB), service.ID) loadServiceSleep(db.(*database.DB), &service) + loadServiceNode(db.(*database.DB), &service) maintenance, basicAuth := serviceAccess(db.(*database.DB), service.ID) service.MaintenanceMode = maintenance service.BasicAuth = basicAuthUsernames(basicAuth) @@ -725,6 +772,17 @@ func handleUpdateService(c *gin.Context) { volumesArg = volumesJSON } + if req.NodeID != nil { + 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 + } + existingService.NodeID = nodeID + } else { + existingService.NodeID = serviceNodeID(db.(*database.DB), serviceID) + } + existingService.UpdatedAt = time.Now() // Update service in database @@ -736,7 +794,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 + sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26 WHERE id = $18`, existingService.Name, existingService.Type, existingService.Image, existingService.Command, existingService.Environment, existingService.GitRepo, existingService.GitBranch, @@ -747,6 +805,7 @@ func handleUpdateService(c *gin.Context) { existingService.Builder, existingService.CPUReserve, existingService.MemoryReserve, existingService.StaticBuildCmd, existingService.StaticDir, existingService.SleepEnabled, existingService.SleepIdleMinutes, + sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""}, ) if err != nil { @@ -816,6 +875,7 @@ func handleUpdateService(c *gin.Context) { } } + loadServiceNode(db.(*database.DB), &existingService) c.JSON(http.StatusOK, gin.H{"service": existingService}) } @@ -865,7 +925,7 @@ func handleDeleteService(c *gin.Context) { // Remove runtime containers before dropping the row. if engineValue, exists := c.Get("deployment_engine"); exists && engineValue != nil { if engine, ok := engineValue.(*deployment.DeploymentEngine); ok { - _ = engine.RemoveServiceContainers(c.Request.Context(), serviceID.String()) + removeServiceRuntime(c.Request.Context(), db.(*database.DB), engine, serviceID) } } diff --git a/app/backend/internal/api/templategraph.go b/app/backend/internal/api/templategraph.go index 868c8e1..48fc57e 100644 --- a/app/backend/internal/api/templategraph.go +++ b/app/backend/internal/api/templategraph.go @@ -637,7 +637,7 @@ func deployTemplateGraph( cleanup := func() { for _, sid := range createdServiceIDs { if engine != nil { - _ = engine.RemoveServiceContainers(context.Background(), sid.String()) + removeServiceRuntime(context.Background(), db, engine, sid) } _, _ = db.Exec(`DELETE FROM environment_variables WHERE service_id = $1`, sid) _, _ = db.Exec(`DELETE FROM services WHERE id = $1`, sid) diff --git a/app/backend/internal/cli/commands/nodes.go b/app/backend/internal/cli/commands/nodes.go index 91b4479..108a3ae 100644 --- a/app/backend/internal/cli/commands/nodes.go +++ b/app/backend/internal/cli/commands/nodes.go @@ -133,6 +133,63 @@ var nodesPruneCmd = &cobra.Command{ }, } +var nodesCordonCmd = &cobra.Command{ + Use: "cordon ", + Short: "Stop new placements on a node (running replicas stay)", + 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]+"/cordon", nil) + if err != nil { + return err + } + PrintRaw(data) + return nil + }, +} + +var nodesUncordonCmd = &cobra.Command{ + Use: "uncordon ", + Short: "Re-open a node for placement", + 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]+"/uncordon", nil) + if err != nil { + return err + } + PrintRaw(data) + return nil + }, +} + +var nodesDrainCmd = &cobra.Command{ + Use: "drain ", + Short: "Cordon a node, remove its service containers, unpin affected services", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + c, err := client() + if err != nil { + return err + } + if !Confirm(fmt.Sprintf("Drain node %s? Its service containers are removed and services unpinned (they redeploy locally).", args[0])) { + return &APIError{Message: "aborted (pass --yes to skip confirmation)", ExitCode: ExitError} + } + data, err := c.Do("POST", "/agents/"+args[0]+"/drain", nil) + if err != nil { + return err + } + PrintRaw(data) + return nil + }, +} + var nodesCommandsCmd = &cobra.Command{ Use: "commands ", Short: "List recent commands sent to a node agent", @@ -220,5 +277,5 @@ func init() { 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)") nodeTokensCmd.AddCommand(nodeTokensIssueCmd, nodeTokensListCmd, nodeTokensRevokeCmd) - NodesCmd.AddCommand(nodesListCmd, nodesGetCmd, nodesUpdateCmd, nodesPruneCmd, nodesCommandsCmd, nodesDeleteCmd, nodeTokensCmd) + NodesCmd.AddCommand(nodesListCmd, nodesGetCmd, nodesUpdateCmd, nodesCordonCmd, nodesUncordonCmd, nodesDrainCmd, nodesPruneCmd, nodesCommandsCmd, nodesDeleteCmd, nodeTokensCmd) } diff --git a/app/backend/internal/cli/commands/services.go b/app/backend/internal/cli/commands/services.go index 8bcef75..8346f56 100644 --- a/app/backend/internal/cli/commands/services.go +++ b/app/backend/internal/cli/commands/services.go @@ -60,7 +60,7 @@ var servicesGetCmd = &cobra.Command{ return nil } s, _ := unwrapObject(data, "service") - for _, k := range []string{"id", "name", "type", "status", "environment", "image", "git_repo", "git_branch", "domain", "port"} { + for _, k := range []string{"id", "name", "type", "status", "environment", "image", "git_repo", "git_branch", "domain", "port", "node_name"} { fmt.Printf("%-14s %s\n", k+":", str(s, k)) } return nil @@ -79,7 +79,7 @@ var servicesCreateCmd = &cobra.Command{ body := map[string]interface{}{ "name": args[1], } - for _, f := range []string{"type", "image", "git-repo", "git-branch", "build-path", "command", "environment", "domain", "restart-policy", "healthcheck-path", "cpu", "memory", "builder", "cpu-reserve", "memory-reserve", "static-cmd", "static-dir"} { + for _, f := range []string{"type", "image", "git-repo", "git-branch", "build-path", "command", "environment", "domain", "restart-policy", "healthcheck-path", "cpu", "memory", "builder", "cpu-reserve", "memory-reserve", "static-cmd", "static-dir", "node"} { if v, _ := cmd.Flags().GetString(f); v != "" { key := f // JSON uses snake_case keys matching the API schema. @@ -102,6 +102,8 @@ var servicesCreateCmd = &cobra.Command{ key = "static_build_cmd" case "static-dir": key = "static_dir" + case "node": + key = "node_id" } body[key] = v } @@ -159,7 +161,7 @@ var servicesUpdateCmd = &cobra.Command{ return err } body := map[string]interface{}{} - for _, f := range []string{"name", "image", "command", "domain", "restart-policy", "healthcheck-path", "cpu", "memory", "builder", "cpu-reserve", "memory-reserve", "static-cmd", "static-dir"} { + for _, f := range []string{"name", "image", "command", "domain", "restart-policy", "healthcheck-path", "cpu", "memory", "builder", "cpu-reserve", "memory-reserve", "static-cmd", "static-dir", "node"} { if cmd.Flags().Changed(f) { key := f switch f { @@ -175,6 +177,8 @@ var servicesUpdateCmd = &cobra.Command{ key = "static_build_cmd" case "static-dir": key = "static_dir" + case "node": + key = "node_id" } v, _ := cmd.Flags().GetString(f) body[key] = v @@ -490,6 +494,7 @@ func init() { f.String("memory-reserve", "", "soft memory reservation (e.g. 128Mi)") f.String("static-cmd", "", "build command for the static builder") 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)") uf := servicesUpdateCmd.Flags() uf.String("name", "", "service name") @@ -514,6 +519,7 @@ func init() { uf.String("static-dir", "", "output dir for the static builder (empty clears)") uf.Bool("sleep", false, "enable scale-to-zero on idle (sleep mode)") 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") for _, a := range []string{"start", "stop", "restart", "redeploy", "sleep", "wake"} { verb := a diff --git a/app/backend/internal/database/sqlcdb/agents.sql.go b/app/backend/internal/database/sqlcdb/agents.sql.go index 8976f82..73e6454 100644 --- a/app/backend/internal/database/sqlcdb/agents.sql.go +++ b/app/backend/internal/database/sqlcdb/agents.sql.go @@ -14,6 +14,18 @@ import ( "github.com/sqlc-dev/pqtype" ) +const clearServiceNodePins = `-- name: ClearServiceNodePins :execrows +UPDATE services SET node_id = NULL, updated_at = NOW() WHERE node_id = $1 +` + +func (q *Queries) ClearServiceNodePins(ctx context.Context, nodeID sql.NullString) (int64, error) { + result, err := q.db.ExecContext(ctx, clearServiceNodePins, nodeID) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + const completeCommand = `-- name: CompleteCommand :one UPDATE agent_commands SET status = $3, @@ -63,7 +75,7 @@ INSERT INTO node_agents ( id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, metadata ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) -RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune +RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable ` type CreateAgentParams struct { @@ -110,6 +122,7 @@ func (q *Queries) CreateAgent(ctx context.Context, arg CreateAgentParams) (NodeA &i.UpdatedAt, &i.Metadata, &i.AutoPrune, + &i.Schedulable, ) return i, err } @@ -229,8 +242,22 @@ func (q *Queries) DeleteAgent(ctx context.Context, id string) error { return err } +const deleteServiceContainersOnAgent = `-- name: DeleteServiceContainersOnAgent :exec +DELETE FROM container_instances WHERE node_agent_id = $1 AND service_id = $2 +` + +type DeleteServiceContainersOnAgentParams struct { + NodeAgentID string `json:"node_agent_id"` + ServiceID string `json:"service_id"` +} + +func (q *Queries) DeleteServiceContainersOnAgent(ctx context.Context, arg DeleteServiceContainersOnAgentParams) error { + _, err := q.db.ExecContext(ctx, deleteServiceContainersOnAgent, arg.NodeAgentID, arg.ServiceID) + return err +} + const getAgent = `-- name: GetAgent :one -SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune FROM node_agents WHERE id = $1 +SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable FROM node_agents WHERE id = $1 ` func (q *Queries) GetAgent(ctx context.Context, id string) (NodeAgent, error) { @@ -251,12 +278,13 @@ func (q *Queries) GetAgent(ctx context.Context, id string) (NodeAgent, error) { &i.UpdatedAt, &i.Metadata, &i.AutoPrune, + &i.Schedulable, ) return i, err } const getAgentByHostAndIP = `-- name: GetAgentByHostAndIP :one -SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune FROM node_agents +SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable FROM node_agents WHERE hostname = $1 AND ip_address = $2 LIMIT 1 ` @@ -284,6 +312,7 @@ func (q *Queries) GetAgentByHostAndIP(ctx context.Context, arg GetAgentByHostAnd &i.UpdatedAt, &i.Metadata, &i.AutoPrune, + &i.Schedulable, ) return i, err } @@ -491,7 +520,7 @@ func (q *Queries) ListAgentHeartbeatsSince(ctx context.Context, arg ListAgentHea } const listAgents = `-- name: ListAgents :many -SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune FROM node_agents ORDER BY created_at ASC +SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable FROM node_agents ORDER BY created_at ASC ` func (q *Queries) ListAgents(ctx context.Context) ([]NodeAgent, error) { @@ -518,6 +547,7 @@ func (q *Queries) ListAgents(ctx context.Context) ([]NodeAgent, error) { &i.UpdatedAt, &i.Metadata, &i.AutoPrune, + &i.Schedulable, ); err != nil { return nil, err } @@ -661,6 +691,53 @@ func (q *Queries) ListPendingCommands(ctx context.Context, nodeAgentID string) ( return items, nil } +const pickLeastLoadedAgent = `-- name: PickLeastLoadedAgent :one +SELECT a.id FROM node_agents a +LEFT JOIN LATERAL ( + SELECT h.container_count FROM agent_heartbeats h + WHERE h.node_agent_id = a.id + ORDER BY h.timestamp DESC LIMIT 1 +) h ON true +WHERE a.status = 'online' AND a.schedulable +ORDER BY COALESCE(h.container_count, 0) ASC, a.created_at ASC +LIMIT 1 +` + +func (q *Queries) PickLeastLoadedAgent(ctx context.Context) (string, error) { + row := q.db.QueryRowContext(ctx, pickLeastLoadedAgent) + var id string + err := row.Scan(&id) + return id, err +} + +const scrubCommandPayload = `-- name: ScrubCommandPayload :exec +UPDATE agent_commands SET payload = $2, updated_at = NOW() WHERE id = $1 +` + +type ScrubCommandPayloadParams struct { + ID string `json:"id"` + Payload pqtype.NullRawMessage `json:"payload"` +} + +func (q *Queries) ScrubCommandPayload(ctx context.Context, arg ScrubCommandPayloadParams) error { + _, err := q.db.ExecContext(ctx, scrubCommandPayload, arg.ID, arg.Payload) + return err +} + +const setAgentSchedulable = `-- name: SetAgentSchedulable :exec +UPDATE node_agents SET schedulable = $2, updated_at = NOW() WHERE id = $1 +` + +type SetAgentSchedulableParams struct { + ID string `json:"id"` + Schedulable bool `json:"schedulable"` +} + +func (q *Queries) SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulableParams) error { + _, err := q.db.ExecContext(ctx, setAgentSchedulable, arg.ID, arg.Schedulable) + return err +} + const updateAgent = `-- name: UpdateAgent :one UPDATE node_agents SET name = $2, @@ -674,9 +751,10 @@ UPDATE node_agents SET last_heartbeat = $10, metadata = $11, auto_prune = $12, + schedulable = $13, updated_at = NOW() WHERE id = $1 -RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune +RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable ` type UpdateAgentParams struct { @@ -692,6 +770,7 @@ type UpdateAgentParams struct { LastHeartbeat sql.NullTime `json:"last_heartbeat"` Metadata pqtype.NullRawMessage `json:"metadata"` AutoPrune bool `json:"auto_prune"` + Schedulable bool `json:"schedulable"` } func (q *Queries) UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeAgent, error) { @@ -708,6 +787,7 @@ func (q *Queries) UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeA arg.LastHeartbeat, arg.Metadata, arg.AutoPrune, + arg.Schedulable, ) var i NodeAgent err := row.Scan( @@ -725,6 +805,7 @@ func (q *Queries) UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeA &i.UpdatedAt, &i.Metadata, &i.AutoPrune, + &i.Schedulable, ) return i, err } @@ -769,3 +850,52 @@ func (q *Queries) UpdateContainerStatus(ctx context.Context, arg UpdateContainer _, err := q.db.ExecContext(ctx, updateContainerStatus, arg.ID, arg.Status) return err } + +const upsertServiceContainer = `-- name: UpsertServiceContainer :exec +INSERT INTO container_instances ( + id, name, image, project_id, service_id, node_agent_id, + status, resources, ports, environment, volumes, networks, + restart_policy, health_check +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) +ON CONFLICT (id) DO UPDATE SET + image = EXCLUDED.image, + status = EXCLUDED.status, + updated_at = NOW() +` + +type UpsertServiceContainerParams struct { + ID string `json:"id"` + Name string `json:"name"` + Image string `json:"image"` + ProjectID string `json:"project_id"` + ServiceID string `json:"service_id"` + NodeAgentID string `json:"node_agent_id"` + Status pqtype.NullRawMessage `json:"status"` + Resources pqtype.NullRawMessage `json:"resources"` + Ports pqtype.NullRawMessage `json:"ports"` + Environment pqtype.NullRawMessage `json:"environment"` + Volumes pqtype.NullRawMessage `json:"volumes"` + Networks pqtype.NullRawMessage `json:"networks"` + RestartPolicy pqtype.NullRawMessage `json:"restart_policy"` + HealthCheck pqtype.NullRawMessage `json:"health_check"` +} + +func (q *Queries) UpsertServiceContainer(ctx context.Context, arg UpsertServiceContainerParams) error { + _, err := q.db.ExecContext(ctx, upsertServiceContainer, + arg.ID, + arg.Name, + arg.Image, + arg.ProjectID, + arg.ServiceID, + arg.NodeAgentID, + arg.Status, + arg.Resources, + arg.Ports, + arg.Environment, + arg.Volumes, + arg.Networks, + arg.RestartPolicy, + arg.HealthCheck, + ) + return err +} diff --git a/app/backend/internal/database/sqlcdb/models.go b/app/backend/internal/database/sqlcdb/models.go index bed7c64..4310934 100644 --- a/app/backend/internal/database/sqlcdb/models.go +++ b/app/backend/internal/database/sqlcdb/models.go @@ -176,6 +176,7 @@ type NodeAgent struct { UpdatedAt sql.NullTime `json:"updated_at"` Metadata pqtype.NullRawMessage `json:"metadata"` AutoPrune bool `json:"auto_prune"` + Schedulable bool `json:"schedulable"` } type Notification struct { @@ -274,6 +275,7 @@ type Service struct { MemoryReserve string `json:"memory_reserve"` StaticBuildCmd string `json:"static_build_cmd"` StaticDir string `json:"static_dir"` + NodeID sql.NullString `json:"node_id"` } type ServiceDomain struct { diff --git a/app/backend/internal/database/sqlcdb/querier.go b/app/backend/internal/database/sqlcdb/querier.go index 4f7bb29..af94fb2 100644 --- a/app/backend/internal/database/sqlcdb/querier.go +++ b/app/backend/internal/database/sqlcdb/querier.go @@ -12,6 +12,7 @@ import ( ) type Querier interface { + ClearServiceNodePins(ctx context.Context, nodeID sql.NullString) (int64, error) CompleteCommand(ctx context.Context, arg CompleteCommandParams) (AgentCommand, error) CountDatabaseServicesByUserAndName(ctx context.Context, arg CountDatabaseServicesByUserAndNameParams) (int64, error) CountDatabasesUsingBackupTarget(ctx context.Context, targetID sql.NullString) (int64, error) @@ -43,6 +44,7 @@ type Querier interface { DeleteDatabaseServiceByIDAndUser(ctx context.Context, arg DeleteDatabaseServiceByIDAndUserParams) error DeleteOutboundWebhookByIDAndUser(ctx context.Context, arg DeleteOutboundWebhookByIDAndUserParams) error DeleteProjectByID(ctx context.Context, projectID uuid.UUID) (int64, error) + DeleteServiceContainersOnAgent(ctx context.Context, arg DeleteServiceContainersOnAgentParams) error DeleteUserInvite(ctx context.Context, id uuid.UUID) error DeleteUserTemplate(ctx context.Context, arg DeleteUserTemplateParams) (int64, error) GetActiveAgentAuthTokenByHash(ctx context.Context, tokenHash string) (AgentAuthToken, error) @@ -100,8 +102,11 @@ type Querier interface { MarkAllNotificationsReadByUser(ctx context.Context, arg MarkAllNotificationsReadByUserParams) error MarkNotificationReadByIDAndUser(ctx context.Context, arg MarkNotificationReadByIDAndUserParams) error MarkUserInviteUsed(ctx context.Context, arg MarkUserInviteUsedParams) error + PickLeastLoadedAgent(ctx context.Context) (string, error) RevokeAgentAuthToken(ctx context.Context, id uuid.UUID) (AgentAuthToken, error) RevokeUserToken(ctx context.Context, arg RevokeUserTokenParams) (int64, error) + ScrubCommandPayload(ctx context.Context, arg ScrubCommandPayloadParams) error + SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulableParams) error SetDatabaseBackupRemoteKeyByID(ctx context.Context, arg SetDatabaseBackupRemoteKeyByIDParams) error SetDatabaseBackupScheduleByIDAndUser(ctx context.Context, arg SetDatabaseBackupScheduleByIDAndUserParams) error SetDatabaseBackupStatusByID(ctx context.Context, arg SetDatabaseBackupStatusByIDParams) error @@ -127,6 +132,7 @@ type Querier interface { UpdateUserTemplate(ctx context.Context, arg UpdateUserTemplateParams) (int64, error) UpdateWebhookDeliveryResult(ctx context.Context, arg UpdateWebhookDeliveryResultParams) error UpsertEnvironmentVariable(ctx context.Context, arg UpsertEnvironmentVariableParams) error + UpsertServiceContainer(ctx context.Context, arg UpsertServiceContainerParams) error UpsertServiceTemplate(ctx context.Context, arg UpsertServiceTemplateParams) error } diff --git a/app/backend/internal/deployment/engine.go b/app/backend/internal/deployment/engine.go index d3e85b0..54d2363 100644 --- a/app/backend/internal/deployment/engine.go +++ b/app/backend/internal/deployment/engine.go @@ -11,10 +11,24 @@ import ( "containr/internal/types" ) +// NodeRunner deploys a service's containers onto a remote node agent. +// Wired at boot; nil means every service runs on the local Docker host. +type NodeRunner interface { + // ReconcileOnNode brings the node's containers in line with the spec and + // returns the resulting runtime state. + ReconcileOnNode(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error) + // ControlService fans a lifecycle action ("start"|"stop"|"restart"|"remove") + // out to every replica of the service on the agent. + ControlService(ctx context.Context, serviceID, agentID, action string) error + // RemoveService tears down every replica of the service on the agent. + RemoveService(ctx context.Context, serviceID, agentID string) error +} + type DeploymentEngine struct { buildManager *build.BuildManager dockerClient *docker.Client scheduler *Scheduler + nodeRunner NodeRunner deployments map[string]*Deployment deploymentLog chan *DeploymentEvent } @@ -67,6 +81,10 @@ type ServiceConfig struct { // PublishedPort is the host port bound by the previous deployment; replica 0 // reuses it so public URLs survive restarts. 0 = pick ephemeral. PublishedPort int32 `json:"published_port,omitempty"` + // NodeID pins the service to a registered node agent; "" = local Docker + // host. Remote placement requires a registry-pullable image — builds stay + // on the local host (build-on-node is not yet supported). + NodeID string `json:"node_id,omitempty"` } type PortMapping struct { @@ -177,6 +195,11 @@ func (de *DeploymentEngine) DockerClient() *docker.Client { return de.dockerClient } +// SetNodeRunner wires remote-node dispatch for services pinned to an agent. +func (de *DeploymentEngine) SetNodeRunner(r NodeRunner) { + de.nodeRunner = r +} + func NewDeploymentEngine(buildManager *build.BuildManager, dockerClient *docker.Client) *DeploymentEngine { return &DeploymentEngine{ buildManager: buildManager, @@ -352,6 +375,7 @@ func (de *DeploymentEngine) deployService(ctx context.Context, deployment *Deplo Maintenance: cfg.Maintenance, MaintenanceURL: cfg.MaintenanceURL, BasicAuthUsers: cfg.BasicAuthUsers, + NodeID: cfg.NodeID, } for _, pm := range cfg.PortMappings { if spec.Port == 0 { diff --git a/app/backend/internal/deployment/runtime.go b/app/backend/internal/deployment/runtime.go index d81a0f9..3ed0e7c 100644 --- a/app/backend/internal/deployment/runtime.go +++ b/app/backend/internal/deployment/runtime.go @@ -55,6 +55,8 @@ type RuntimeSpec struct { Maintenance bool // redirect traffic to MaintenanceURL MaintenanceURL string // absolute URL; empty = serve empty response BasicAuthUsers string // htpasswd-format user:hash pairs, comma-separated + // NodeID pins the service to a node agent; "" runs on the local host. + NodeID string } // RuntimeContainer describes one live replica. @@ -151,8 +153,15 @@ func (de *DeploymentEngine) ListServiceContainers(ctx context.Context, serviceID // ReconcileService brings live containers in line with the spec. When the // spec hash changed (image/env/ports/etc.) all replicas are recreated; -// otherwise only the replica delta is applied. +// otherwise only the replica delta is applied. A pinned NodeID routes the +// whole reconcile to that agent's command queue instead of local Docker. func (de *DeploymentEngine) ReconcileService(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error) { + if spec.NodeID != "" { + if de.nodeRunner == nil { + return nil, fmt.Errorf("service is pinned to node %s but remote dispatch is unavailable", spec.NodeID) + } + return de.nodeRunner.ReconcileOnNode(ctx, spec) + } if spec.Replicas < 1 { spec.Replicas = 1 } @@ -498,6 +507,25 @@ func (de *DeploymentEngine) RemoveServiceContainers(ctx context.Context, service return nil } +// ControlServiceOnNode fans a lifecycle action out to the service's replicas +// on a remote node agent. Callers resolve the pin; nil runner means the +// platform was booted without node support. +func (de *DeploymentEngine) ControlServiceOnNode(ctx context.Context, serviceID, nodeID, action string) error { + if de.nodeRunner == nil { + return fmt.Errorf("remote node control is not configured") + } + return de.nodeRunner.ControlService(ctx, serviceID, nodeID, action) +} + +// RemoveServiceContainersOnNode tears down the service's replicas on a remote +// node agent. +func (de *DeploymentEngine) RemoveServiceContainersOnNode(ctx context.Context, serviceID, nodeID string) error { + if de.nodeRunner == nil { + return fmt.Errorf("remote node control is not configured") + } + return de.nodeRunner.RemoveService(ctx, serviceID, nodeID) +} + // RemoveProjectContainers deletes every managed container in a project. func (de *DeploymentEngine) RemoveProjectContainers(ctx context.Context, projectID string) error { containers, err := de.dockerClient.ListContainersFiltered(ctx, filters.NewArgs( diff --git a/app/backend/internal/mcpserver/server.go b/app/backend/internal/mcpserver/server.go index 16dfdf3..35a1016 100644 --- a/app/backend/internal/mcpserver/server.go +++ b/app/backend/internal/mcpserver/server.go @@ -53,7 +53,7 @@ func init() { // 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_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.", "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).", "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_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}) @@ -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_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_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.", "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).", "PUT", "/services/{id}", []string{"id"}, nil, false}) // Domains register(spec{"containr_domains_list", "List domains attached to a service", "GET", "/services/{id}/domains", []string{"id"}, nil, false}) @@ -174,6 +174,9 @@ func init() { register(spec{"containr_agent_tokens_issue", "Issue an agent onboarding token (admin). Args: label.", "POST", "/agent-tokens", nil, []string{"label"}, false}) register(spec{"containr_agent_tokens_list", "List agent onboarding tokens (admin)", "GET", "/agent-tokens", nil, nil, false}) register(spec{"containr_nodes_update", "Update a node agent (admin). Args: id; optional auto_prune, name.", "PUT", "/agents/{id}", []string{"id"}, []string{"auto_prune", "name"}, false}) + register(spec{"containr_nodes_cordon", "Cordon a node: blocks new pins and 'auto' placements; running replicas stay (admin). Args: id.", "POST", "/agents/{id}/cordon", []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_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_commands", "List recent commands sent to a node agent (admin). Args: id.", "GET", "/agents/{id}/commands", []string{"id"}, nil, false}) diff --git a/app/backend/migrations_goose/20261108000000_service_node_pin.sql b/app/backend/migrations_goose/20261108000000_service_node_pin.sql new file mode 100644 index 0000000..6928c0e --- /dev/null +++ b/app/backend/migrations_goose/20261108000000_service_node_pin.sql @@ -0,0 +1,19 @@ +-- +goose Up +ALTER TABLE services ADD COLUMN IF NOT EXISTS node_id VARCHAR(255); + +-- +goose StatementBegin +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'services_node_id_fkey') THEN + ALTER TABLE services + ADD CONSTRAINT services_node_id_fkey + FOREIGN KEY (node_id) REFERENCES node_agents(id) ON DELETE SET NULL; + END IF; +END $$; +-- +goose StatementEnd + +CREATE INDEX IF NOT EXISTS idx_services_node_id ON services (node_id) WHERE node_id IS NOT NULL; + +-- +goose Down +DROP INDEX IF EXISTS idx_services_node_id; +ALTER TABLE services DROP COLUMN IF EXISTS node_id; diff --git a/app/backend/migrations_goose/20261109000000_node_schedulable.sql b/app/backend/migrations_goose/20261109000000_node_schedulable.sql new file mode 100644 index 0000000..d6c43be --- /dev/null +++ b/app/backend/migrations_goose/20261109000000_node_schedulable.sql @@ -0,0 +1,7 @@ +-- +goose Up +ALTER TABLE node_agents ADD COLUMN IF NOT EXISTS schedulable BOOLEAN NOT NULL DEFAULT true; +CREATE INDEX IF NOT EXISTS idx_node_agents_schedulable ON node_agents(schedulable) WHERE NOT schedulable; + +-- +goose Down +DROP INDEX IF EXISTS idx_node_agents_schedulable; +ALTER TABLE node_agents DROP COLUMN IF EXISTS schedulable; diff --git a/app/backend/sqlc/queries/agents.sql b/app/backend/sqlc/queries/agents.sql index 9d5df7d..a35fef8 100644 --- a/app/backend/sqlc/queries/agents.sql +++ b/app/backend/sqlc/queries/agents.sql @@ -29,6 +29,7 @@ UPDATE node_agents SET last_heartbeat = $10, metadata = $11, auto_prune = $12, + schedulable = $13, updated_at = NOW() WHERE id = $1 RETURNING *; @@ -113,3 +114,37 @@ WHERE id = $1; SELECT * FROM agent_commands WHERE node_agent_id = $1 AND type = $2 ORDER BY created_at DESC LIMIT 1; + +-- name: PickLeastLoadedAgent :one +SELECT a.id FROM node_agents a +LEFT JOIN LATERAL ( + SELECT h.container_count FROM agent_heartbeats h + WHERE h.node_agent_id = a.id + ORDER BY h.timestamp DESC LIMIT 1 +) h ON true +WHERE a.status = 'online' AND a.schedulable +ORDER BY COALESCE(h.container_count, 0) ASC, a.created_at ASC +LIMIT 1; + +-- name: SetAgentSchedulable :exec +UPDATE node_agents SET schedulable = $2, updated_at = NOW() WHERE id = $1; + +-- name: ClearServiceNodePins :execrows +UPDATE services SET node_id = NULL, updated_at = NOW() WHERE node_id = $1; + +-- name: UpsertServiceContainer :exec +INSERT INTO container_instances ( + id, name, image, project_id, service_id, node_agent_id, + status, resources, ports, environment, volumes, networks, + restart_policy, health_check +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) +ON CONFLICT (id) DO UPDATE SET + image = EXCLUDED.image, + status = EXCLUDED.status, + updated_at = NOW(); + +-- name: DeleteServiceContainersOnAgent :exec +DELETE FROM container_instances WHERE node_agent_id = $1 AND service_id = $2; + +-- name: ScrubCommandPayload :exec +UPDATE agent_commands SET payload = $2, updated_at = NOW() WHERE id = $1; diff --git a/app/backend/sqlc/schema.sql b/app/backend/sqlc/schema.sql index 237f554..bbc6929 100644 --- a/app/backend/sqlc/schema.sql +++ b/app/backend/sqlc/schema.sql @@ -66,7 +66,8 @@ CREATE TABLE services ( cpu_reserve VARCHAR(20) NOT NULL DEFAULT '', memory_reserve VARCHAR(20) 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) ); CREATE TABLE registries ( @@ -193,7 +194,8 @@ CREATE TABLE node_agents ( created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), metadata JSONB, - auto_prune BOOLEAN NOT NULL DEFAULT false + auto_prune BOOLEAN NOT NULL DEFAULT false, + schedulable BOOLEAN NOT NULL DEFAULT true ); CREATE TABLE container_instances ( diff --git a/app/frontend/src/features/nodes/NodeDetailPage.tsx b/app/frontend/src/features/nodes/NodeDetailPage.tsx index ef45c45..1048d2d 100644 --- a/app/frontend/src/features/nodes/NodeDetailPage.tsx +++ b/app/frontend/src/features/nodes/NodeDetailPage.tsx @@ -2,7 +2,9 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; import { useMemo, useState } from 'react'; import { Link, useNavigate, useParams } from 'react-router-dom'; import { + cordonAgent, deleteAgent, + drainAgent, getAgent, getAgentMetrics, listAgentCommands, @@ -43,6 +45,7 @@ export function NodeDetailPage() { const [rename, setRename] = useState(null); const [confirmDelete, setConfirmDelete] = useState(false); const [confirmVolumes, setConfirmVolumes] = useState(false); + const [confirmDrain, setConfirmDrain] = useState(false); const agentQuery = useQuery({ queryKey: ['node-agent', id], @@ -86,6 +89,14 @@ export function NodeDetailPage() { mutationFn: () => deleteAgent(id), onSuccess: () => navigate('/usage'), }); + const cordonMutation = useMutation({ + mutationFn: (cordoned: boolean) => cordonAgent(id, cordoned), + onSuccess: invalidate, + }); + const drainMutation = useMutation({ + mutationFn: () => drainAgent(id), + onSuccess: () => { setConfirmDrain(false); invalidate(); }, + }); const agent = agentQuery.data; const metrics = useMemo(() => metricsQuery.data ?? [], [metricsQuery.data]); @@ -168,6 +179,11 @@ export function NodeDetailPage() { : 'border-[var(--error)]/40 text-[var(--error)]'}`}> {agent.status} + {!agent.schedulable && ( + + cordoned + + )} {isAdmin && ( <> + + + + + )} + {service.nodeName && nodeDraft === null && ( + running on {service.nodeName} + )} + + {updateServiceMutation.isError && ( +

{(updateServiceMutation.error as Error).message}

+ )} + + )} + {!isDemoMode && (
diff --git a/app/frontend/src/generated/api-types.ts b/app/frontend/src/generated/api-types.ts index 919a330..645afcd 100644 --- a/app/frontend/src/generated/api-types.ts +++ b/app/frontend/src/generated/api-types.ts @@ -7977,6 +7977,51 @@ export interface paths { patch?: never; trace?: never; }; + "/nodes/options": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Node options for service placement + * @description Minimal id/name/status list of registered node agents — available to any authenticated user for the service placement picker. Full agent data is admin-only under /agents. + */ + get: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Node options */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + nodes?: { + id?: string; + name?: string; + status?: string; + }[]; + }; + }; + }; + }; + }; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/agents": { parameters: { query?: never; @@ -8514,6 +8559,179 @@ export interface paths { patch?: never; trace?: never; }; + "/agents/{id}/cordon": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Cordon a node + * @description Sets `schedulable=false`: new explicit pins are rejected and `auto` placement skips the node. Already-pinned services keep running and can still reconcile — drain to evict them. + */ + post: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Node agent ID */ + id: components["parameters"]["AgentIdParam"]; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Node cordoned */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + id?: string; + schedulable?: boolean; + }; + }; + }; + /** @description Agent not found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["Error"]; + }; + }; + }; + }; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/agents/{id}/uncordon": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Uncordon a node + * @description Re-opens the node for placement (`schedulable=true`). + */ + post: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Node agent ID */ + id: components["parameters"]["AgentIdParam"]; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Node uncordoned */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + id?: string; + schedulable?: boolean; + }; + }; + }; + /** @description Agent not found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["Error"]; + }; + }; + }; + }; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/agents/{id}/drain": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Drain a node + * @description Cordons the node, removes every service container via the agent command queue, and clears the affected services' `node_id` pins — they fall back to the local host on next deploy and can be re-pinned elsewhere. Requires the node to be online so it can acknowledge the removals. + */ + post: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Node agent ID */ + id: components["parameters"]["AgentIdParam"]; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Node drained */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + id?: string; + schedulable?: boolean; + services_drained?: number; + services_unpinned?: number; + }; + }; + }; + /** @description Agent not found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["Error"]; + }; + }; + /** @description Node is not online */ + 409: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["Error"]; + }; + }; + }; + }; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/agents/{id}/metrics": { parameters: { query?: never; @@ -10656,6 +10874,10 @@ export interface components { sleep_enabled?: boolean; /** @description Idle timeout in minutes before sleeping (1-1440, default 15) */ sleep_idle_minutes?: number; + /** @description Node agent id this service is pinned to; empty/null = local Docker host. 'auto' on create/update resolves to the least-loaded online agent. Remote nodes run registry-pulled images only (no git builds, no Traefik domains). */ + node_id?: string; + /** @description Display name of the pinned node agent (read-only) */ + node_name?: string; /** @description Computed public URL (published port or domain) */ public_url?: string; /** @@ -10726,6 +10948,8 @@ export interface components { static_build_cmd?: string; /** @description Output directory for the static builder (default dist) */ static_dir?: string; + /** @description Node agent id to pin to, 'auto' = least-loaded online agent; omit/empty = local host */ + node_id?: string; }; UpdateServiceRequest: { /** @description Service name */ @@ -10787,6 +11011,8 @@ export interface components { sleep_enabled?: boolean; /** @description Idle timeout in minutes before sleeping (1-1440) */ 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) */ + node_id?: string; }; ServiceDomain: { /** Format: uuid */ @@ -11904,6 +12130,8 @@ export interface components { }; /** @description Enqueue a bounded docker prune on this node once per day. */ auto_prune?: boolean; + /** @description False on a cordoned/drained node — `auto` placement skips it and new explicit pins are rejected. */ + schedulable?: boolean; /** Format: date-time */ last_heartbeat?: string; /** Format: date-time */ diff --git a/app/frontend/src/generated/docs-snapshot.ts b/app/frontend/src/generated/docs-snapshot.ts index b9e5a6c..f42731a 100644 --- a/app/frontend/src/generated/docs-snapshot.ts +++ b/app/frontend/src/generated/docs-snapshot.ts @@ -40,7 +40,7 @@ export const bundledDocs: BundledDoc[] = [ "path": "guides/CLI.md", "section": "guides", "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 trigger a deployment (--no-cache, --commit-hash, --branch)\ndeployments list|get|logs|rollback|cancel\nlogs [--follow] runtime logs (SSE follow)\nexec -- 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 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 --public/--public=false toggles it (container recreates, data persists)\ndatabases register 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 --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 converts a compose file into\n a v2 graph config; deploy-graph \n deploys an ad-hoc service graph\nimport compose \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 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|analytics (admin)\nadmin overview|users|settings|audit-logs\nnotifications list|read\nwebhooks list|create|update|delete|deliveries|test\n outbound signed webhooks — create \n --event 'service.*' (repeatable, '*' = all);\n deliveries are HMAC-signed (X-Containr-Signature)\noperations live ops view: active deploys, queue depth,\n 24h failures, cron runs, backups\nactivity enriched activity feed; --severity/--category/\n --resource/--project/--limit filters\nshell interactive PTY shell inside the service\n container (bash→sh fallback; resizes follow)\nbanners list|create|update|delete announcement banners (admin);\n active banners print on `auth status`\ninvites list|create|revoke team invite links (admin); create\n prints the one-time acceptance URL\nadmin impersonate mint a 15-min token acting as a user\nadmin github-app status|manifest|convert\n self-provision the instance GitHub App\nup deploy the current directory\n```\n\n## containr up\n\nRun inside a git checkout: uses `remote.origin.url` + the current branch\nas the build source, creates a service named after the directory when none\nexists, then triggers a deployment.\n\n```bash\ncontainr up --project \ncontainr up --project --image ghcr.io/example/app:latest # image, no build\n```\n\nSet `project_id` on the active profile to omit `--project` (stored in\n`~/.containr.yaml` under `profiles..project_id`).\n\n## Service volumes\n\n`services update` manages mounts; flags replace the whole mount list:\n\n```bash\ncontainr services update --volume data:/data --bind /srv/cfg:/etc/app:ro\ncontainr services update --clear-volumes\n```\n\n`type` is `volume` (docker named volume, auto-created) or `bind` (host\npath). `target` must be absolute. Mounts apply on the next deploy or\nredeploy — `containr deploy ` to apply immediately.\n\n`volumes list` shows the node's docker volume inventory with in-use\nflags; `volumes delete ` removes an unused volume (admin scope).\n\n## Service domains and access\n\nA service can carry multiple hostnames; the default is mirrored to\n`services.domain`:\n\n```bash\ncontainr services domains add app.example.com --default\ncontainr services domains check # DNS preflight per domain\ncontainr services domains remove \n```\n\nDNS check compares each domain to the expected target — set\n`app_settings.public_ip` (or `PUBLIC_IP` env) so `wrong-target` is\nmeaningful; without it any resolvable domain reports `ok`.\n\nAccess gates apply via Traefik middlewares (no extra proxy):\n\n```bash\ncontainr services update --maintenance on\ncontainr services update --basic-auth admin:secret --basic-auth ops:pw2\n```\n\nMaintenance mode redirects all traffic to `/api/v1/maintenance`\n— set `app_settings.base_url` or `BASE_URL`; without it the service is\ntaken offline with empty responses. Basic-auth credentials are stored\nbcrypt-hashed (htpasswd) and gate every request.\n\n## Build strategy and resources\n\nSource-built services pick a builder per service:\n\n```bash\ncontainr services update --builder railpack # auto|railpack|nixpacks|dockerfile|static\ncontainr services update --builder static --static-cmd \"npm run build\" --static-dir dist\n```\n\n`auto` detects from repo contents (Dockerfile → railpack → nixpacks).\n`static` builds assets in a node stage and serves them from nginx —\n`--static-cmd` runs inside the build stage (default\n`npm ci && npm run build`), `--static-dir` is copied into the nginx image\n(default `dist`, exposed on port 80).\n\nResource reservations complement the hard `cpu`/`memory` limits:\n\n```bash\ncontainr services update --cpu-reserve 0.25 --memory-reserve 128Mi\n```\n\n`cpu`/`memory` are hard caps (NanoCPUs/Memory); `cpu_reserve`/\n`memory_reserve` are soft hints (CPU shares, MemoryReservation). Empty\nstring clears. `app_settings.default_cpu`/`default_memory` change the\ninstance-level defaults used at service creation;\n`app_settings.capacity_policy=block` fails deployments whose requested\nmemory exceeds node capacity (default `warn` logs only).\n\n## Sleep mode (scale-to-zero)\n\nServices can scale to zero after an idle window and resume on traffic:\n\n```bash\ncontainr services update --sleep --sleep-idle 15 # enable, 15 min idle timeout\ncontainr services update --sleep=false # disable\ncontainr services sleep # sleep now\ncontainr services wake # resume now\ncontainr services env-check # flag empty vars + unresolved ${{refs}}\n```\n\nIdleness is measured as flat container network I/O over the configured\nwindow (swept every 30s). Sleeping removes the workload containers and —\nwhen the service has domains — installs a busybox placeholder that holds\nthe Traefik route at higher priority, serves a reloading \"waking\" page,\nand pings `GET /api/v1/internal/wake/:id` to reconcile the service back.\nThe API must be reachable from containers for the callback (containerized\ndeploys work via the shared network; bare-metal needs `HOST=0.0.0.0`).\n\n## Clone, move, registries\n\n`services clone` duplicates config, volumes, domains, access gates,\nbuilder settings, and variables — ciphertext secret values are copied\nverbatim (no decrypt/re-encrypt). The clone starts stopped:\n\n```bash\ncontainr services clone --name staging-api --project --environment preview\ncontainr services move # name conflicts → 409\n```\n\nNote: cloned domains are verbatim copies — the same hostname attached to\ntwo services makes Traefik routing ambiguous. Re-domain or undeploy one.\n\n`registries` stores private-image pull credentials, owner-scoped and\nencrypted at rest. Matched by image host on every image-sourced deploy:\n\n```bash\ncontainr registries add ghcr ghcr.io --username --password \ncontainr registries add hub docker.io --username --password # Docker Hub\ncontainr registries update --username # empty --password keeps stored\ncontainr registries remove \n```\n\n## MCP server\n\n`containr-mcp` (`go build ./cmd/mcp`) exposes the same API as MCP tools\nover stdio — every command group above maps to a `containr_*` tool\n(~60 tools). Destructive tools require the caller to pass\n`confirm: true`; results are pretty JSON with `isError` on failures.\n\nClient configuration (Claude Code, Cursor, Devin, any stdio MCP host):\n\n```json\n{\n \"mcpServers\": {\n \"containr\": {\n \"command\": \"containr-mcp\",\n \"env\": {\n \"CONTAINR_API_URL\": \"https://containr.example.com\",\n \"CONTAINR_TOKEN\": \"cnp_...\"\n }\n }\n }\n}\n```\n\nScope the token deliberately: `read` is enough for inspection agents,\n`write` for deploy automation, `admin` only for platform operations.\n\nDeployments are serialized per service: a second `deploy` on a busy\nservice lands in `queued` state and runs when the active one finishes.\n`deployments cancel ` drops a queued deployment or aborts a running\none (builds/reconciles stop via context cancellation).\n" + "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 trigger a deployment (--no-cache, --commit-hash, --branch)\ndeployments list|get|logs|rollback|cancel\nlogs [--follow] runtime logs (SSE follow)\nexec -- 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 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 --public/--public=false toggles it (container recreates, data persists)\ndatabases register 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 --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 converts a compose file into\n a v2 graph config; deploy-graph \n deploys an ad-hoc service graph\nimport compose \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 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|analytics (admin)\nadmin overview|users|settings|audit-logs\nnotifications list|read\nwebhooks list|create|update|delete|deliveries|test\n outbound signed webhooks — create \n --event 'service.*' (repeatable, '*' = all);\n deliveries are HMAC-signed (X-Containr-Signature)\noperations live ops view: active deploys, queue depth,\n 24h failures, cron runs, backups\nactivity enriched activity feed; --severity/--category/\n --resource/--project/--limit filters\nshell interactive PTY shell inside the service\n container (bash→sh fallback; resizes follow)\nbanners list|create|update|delete announcement banners (admin);\n active banners print on `auth status`\ninvites list|create|revoke team invite links (admin); create\n prints the one-time acceptance URL\nadmin impersonate mint a 15-min token acting as a user\nadmin github-app status|manifest|convert\n self-provision the instance GitHub App\nup deploy the current directory\n```\n\n## containr up\n\nRun inside a git checkout: uses `remote.origin.url` + the current branch\nas the build source, creates a service named after the directory when none\nexists, then triggers a deployment.\n\n```bash\ncontainr up --project \ncontainr up --project --image ghcr.io/example/app:latest # image, no build\n```\n\nSet `project_id` on the active profile to omit `--project` (stored in\n`~/.containr.yaml` under `profiles..project_id`).\n\n## Service volumes\n\n`services update` manages mounts; flags replace the whole mount list:\n\n```bash\ncontainr services update --volume data:/data --bind /srv/cfg:/etc/app:ro\ncontainr services update --clear-volumes\n```\n\n`type` is `volume` (docker named volume, auto-created) or `bind` (host\npath). `target` must be absolute. Mounts apply on the next deploy or\nredeploy — `containr deploy ` to apply immediately.\n\n`volumes list` shows the node's docker volume inventory with in-use\nflags; `volumes delete ` removes an unused volume (admin scope).\n\n## Node placement\n\nPin a service to a registered node agent, or let the scheduler pick the\nleast-loaded online node:\n\n```bash\ncontainr services create --project --name api --image ghcr.io/x/api:v1 --node \ncontainr services update --node auto # least-loaded online node\ncontainr services update --node local # move back to this server\n```\n\nRemote nodes run registry-pulled images only — git builds, domains\n(Traefik lives on the host), and sleep detection stay local. Start,\nstop, restart, redeploy, and delete all fan out to the pinned node's\nagent. `services get` shows the resolved `node_name`.\n\nTake a node out of rotation or empty it entirely (admin):\n\n```bash\ncontainr nodes cordon # no new pins; running replicas stay\ncontainr nodes uncordon \ncontainr nodes drain # cordon + remove its containers + unpin services\n```\n\n## Service domains and access\n\nA service can carry multiple hostnames; the default is mirrored to\n`services.domain`:\n\n```bash\ncontainr services domains add app.example.com --default\ncontainr services domains check # DNS preflight per domain\ncontainr services domains remove \n```\n\nDNS check compares each domain to the expected target — set\n`app_settings.public_ip` (or `PUBLIC_IP` env) so `wrong-target` is\nmeaningful; without it any resolvable domain reports `ok`.\n\nAccess gates apply via Traefik middlewares (no extra proxy):\n\n```bash\ncontainr services update --maintenance on\ncontainr services update --basic-auth admin:secret --basic-auth ops:pw2\n```\n\nMaintenance mode redirects all traffic to `/api/v1/maintenance`\n— set `app_settings.base_url` or `BASE_URL`; without it the service is\ntaken offline with empty responses. Basic-auth credentials are stored\nbcrypt-hashed (htpasswd) and gate every request.\n\n## Build strategy and resources\n\nSource-built services pick a builder per service:\n\n```bash\ncontainr services update --builder railpack # auto|railpack|nixpacks|dockerfile|static\ncontainr services update --builder static --static-cmd \"npm run build\" --static-dir dist\n```\n\n`auto` detects from repo contents (Dockerfile → railpack → nixpacks).\n`static` builds assets in a node stage and serves them from nginx —\n`--static-cmd` runs inside the build stage (default\n`npm ci && npm run build`), `--static-dir` is copied into the nginx image\n(default `dist`, exposed on port 80).\n\nResource reservations complement the hard `cpu`/`memory` limits:\n\n```bash\ncontainr services update --cpu-reserve 0.25 --memory-reserve 128Mi\n```\n\n`cpu`/`memory` are hard caps (NanoCPUs/Memory); `cpu_reserve`/\n`memory_reserve` are soft hints (CPU shares, MemoryReservation). Empty\nstring clears. `app_settings.default_cpu`/`default_memory` change the\ninstance-level defaults used at service creation;\n`app_settings.capacity_policy=block` fails deployments whose requested\nmemory exceeds node capacity (default `warn` logs only).\n\n## Sleep mode (scale-to-zero)\n\nServices can scale to zero after an idle window and resume on traffic:\n\n```bash\ncontainr services update --sleep --sleep-idle 15 # enable, 15 min idle timeout\ncontainr services update --sleep=false # disable\ncontainr services sleep # sleep now\ncontainr services wake # resume now\ncontainr services env-check # flag empty vars + unresolved ${{refs}}\n```\n\nIdleness is measured as flat container network I/O over the configured\nwindow (swept every 30s). Sleeping removes the workload containers and —\nwhen the service has domains — installs a busybox placeholder that holds\nthe Traefik route at higher priority, serves a reloading \"waking\" page,\nand pings `GET /api/v1/internal/wake/:id` to reconcile the service back.\nThe API must be reachable from containers for the callback (containerized\ndeploys work via the shared network; bare-metal needs `HOST=0.0.0.0`).\n\n## Clone, move, registries\n\n`services clone` duplicates config, volumes, domains, access gates,\nbuilder settings, and variables — ciphertext secret values are copied\nverbatim (no decrypt/re-encrypt). The clone starts stopped:\n\n```bash\ncontainr services clone --name staging-api --project --environment preview\ncontainr services move # name conflicts → 409\n```\n\nNote: cloned domains are verbatim copies — the same hostname attached to\ntwo services makes Traefik routing ambiguous. Re-domain or undeploy one.\n\n`registries` stores private-image pull credentials, owner-scoped and\nencrypted at rest. Matched by image host on every image-sourced deploy:\n\n```bash\ncontainr registries add ghcr ghcr.io --username --password \ncontainr registries add hub docker.io --username --password # Docker Hub\ncontainr registries update --username # empty --password keeps stored\ncontainr registries remove \n```\n\n## MCP server\n\n`containr-mcp` (`go build ./cmd/mcp`) exposes the same API as MCP tools\nover stdio — every command group above maps to a `containr_*` tool\n(~60 tools). Destructive tools require the caller to pass\n`confirm: true`; results are pretty JSON with `isError` on failures.\n\nClient configuration (Claude Code, Cursor, Devin, any stdio MCP host):\n\n```json\n{\n \"mcpServers\": {\n \"containr\": {\n \"command\": \"containr-mcp\",\n \"env\": {\n \"CONTAINR_API_URL\": \"https://containr.example.com\",\n \"CONTAINR_TOKEN\": \"cnp_...\"\n }\n }\n }\n}\n```\n\nScope the token deliberately: `read` is enough for inspection agents,\n`write` for deploy automation, `admin` only for platform operations.\n\nDeployments are serialized per service: a second `deploy` on a busy\nservice lands in `queued` state and runs when the active one finishes.\n`deployments cancel ` drops a queued deployment or aborts a running\none (builds/reconciles stop via context cancellation).\n" }, { "path": "guides/CLOUDFLARE_SETUP.md", diff --git a/app/frontend/src/lib/api-client.ts b/app/frontend/src/lib/api-client.ts index 0d7e86f..ad340d8 100644 --- a/app/frontend/src/lib/api-client.ts +++ b/app/frontend/src/lib/api-client.ts @@ -84,9 +84,22 @@ export type ServiceEntity = { staticDir?: string; sleepEnabled?: boolean; sleepIdleMinutes?: number; + nodeId?: string; + nodeName?: string; publicUrl?: string; }; +export type NodeOption = { + id: string; + name: string; + status: string; +}; + +export async function listNodeOptions(): Promise { + const payload = await requestJson<{ nodes?: NodeOption[] }>('/nodes/options'); + return payload.nodes ?? []; +} + export type ServiceDomain = { id: string; service_id: string; @@ -343,6 +356,7 @@ export type NodeAgentEntity = { capabilities: AgentCapabilities; resources: NodeResources; autoPrune: boolean; + schedulable: boolean; metadata: Record; lastHeartbeat?: string; createdAt?: string; @@ -680,6 +694,8 @@ function normalizeService(service: RawService): ServiceEntity | null { staticDir: service.static_dir, sleepEnabled: service.sleep_enabled, sleepIdleMinutes: service.sleep_idle_minutes, + nodeId: service.node_id, + nodeName: service.node_name, publicUrl: service.public_url, }; } @@ -1081,6 +1097,7 @@ function normalizeAgent(agent: RawNodeAgent): NodeAgentEntity | null { }, lastHeartbeat: agent.last_heartbeat, autoPrune: agent.auto_prune ?? false, + schedulable: agent.schedulable ?? true, metadata: (agent.metadata ?? {}) as Record, createdAt: agent.created_at, updatedAt: agent.updated_at, @@ -1135,6 +1152,16 @@ export async function pruneAgent(id: string, body?: { until?: string; volumes?: }); } +export async function cordonAgent(id: string, cordoned: boolean): Promise { + await requestJson(`/agents/${id}/${cordoned ? 'cordon' : 'uncordon'}`, { method: 'POST' }); +} + +export type DrainResult = { services_drained: number; services_unpinned: number }; + +export async function drainAgent(id: string): Promise { + return requestJson(`/agents/${id}/drain`, { method: 'POST' }); +} + export async function getAgent(id: string): Promise { const payload = await requestJson<{ agent?: RawNodeAgent }>(`/agents/${id}`); return payload.agent ? normalizeAgent(payload.agent) : null; diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 8deccf5..ed1f77e 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -5080,6 +5080,34 @@ paths: $ref: '#/components/schemas/MessageResponse' # Node agents — management (JWT-authenticated) + /nodes/options: + get: + tags: + - Agents + summary: Node options for service placement + description: Minimal id/name/status list of registered node agents — available to any authenticated user for the service placement picker. Full agent data is admin-only under /agents. + security: + - bearerAuth: [] + responses: + '200': + description: Node options + content: + application/json: + schema: + type: object + properties: + nodes: + type: array + items: + type: object + properties: + id: + type: string + name: + type: string + status: + type: string + /agents: get: tags: @@ -5378,6 +5406,103 @@ paths: schema: $ref: '#/components/schemas/Error' + /agents/{id}/cordon: + post: + tags: + - Agents + summary: Cordon a node + description: >- + Sets `schedulable=false`: new explicit pins are rejected and `auto` + placement skips the node. Already-pinned services keep running and can + still reconcile — drain to evict them. + security: + - bearerAuth: [] + parameters: + - $ref: '#/components/parameters/AgentIdParam' + responses: + '200': + description: Node cordoned + content: + application/json: + schema: + type: object + properties: + id: { type: string } + schedulable: { type: boolean } + '404': + description: Agent not found + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + + /agents/{id}/uncordon: + post: + tags: + - Agents + summary: Uncordon a node + description: Re-opens the node for placement (`schedulable=true`). + security: + - bearerAuth: [] + parameters: + - $ref: '#/components/parameters/AgentIdParam' + responses: + '200': + description: Node uncordoned + content: + application/json: + schema: + type: object + properties: + id: { type: string } + schedulable: { type: boolean } + '404': + description: Agent not found + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + + /agents/{id}/drain: + post: + tags: + - Agents + summary: Drain a node + description: >- + Cordons the node, removes every service container via the agent + command queue, and clears the affected services' `node_id` pins — + they fall back to the local host on next deploy and can be re-pinned + elsewhere. Requires the node to be online so it can acknowledge the + removals. + security: + - bearerAuth: [] + parameters: + - $ref: '#/components/parameters/AgentIdParam' + responses: + '200': + description: Node drained + content: + application/json: + schema: + type: object + properties: + id: { type: string } + schedulable: { type: boolean } + services_drained: { type: integer } + services_unpinned: { type: integer } + '404': + description: Agent not found + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '409': + description: Node is not online + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + /agents/{id}/metrics: get: tags: @@ -6843,6 +6968,12 @@ components: sleep_idle_minutes: type: integer description: Idle timeout in minutes before sleeping (1-1440, default 15) + node_id: + type: string + description: Node agent id this service is pinned to; empty/null = local Docker host. 'auto' on create/update resolves to the least-loaded online agent. Remote nodes run registry-pulled images only (no git builds, no Traefik domains). + node_name: + type: string + description: Display name of the pinned node agent (read-only) public_url: type: string description: Computed public URL (published port or domain) @@ -6934,6 +7065,9 @@ components: static_dir: type: string description: Output directory for the static builder (default dist) + node_id: + type: string + description: Node agent id to pin to, 'auto' = least-loaded online agent; omit/empty = local host UpdateServiceRequest: type: object @@ -7020,6 +7154,9 @@ components: sleep_idle_minutes: type: integer description: Idle timeout in minutes before sleeping (1-1440) + node_id: + type: string + description: Node agent id to pin to, 'auto' = least-loaded online agent, 'local' = run on the Containr host (clears the pin) ServiceDomain: type: object @@ -8900,6 +9037,11 @@ components: auto_prune: type: boolean description: Enqueue a bounded docker prune on this node once per day. + schedulable: + type: boolean + description: >- + False on a cordoned/drained node — `auto` placement skips it and + new explicit pins are rejected. last_heartbeat: type: string format: date-time diff --git a/docs/guides/CLI.md b/docs/guides/CLI.md index 2691ac9..1e6aa9d 100644 --- a/docs/guides/CLI.md +++ b/docs/guides/CLI.md @@ -132,6 +132,30 @@ redeploy — `containr deploy ` to apply immediately. `volumes list` shows the node's docker volume inventory with in-use flags; `volumes delete ` removes an unused volume (admin scope). +## Node placement + +Pin a service to a registered node agent, or let the scheduler pick the +least-loaded online node: + +```bash +containr services create --project --name api --image ghcr.io/x/api:v1 --node +containr services update --node auto # least-loaded online node +containr services update --node local # move back to this server +``` + +Remote nodes run registry-pulled images only — git builds, domains +(Traefik lives on the host), and sleep detection stay local. Start, +stop, restart, redeploy, and delete all fan out to the pinned node's +agent. `services get` shows the resolved `node_name`. + +Take a node out of rotation or empty it entirely (admin): + +```bash +containr nodes cordon # no new pins; running replicas stay +containr nodes uncordon +containr nodes drain # cordon + remove its containers + unpin services +``` + ## Service domains and access A service can carry multiple hostnames; the default is mirrored to