feat(scheduling): multi-node service placement + remote lifecycle (#54)

* 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>
This commit is contained in:
Tomáš DvořákandDevin authored and GitHub committed 2026-10-09 16:19:12 +02:00
1 parent 401daf69aa
commit bc93c83cf9
28 files changed
+1565 -48

No files matched your search

+25 -6
View File
@@ -277,12 +277,31 @@ and wired refs, one click or one CLI call.
`nodes commands`, MCP `containr_nodes_update|prune|commands` `nodes commands`, MCP `containr_nodes_update|prune|commands`
(confirm-gated), Usage-page per-node Prune button + auto toggle. (confirm-gated), Usage-page per-node Prune button + auto toggle.
`cmd/agent/main.go`, `notification_producers.go:scheduleAutoPrune`. `cmd/agent/main.go`, `notification_producers.go:scheduleAutoPrune`.
- [ ] **Multi-node scheduling** — wire `scheduling_rules` (table exists): - [🟡] **Multi-node scheduling** — phase 1 shipped: `services.node_id`
pin/spread/affinity, prefer-lowest-usage; deploy engine routes to pin (explicit / `auto` = least-loaded online / `local` = clear),
`services.node_id` via agent commands (agent already runs containers). `NodeRunner` seam in the deploy engine routes reconcile to the agent's
- [ ] **Build-on-node** — agents currently `docker run` prebuilt images command queue (`create_container` with image/env/cmd/ports/volumes/
only; extend agent to pull registry images (build stays central) — labels/limits), `container_instances` inventory per (agent, service,
private registry creds flow through Phase C. 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/ - [ ] **IaC provisioning (ROADMAP §2.2, unchanged)** — Terraform/
OpenTofu, `infra_connections`, Proxmox + AWS modules, + Hetzner module OpenTofu, `infra_connections`, Proxmox + AWS modules, + Hetzner module
(dflow's second provider), cloud-init → auto-enroll. (dflow's second provider), cloud-init → auto-enroll.
+98 -3
View File
@@ -283,7 +283,38 @@ func createContainer(payload map[string]interface{}) (string, error) {
return "", errors.New("container name and image are required") 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 { if env, ok := container["environment"].(map[string]interface{}); ok {
for key, value := range env { for key, value := range env {
args = append(args, "-e", fmt.Sprintf("%s=%v", key, value)) 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"]) hostPort := intValue(port["host_port"])
containerPort := intValue(port["container_port"]) containerPort := intValue(port["container_port"])
protocol := firstString(stringValue(port["protocol"]), "tcp") 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)) 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) 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 { func commandTarget(payload map[string]interface{}) string {
@@ -620,3 +683,35 @@ func boolValue(value interface{}) bool {
typed, _ := value.(bool) typed, _ := value.(bool)
return typed 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
}
}
+152 -9
View File
@@ -11,6 +11,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"log"
"math" "math"
"net/http" "net/http"
"os" "os"
@@ -39,6 +40,7 @@ type NodeAgent struct {
Resources NodeResources `json:"resources"` Resources NodeResources `json:"resources"`
LastHeartbeat time.Time `json:"last_heartbeat"` LastHeartbeat time.Time `json:"last_heartbeat"`
AutoPrune bool `json:"auto_prune"` AutoPrune bool `json:"auto_prune"`
Schedulable bool `json:"schedulable"`
CreatedAt time.Time `json:"created_at"` CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"` UpdatedAt time.Time `json:"updated_at"`
Metadata map[string]interface{} `json:"metadata"` 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 { func agentFromRow(row sqlcdb.NodeAgent) NodeAgent {
agent := NodeAgent{ agent := NodeAgent{
ID: row.ID, ID: row.ID,
Name: row.Name, Name: row.Name,
Hostname: row.Hostname, Hostname: row.Hostname,
IPAddress: row.IpAddress, IPAddress: row.IpAddress,
Port: int(row.Port), Port: int(row.Port),
Status: row.Status.String, Status: row.Status.String,
Version: row.Version.String, Version: row.Version.String,
AutoPrune: row.AutoPrune, AutoPrune: row.AutoPrune,
Metadata: map[string]interface{}{}, Schedulable: row.Schedulable,
Metadata: map[string]interface{}{},
} }
if row.LastHeartbeat.Valid { if row.LastHeartbeat.Valid {
agent.LastHeartbeat = row.LastHeartbeat.Time agent.LastHeartbeat = row.LastHeartbeat.Time
@@ -380,6 +403,8 @@ func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) {
Resources: existing.Resources, Resources: existing.Resources,
LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true}, LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true},
Metadata: rawJSON(metadata), Metadata: rawJSON(metadata),
AutoPrune: existing.AutoPrune,
Schedulable: existing.Schedulable,
}) })
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent"}) 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()}, LastHeartbeat: sql.NullTime{Time: agent.LastHeartbeat, Valid: !agent.LastHeartbeat.IsZero()},
Metadata: rawJSON(agent.Metadata), Metadata: rawJSON(agent.Metadata),
AutoPrune: agent.AutoPrune, AutoPrune: agent.AutoPrune,
Schedulable: agent.Schedulable,
}) })
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent"}) 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("resources", &agent.Resources)
remarshal("metadata", &agent.Metadata) remarshal("metadata", &agent.Metadata)
remarshal("auto_prune", &agent.AutoPrune) remarshal("auto_prune", &agent.AutoPrune)
remarshal("schedulable", &agent.Schedulable)
} }
// DeleteAgent removes an agent // DeleteAgent removes an agent
@@ -673,9 +700,40 @@ func (h *NodeAgentHandler) CompleteCommand(c *gin.Context) {
h.updateContainerStatusAfterCommand(ctx, out) 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}) 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) { func (h *NodeAgentHandler) updateContainerStatusAfterCommand(ctx context.Context, command AgentCommand) {
row, err := h.q.GetContainer(ctx, *command.ContainerID) row, err := h.q.GetContainer(ctx, *command.ContainerID)
if err != nil { if err != nil {
@@ -1329,9 +1387,94 @@ func (h *NodeAgentHandler) SetupAdminRoutes(router *gin.RouterGroup) {
agents.DELETE("/:id/containers/:containerId", h.ContainerAction) agents.DELETE("/:id/containers/:containerId", h.ContainerAction)
agents.GET("/:id/metrics", h.GetAgentMetrics) 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/prune", h.PruneAgent)
agents.POST("/:id/commands", h.ExecuteCommand) agents.POST("/:id/commands", h.ExecuteCommand)
agents.GET("/:id/commands", h.GetAgentCommands) agents.GET("/:id/commands", h.GetAgentCommands)
agents.GET("/:id/commands/:commandId", h.GetCommandStatus) 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})
}
+15
View File
@@ -421,6 +421,20 @@ func runDeploymentAndSyncWithImage(
maintURL = maintenanceURL(db) 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{ deployReq := &deployment.DeploymentRequest{
ProjectID: service.ProjectID.String(), ProjectID: service.ProjectID.String(),
ServiceID: service.ID.String(), ServiceID: service.ID.String(),
@@ -447,6 +461,7 @@ func runDeploymentAndSyncWithImage(
Maintenance: maintenanceMode, Maintenance: maintenanceMode,
MaintenanceURL: maintURL, MaintenanceURL: maintURL,
BasicAuthUsers: basicAuthUsers, BasicAuthUsers: basicAuthUsers,
NodeID: nodeID,
}, },
Trigger: deployment.TriggerConfig{ Trigger: deployment.TriggerConfig{
Type: req.Trigger, Type: req.Trigger,
+304
View File
@@ -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
}
+4
View File
@@ -43,6 +43,9 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg
dockerClient = client dockerClient = client
buildManager = build.NewBuildManager("/tmp/containr-builds", dockerClient) buildManager = build.NewBuildManager("/tmp/containr-builds", dockerClient)
deploymentEngine = deployment.NewDeploymentEngine(buildManager, 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. // Restore managed sidecars (e.g. cloudflared) from saved settings.
syncManagedContainers(dockerClient, db) 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 // Cross-cutting ops view — active/queued/failed work
authed.GET("/operations", handleGetOperations) authed.GET("/operations", handleGetOperations)
authed.GET("/nodes/options", agentHandler.ListNodeOptions)
authed.GET("/activity", handleGetActivity) authed.GET("/activity", handleGetActivity)
authed.GET("/projects/:id/activity", handleGetProjectActivity) authed.GET("/projects/:id/activity", handleGetProjectActivity)
+36 -7
View File
@@ -18,6 +18,38 @@ import (
"github.com/google/uuid" "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 // serviceRuntimeSpec builds the container spec for a service: stored env vars
// merged with ${{Service.KEY}} references resolved across the project. // merged with ${{Service.KEY}} references resolved across the project.
func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpec, error) { 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, Domain: service.Domain,
HealthPath: service.HealthCheckPath, HealthPath: service.HealthCheckPath,
RestartPolicy: service.RestartPolicy, RestartPolicy: service.RestartPolicy,
NodeID: serviceNodeID(db, service.ID),
} }
// Best effort: reuse the host port from the last live deployment so the // Best effort: reuse the host port from the last live deployment so the
// public URL survives restarts. Column exists post-migration; older DBs // public URL survives restarts. Column exists post-migration; older DBs
@@ -384,7 +417,7 @@ func handleServiceStart(c *gin.Context) {
if !ok { if !ok {
return 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()}) c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
return return
} }
@@ -402,7 +435,7 @@ func handleServiceStop(c *gin.Context) {
if !ok { if !ok {
return 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()}) c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
return return
} }
@@ -421,11 +454,7 @@ func handleServiceRestart(c *gin.Context) {
return return
} }
ctx := c.Request.Context() ctx := c.Request.Context()
if err := engine.StopService(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
}
if err := engine.StartService(ctx, service.ID.String()); err != nil {
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
return return
} }
+67 -7
View File
@@ -5,6 +5,7 @@ import (
"containr/internal/deployment" "containr/internal/deployment"
"containr/internal/deployqueue" "containr/internal/deployqueue"
"context" "context"
"database/sql"
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
@@ -48,8 +49,12 @@ type Service struct {
StaticDir string `json:"static_dir,omitempty" db:"static_dir"` StaticDir string `json:"static_dir,omitempty" db:"static_dir"`
SleepEnabled bool `json:"sleep_enabled" db:"-"` SleepEnabled bool `json:"sleep_enabled" db:"-"`
SleepIdleMinutes int `json:"sleep_idle_minutes" db:"-"` SleepIdleMinutes int `json:"sleep_idle_minutes" db:"-"`
CreatedAt time.Time `json:"created_at" db:"created_at"` // NodeID pins the service to a node agent ("" = local Docker host,
UpdatedAt time.Time `json:"updated_at" db:"updated_at"` // 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 // 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. // ServiceVolume is a volume/bind mount applied to every replica.
type ServiceVolume struct { type ServiceVolume struct {
Type string `json:"type"` // volume | bind (empty = volume) Type string `json:"type"` // volume | bind (empty = volume)
@@ -141,6 +168,9 @@ type CreateServiceRequest struct {
MemoryReserve string `json:"memory_reserve"` MemoryReserve string `json:"memory_reserve"`
StaticBuildCmd string `json:"static_build_cmd"` StaticBuildCmd string `json:"static_build_cmd"`
StaticDir string `json:"static_dir"` 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 // UpdateServiceRequest represents a request to update a service
@@ -170,6 +200,8 @@ type UpdateServiceRequest struct {
StaticDir *string `json:"static_dir"` StaticDir *string `json:"static_dir"`
SleepEnabled *bool `json:"sleep_enabled"` SleepEnabled *bool `json:"sleep_enabled"`
SleepIdleMinutes *int `json:"sleep_idle_minutes"` 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 // 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). // Reconcile stored status with live container state (no-op without Docker).
for i := range services { for i := range services {
liveServiceStatus(c, db.(*database.DB), &services[i]) liveServiceStatus(c, db.(*database.DB), &services[i])
loadServiceNode(db.(*database.DB), &services[i])
} }
c.JSON(http.StatusOK, gin.H{"services": services}) c.JSON(http.StatusOK, gin.H{"services": services})
@@ -409,6 +442,15 @@ func handleCreateService(c *gin.Context) {
sourceType := inferServiceSourceType(service) 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 // Insert service into database
_, err = db.(*database.DB).Exec( _, err = db.(*database.DB).Exec(
`INSERT INTO services `INSERT INTO services
@@ -417,16 +459,18 @@ func handleCreateService(c *gin.Context) {
git_repo, git_branch, build_path, cpu, memory, replicas, port, domain, git_repo, git_branch, build_path, cpu, memory, replicas, port, domain,
healthcheck_path, restart_policy, volumes, healthcheck_path, restart_policy, volumes,
builder, cpu_reserve, memory_reserve, static_build_cmd, static_dir, builder, cpu_reserve, memory_reserve, static_build_cmd, static_dir,
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, 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, service.ID, service.ProjectID, service.Name, environmentID, service.Type,
sourceType, firstNonEmpty(service.GitRepo, service.Image), service.Image, sourceType, firstNonEmpty(service.GitRepo, service.Image), service.Image,
"", service.Command, service.Type, service.Status, service.Image, service.Command, "", service.Command, service.Type, service.Status, service.Image, service.Command,
service.Environment, service.GitRepo, service.GitBranch, service.BuildPath, service.CPU, service.Memory, service.Environment, service.GitRepo, service.GitBranch, service.BuildPath, service.CPU, service.Memory,
service.Replicas, service.Port, service.Domain, service.HealthCheckPath, service.RestartPolicy, service.Replicas, service.Port, service.Domain, service.HealthCheckPath, service.RestartPolicy,
volumesJSON, service.Builder, service.CPUReserve, service.MemoryReserve, 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 { if err != nil {
@@ -456,7 +500,9 @@ func handleCreateService(c *gin.Context) {
LogAuditWithRequest(c, "service", service.ID.String(), "create", map[string]interface{}{ LogAuditWithRequest(c, "service", service.ID.String(), "create", map[string]interface{}{
"name": service.Name, "type": service.Type, "project_id": service.ProjectID, "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}) 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) service.Domains = loadServiceDomains(db.(*database.DB), service.ID)
loadServiceSleep(db.(*database.DB), &service) loadServiceSleep(db.(*database.DB), &service)
loadServiceNode(db.(*database.DB), &service)
maintenance, basicAuth := serviceAccess(db.(*database.DB), service.ID) maintenance, basicAuth := serviceAccess(db.(*database.DB), service.ID)
service.MaintenanceMode = maintenance service.MaintenanceMode = maintenance
service.BasicAuth = basicAuthUsernames(basicAuth) service.BasicAuth = basicAuthUsernames(basicAuth)
@@ -725,6 +772,17 @@ func handleUpdateService(c *gin.Context) {
volumesArg = volumesJSON 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() existingService.UpdatedAt = time.Now()
// Update service in database // Update service in database
@@ -736,7 +794,7 @@ func handleUpdateService(c *gin.Context) {
restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16, restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16,
builder = $19, cpu_reserve = $20, memory_reserve = $21, builder = $19, cpu_reserve = $20, memory_reserve = $21,
static_build_cmd = $22, static_dir = $23, static_build_cmd = $22, static_dir = $23,
sleep_enabled = $24, sleep_idle_minutes = $25 sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26
WHERE id = $18`, WHERE id = $18`,
existingService.Name, existingService.Type, existingService.Image, existingService.Command, existingService.Name, existingService.Type, existingService.Image, existingService.Command,
existingService.Environment, existingService.GitRepo, existingService.GitBranch, existingService.Environment, existingService.GitRepo, existingService.GitBranch,
@@ -747,6 +805,7 @@ func handleUpdateService(c *gin.Context) {
existingService.Builder, existingService.CPUReserve, existingService.MemoryReserve, existingService.Builder, existingService.CPUReserve, existingService.MemoryReserve,
existingService.StaticBuildCmd, existingService.StaticDir, existingService.StaticBuildCmd, existingService.StaticDir,
existingService.SleepEnabled, existingService.SleepIdleMinutes, existingService.SleepEnabled, existingService.SleepIdleMinutes,
sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""},
) )
if err != nil { 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}) c.JSON(http.StatusOK, gin.H{"service": existingService})
} }
@@ -865,7 +925,7 @@ func handleDeleteService(c *gin.Context) {
// Remove runtime containers before dropping the row. // Remove runtime containers before dropping the row.
if engineValue, exists := c.Get("deployment_engine"); exists && engineValue != nil { if engineValue, exists := c.Get("deployment_engine"); exists && engineValue != nil {
if engine, ok := engineValue.(*deployment.DeploymentEngine); ok { if engine, ok := engineValue.(*deployment.DeploymentEngine); ok {
_ = engine.RemoveServiceContainers(c.Request.Context(), serviceID.String()) removeServiceRuntime(c.Request.Context(), db.(*database.DB), engine, serviceID)
} }
} }
+1 -1
View File
@@ -637,7 +637,7 @@ func deployTemplateGraph(
cleanup := func() { cleanup := func() {
for _, sid := range createdServiceIDs { for _, sid := range createdServiceIDs {
if engine != nil { 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 environment_variables WHERE service_id = $1`, sid)
_, _ = db.Exec(`DELETE FROM services WHERE id = $1`, sid) _, _ = db.Exec(`DELETE FROM services WHERE id = $1`, sid)
+58 -1
View File
@@ -133,6 +133,63 @@ var nodesPruneCmd = &cobra.Command{
}, },
} }
var nodesCordonCmd = &cobra.Command{
Use: "cordon <id>",
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 <id>",
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 <id>",
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{ var nodesCommandsCmd = &cobra.Command{
Use: "commands <id>", Use: "commands <id>",
Short: "List recent commands sent to a node agent", Short: "List recent commands sent to a node agent",
@@ -220,5 +277,5 @@ func init() {
nodesPruneCmd.Flags().String("until", "", "only prune objects older than this duration (e.g. 168h)") nodesPruneCmd.Flags().String("until", "", "only prune objects older than this duration (e.g. 168h)")
nodesPruneCmd.Flags().Bool("volumes", false, "also prune volumes (destructive — asks twice)") nodesPruneCmd.Flags().Bool("volumes", false, "also prune volumes (destructive — asks twice)")
nodeTokensCmd.AddCommand(nodeTokensIssueCmd, nodeTokensListCmd, nodeTokensRevokeCmd) nodeTokensCmd.AddCommand(nodeTokensIssueCmd, nodeTokensListCmd, nodeTokensRevokeCmd)
NodesCmd.AddCommand(nodesListCmd, nodesGetCmd, nodesUpdateCmd, nodesPruneCmd, nodesCommandsCmd, nodesDeleteCmd, nodeTokensCmd) NodesCmd.AddCommand(nodesListCmd, nodesGetCmd, nodesUpdateCmd, nodesCordonCmd, nodesUncordonCmd, nodesDrainCmd, nodesPruneCmd, nodesCommandsCmd, nodesDeleteCmd, nodeTokensCmd)
} }
@@ -60,7 +60,7 @@ var servicesGetCmd = &cobra.Command{
return nil return nil
} }
s, _ := unwrapObject(data, "service") 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)) fmt.Printf("%-14s %s\n", k+":", str(s, k))
} }
return nil return nil
@@ -79,7 +79,7 @@ var servicesCreateCmd = &cobra.Command{
body := map[string]interface{}{ body := map[string]interface{}{
"name": args[1], "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 != "" { if v, _ := cmd.Flags().GetString(f); v != "" {
key := f key := f
// JSON uses snake_case keys matching the API schema. // JSON uses snake_case keys matching the API schema.
@@ -102,6 +102,8 @@ var servicesCreateCmd = &cobra.Command{
key = "static_build_cmd" key = "static_build_cmd"
case "static-dir": case "static-dir":
key = "static_dir" key = "static_dir"
case "node":
key = "node_id"
} }
body[key] = v body[key] = v
} }
@@ -159,7 +161,7 @@ var servicesUpdateCmd = &cobra.Command{
return err return err
} }
body := map[string]interface{}{} 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) { if cmd.Flags().Changed(f) {
key := f key := f
switch f { switch f {
@@ -175,6 +177,8 @@ var servicesUpdateCmd = &cobra.Command{
key = "static_build_cmd" key = "static_build_cmd"
case "static-dir": case "static-dir":
key = "static_dir" key = "static_dir"
case "node":
key = "node_id"
} }
v, _ := cmd.Flags().GetString(f) v, _ := cmd.Flags().GetString(f)
body[key] = v body[key] = v
@@ -490,6 +494,7 @@ func init() {
f.String("memory-reserve", "", "soft memory reservation (e.g. 128Mi)") f.String("memory-reserve", "", "soft memory reservation (e.g. 128Mi)")
f.String("static-cmd", "", "build command for the static builder") f.String("static-cmd", "", "build command for the static builder")
f.String("static-dir", "", "output dir for the static builder (default dist)") f.String("static-dir", "", "output dir for the static builder (default dist)")
f.String("node", "", "pin to a node agent id, or 'auto' for least-loaded (image services only)")
uf := servicesUpdateCmd.Flags() uf := servicesUpdateCmd.Flags()
uf.String("name", "", "service name") uf.String("name", "", "service name")
@@ -514,6 +519,7 @@ func init() {
uf.String("static-dir", "", "output dir for the static builder (empty clears)") 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.Bool("sleep", false, "enable scale-to-zero on idle (sleep mode)")
uf.Int("sleep-idle", 0, "idle minutes before sleeping (1-1440)") uf.Int("sleep-idle", 0, "idle minutes before sleeping (1-1440)")
uf.String("node", "", "pin to a node agent id, 'auto' for least-loaded, 'local' to clear")
for _, a := range []string{"start", "stop", "restart", "redeploy", "sleep", "wake"} { for _, a := range []string{"start", "stop", "restart", "redeploy", "sleep", "wake"} {
verb := a verb := a
@@ -14,6 +14,18 @@ import (
"github.com/sqlc-dev/pqtype" "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 const completeCommand = `-- name: CompleteCommand :one
UPDATE agent_commands SET UPDATE agent_commands SET
status = $3, status = $3,
@@ -63,7 +75,7 @@ INSERT INTO node_agents (
id, name, hostname, ip_address, port, status, version, id, name, hostname, ip_address, port, status, version,
capabilities, resources, last_heartbeat, metadata capabilities, resources, last_heartbeat, metadata
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ) 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 { type CreateAgentParams struct {
@@ -110,6 +122,7 @@ func (q *Queries) CreateAgent(ctx context.Context, arg CreateAgentParams) (NodeA
&i.UpdatedAt, &i.UpdatedAt,
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable,
) )
return i, err return i, err
} }
@@ -229,8 +242,22 @@ func (q *Queries) DeleteAgent(ctx context.Context, id string) error {
return err 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 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) { 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.UpdatedAt,
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable,
) )
return i, err return i, err
} }
const getAgentByHostAndIP = `-- name: GetAgentByHostAndIP :one 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 WHERE hostname = $1 AND ip_address = $2
LIMIT 1 LIMIT 1
` `
@@ -284,6 +312,7 @@ func (q *Queries) GetAgentByHostAndIP(ctx context.Context, arg GetAgentByHostAnd
&i.UpdatedAt, &i.UpdatedAt,
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable,
) )
return i, err return i, err
} }
@@ -491,7 +520,7 @@ func (q *Queries) ListAgentHeartbeatsSince(ctx context.Context, arg ListAgentHea
} }
const listAgents = `-- name: ListAgents :many 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) { 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.UpdatedAt,
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@@ -661,6 +691,53 @@ func (q *Queries) ListPendingCommands(ctx context.Context, nodeAgentID string) (
return items, nil 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 const updateAgent = `-- name: UpdateAgent :one
UPDATE node_agents SET UPDATE node_agents SET
name = $2, name = $2,
@@ -674,9 +751,10 @@ UPDATE node_agents SET
last_heartbeat = $10, last_heartbeat = $10,
metadata = $11, metadata = $11,
auto_prune = $12, auto_prune = $12,
schedulable = $13,
updated_at = NOW() updated_at = NOW()
WHERE id = $1 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 { type UpdateAgentParams struct {
@@ -692,6 +770,7 @@ type UpdateAgentParams struct {
LastHeartbeat sql.NullTime `json:"last_heartbeat"` LastHeartbeat sql.NullTime `json:"last_heartbeat"`
Metadata pqtype.NullRawMessage `json:"metadata"` Metadata pqtype.NullRawMessage `json:"metadata"`
AutoPrune bool `json:"auto_prune"` AutoPrune bool `json:"auto_prune"`
Schedulable bool `json:"schedulable"`
} }
func (q *Queries) UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeAgent, error) { 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.LastHeartbeat,
arg.Metadata, arg.Metadata,
arg.AutoPrune, arg.AutoPrune,
arg.Schedulable,
) )
var i NodeAgent var i NodeAgent
err := row.Scan( err := row.Scan(
@@ -725,6 +805,7 @@ func (q *Queries) UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeA
&i.UpdatedAt, &i.UpdatedAt,
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable,
) )
return i, err 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) _, err := q.db.ExecContext(ctx, updateContainerStatus, arg.ID, arg.Status)
return err 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
}
@@ -176,6 +176,7 @@ type NodeAgent struct {
UpdatedAt sql.NullTime `json:"updated_at"` UpdatedAt sql.NullTime `json:"updated_at"`
Metadata pqtype.NullRawMessage `json:"metadata"` Metadata pqtype.NullRawMessage `json:"metadata"`
AutoPrune bool `json:"auto_prune"` AutoPrune bool `json:"auto_prune"`
Schedulable bool `json:"schedulable"`
} }
type Notification struct { type Notification struct {
@@ -274,6 +275,7 @@ type Service struct {
MemoryReserve string `json:"memory_reserve"` MemoryReserve string `json:"memory_reserve"`
StaticBuildCmd string `json:"static_build_cmd"` StaticBuildCmd string `json:"static_build_cmd"`
StaticDir string `json:"static_dir"` StaticDir string `json:"static_dir"`
NodeID sql.NullString `json:"node_id"`
} }
type ServiceDomain struct { type ServiceDomain struct {
@@ -12,6 +12,7 @@ import (
) )
type Querier interface { type Querier interface {
ClearServiceNodePins(ctx context.Context, nodeID sql.NullString) (int64, error)
CompleteCommand(ctx context.Context, arg CompleteCommandParams) (AgentCommand, error) CompleteCommand(ctx context.Context, arg CompleteCommandParams) (AgentCommand, error)
CountDatabaseServicesByUserAndName(ctx context.Context, arg CountDatabaseServicesByUserAndNameParams) (int64, error) CountDatabaseServicesByUserAndName(ctx context.Context, arg CountDatabaseServicesByUserAndNameParams) (int64, error)
CountDatabasesUsingBackupTarget(ctx context.Context, targetID sql.NullString) (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 DeleteDatabaseServiceByIDAndUser(ctx context.Context, arg DeleteDatabaseServiceByIDAndUserParams) error
DeleteOutboundWebhookByIDAndUser(ctx context.Context, arg DeleteOutboundWebhookByIDAndUserParams) error DeleteOutboundWebhookByIDAndUser(ctx context.Context, arg DeleteOutboundWebhookByIDAndUserParams) error
DeleteProjectByID(ctx context.Context, projectID uuid.UUID) (int64, 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 DeleteUserInvite(ctx context.Context, id uuid.UUID) error
DeleteUserTemplate(ctx context.Context, arg DeleteUserTemplateParams) (int64, error) DeleteUserTemplate(ctx context.Context, arg DeleteUserTemplateParams) (int64, error)
GetActiveAgentAuthTokenByHash(ctx context.Context, tokenHash string) (AgentAuthToken, error) GetActiveAgentAuthTokenByHash(ctx context.Context, tokenHash string) (AgentAuthToken, error)
@@ -100,8 +102,11 @@ type Querier interface {
MarkAllNotificationsReadByUser(ctx context.Context, arg MarkAllNotificationsReadByUserParams) error MarkAllNotificationsReadByUser(ctx context.Context, arg MarkAllNotificationsReadByUserParams) error
MarkNotificationReadByIDAndUser(ctx context.Context, arg MarkNotificationReadByIDAndUserParams) error MarkNotificationReadByIDAndUser(ctx context.Context, arg MarkNotificationReadByIDAndUserParams) error
MarkUserInviteUsed(ctx context.Context, arg MarkUserInviteUsedParams) error MarkUserInviteUsed(ctx context.Context, arg MarkUserInviteUsedParams) error
PickLeastLoadedAgent(ctx context.Context) (string, error)
RevokeAgentAuthToken(ctx context.Context, id uuid.UUID) (AgentAuthToken, error) RevokeAgentAuthToken(ctx context.Context, id uuid.UUID) (AgentAuthToken, error)
RevokeUserToken(ctx context.Context, arg RevokeUserTokenParams) (int64, error) RevokeUserToken(ctx context.Context, arg RevokeUserTokenParams) (int64, error)
ScrubCommandPayload(ctx context.Context, arg ScrubCommandPayloadParams) error
SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulableParams) error
SetDatabaseBackupRemoteKeyByID(ctx context.Context, arg SetDatabaseBackupRemoteKeyByIDParams) error SetDatabaseBackupRemoteKeyByID(ctx context.Context, arg SetDatabaseBackupRemoteKeyByIDParams) error
SetDatabaseBackupScheduleByIDAndUser(ctx context.Context, arg SetDatabaseBackupScheduleByIDAndUserParams) error SetDatabaseBackupScheduleByIDAndUser(ctx context.Context, arg SetDatabaseBackupScheduleByIDAndUserParams) error
SetDatabaseBackupStatusByID(ctx context.Context, arg SetDatabaseBackupStatusByIDParams) error SetDatabaseBackupStatusByID(ctx context.Context, arg SetDatabaseBackupStatusByIDParams) error
@@ -127,6 +132,7 @@ type Querier interface {
UpdateUserTemplate(ctx context.Context, arg UpdateUserTemplateParams) (int64, error) UpdateUserTemplate(ctx context.Context, arg UpdateUserTemplateParams) (int64, error)
UpdateWebhookDeliveryResult(ctx context.Context, arg UpdateWebhookDeliveryResultParams) error UpdateWebhookDeliveryResult(ctx context.Context, arg UpdateWebhookDeliveryResultParams) error
UpsertEnvironmentVariable(ctx context.Context, arg UpsertEnvironmentVariableParams) error UpsertEnvironmentVariable(ctx context.Context, arg UpsertEnvironmentVariableParams) error
UpsertServiceContainer(ctx context.Context, arg UpsertServiceContainerParams) error
UpsertServiceTemplate(ctx context.Context, arg UpsertServiceTemplateParams) error UpsertServiceTemplate(ctx context.Context, arg UpsertServiceTemplateParams) error
} }
+24
View File
@@ -11,10 +11,24 @@ import (
"containr/internal/types" "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 { type DeploymentEngine struct {
buildManager *build.BuildManager buildManager *build.BuildManager
dockerClient *docker.Client dockerClient *docker.Client
scheduler *Scheduler scheduler *Scheduler
nodeRunner NodeRunner
deployments map[string]*Deployment deployments map[string]*Deployment
deploymentLog chan *DeploymentEvent deploymentLog chan *DeploymentEvent
} }
@@ -67,6 +81,10 @@ type ServiceConfig struct {
// PublishedPort is the host port bound by the previous deployment; replica 0 // PublishedPort is the host port bound by the previous deployment; replica 0
// reuses it so public URLs survive restarts. 0 = pick ephemeral. // reuses it so public URLs survive restarts. 0 = pick ephemeral.
PublishedPort int32 `json:"published_port,omitempty"` 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 { type PortMapping struct {
@@ -177,6 +195,11 @@ func (de *DeploymentEngine) DockerClient() *docker.Client {
return de.dockerClient 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 { func NewDeploymentEngine(buildManager *build.BuildManager, dockerClient *docker.Client) *DeploymentEngine {
return &DeploymentEngine{ return &DeploymentEngine{
buildManager: buildManager, buildManager: buildManager,
@@ -352,6 +375,7 @@ func (de *DeploymentEngine) deployService(ctx context.Context, deployment *Deplo
Maintenance: cfg.Maintenance, Maintenance: cfg.Maintenance,
MaintenanceURL: cfg.MaintenanceURL, MaintenanceURL: cfg.MaintenanceURL,
BasicAuthUsers: cfg.BasicAuthUsers, BasicAuthUsers: cfg.BasicAuthUsers,
NodeID: cfg.NodeID,
} }
for _, pm := range cfg.PortMappings { for _, pm := range cfg.PortMappings {
if spec.Port == 0 { if spec.Port == 0 {
+29 -1
View File
@@ -55,6 +55,8 @@ type RuntimeSpec struct {
Maintenance bool // redirect traffic to MaintenanceURL Maintenance bool // redirect traffic to MaintenanceURL
MaintenanceURL string // absolute URL; empty = serve empty response MaintenanceURL string // absolute URL; empty = serve empty response
BasicAuthUsers string // htpasswd-format user:hash pairs, comma-separated BasicAuthUsers string // htpasswd-format user:hash pairs, comma-separated
// NodeID pins the service to a node agent; "" runs on the local host.
NodeID string
} }
// RuntimeContainer describes one live replica. // 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 // ReconcileService brings live containers in line with the spec. When the
// spec hash changed (image/env/ports/etc.) all replicas are recreated; // 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) { 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 { if spec.Replicas < 1 {
spec.Replicas = 1 spec.Replicas = 1
} }
@@ -498,6 +507,25 @@ func (de *DeploymentEngine) RemoveServiceContainers(ctx context.Context, service
return nil 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. // RemoveProjectContainers deletes every managed container in a project.
func (de *DeploymentEngine) RemoveProjectContainers(ctx context.Context, projectID string) error { func (de *DeploymentEngine) RemoveProjectContainers(ctx context.Context, projectID string) error {
containers, err := de.dockerClient.ListContainersFiltered(ctx, filters.NewArgs( containers, err := de.dockerClient.ListContainersFiltered(ctx, filters.NewArgs(
+5 -2
View File
@@ -53,7 +53,7 @@ func init() {
// Services // Services
register(spec{"containr_services_list", "List services in a project", "GET", "/projects/{project_id}/services", []string{"project_id"}, nil, false}) register(spec{"containr_services_list", "List services in a project", "GET", "/projects/{project_id}/services", []string{"project_id"}, nil, false})
register(spec{"containr_services_get", "Get a service by id", "GET", "/services/{id}", []string{"id"}, nil, false}) register(spec{"containr_services_get", "Get a service by id", "GET", "/services/{id}", []string{"id"}, nil, false})
register(spec{"containr_services_create", "Create a service in a project. Args: name (required), type, image, git_repo, git_branch, environment, port, domain, replicas, volumes = [{type: volume|bind, source, target, read_only}], builder = auto|railpack|nixpacks|dockerfile|static, cpu, memory, cpu_reserve, memory_reserve, static_build_cmd, static_dir.", "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_delete", "Delete a service. Requires confirm:true.", "DELETE", "/services/{id}", []string{"id"}, nil, true})
register(spec{"containr_services_start", "Start a service", "POST", "/services/{id}/start", []string{"id"}, nil, false}) register(spec{"containr_services_start", "Start a service", "POST", "/services/{id}/start", []string{"id"}, nil, false})
register(spec{"containr_services_stop", "Stop a service", "POST", "/services/{id}/stop", []string{"id"}, nil, false}) register(spec{"containr_services_stop", "Stop a service", "POST", "/services/{id}/stop", []string{"id"}, nil, false})
@@ -66,7 +66,7 @@ func init() {
register(spec{"containr_services_move", "Move a service to another project the caller owns. Args: id, project_id (required).", "POST", "/services/{id}/move", []string{"id"}, []string{"project_id"}, false}) register(spec{"containr_services_move", "Move a service to another project the caller owns. Args: id, project_id (required).", "POST", "/services/{id}/move", []string{"id"}, []string{"project_id"}, false})
register(spec{"containr_services_logs", "Get service runtime logs. Args: id, tail (default 100).", "GET", "/services/{id}/logs", []string{"id"}, nil, false}) register(spec{"containr_services_logs", "Get service runtime logs. Args: id, tail (default 100).", "GET", "/services/{id}/logs", []string{"id"}, nil, false})
register(spec{"containr_services_exec", "Run a one-off command in the service container (30s ceiling). Args: id, command (required).", "POST", "/services/{id}/exec", []string{"id"}, []string{"command"}, false}) register(spec{"containr_services_exec", "Run a one-off command in the service container (30s ceiling). Args: id, command (required).", "POST", "/services/{id}/exec", []string{"id"}, []string{"command"}, false})
register(spec{"containr_services_update", "Update a service. Args: id, name, image, command, domain, restart_policy, healthcheck_path, cpu, memory, replicas, port, volumes = [{type: volume|bind, source, target, read_only}] — volumes replaces the whole mount list; pass [] to clear. maintenance_mode (bool), basic_auth = [{username, password}] — replaces all creds; [] clears. builder = auto|railpack|nixpacks|dockerfile|static, cpu_reserve, memory_reserve, static_build_cmd, static_dir (empty string clears reserves/static fields). sleep_enabled (bool) + sleep_idle_minutes (1-1440) enable scale-to-zero on idle; a wake placeholder answers traffic while the service resumes.", "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 // Domains
register(spec{"containr_domains_list", "List domains attached to a service", "GET", "/services/{id}/domains", []string{"id"}, nil, false}) register(spec{"containr_domains_list", "List domains attached to a service", "GET", "/services/{id}/domains", []string{"id"}, nil, false})
@@ -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_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_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_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_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}) register(spec{"containr_nodes_commands", "List recent commands sent to a node agent (admin). Args: id.", "GET", "/agents/{id}/commands", []string{"id"}, nil, false})
@@ -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;
@@ -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;
+35
View File
@@ -29,6 +29,7 @@ UPDATE node_agents SET
last_heartbeat = $10, last_heartbeat = $10,
metadata = $11, metadata = $11,
auto_prune = $12, auto_prune = $12,
schedulable = $13,
updated_at = NOW() updated_at = NOW()
WHERE id = $1 WHERE id = $1
RETURNING *; RETURNING *;
@@ -113,3 +114,37 @@ WHERE id = $1;
SELECT * FROM agent_commands SELECT * FROM agent_commands
WHERE node_agent_id = $1 AND type = $2 WHERE node_agent_id = $1 AND type = $2
ORDER BY created_at DESC LIMIT 1; ORDER BY created_at DESC LIMIT 1;
-- name: PickLeastLoadedAgent :one
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;
+4 -2
View File
@@ -66,7 +66,8 @@ CREATE TABLE services (
cpu_reserve VARCHAR(20) NOT NULL DEFAULT '', cpu_reserve VARCHAR(20) NOT NULL DEFAULT '',
memory_reserve VARCHAR(20) NOT NULL DEFAULT '', memory_reserve VARCHAR(20) NOT NULL DEFAULT '',
static_build_cmd VARCHAR(255) NOT NULL DEFAULT '', static_build_cmd VARCHAR(255) NOT NULL DEFAULT '',
static_dir VARCHAR(255) NOT NULL DEFAULT '' static_dir VARCHAR(255) NOT NULL DEFAULT '',
node_id VARCHAR(255)
); );
CREATE TABLE registries ( CREATE TABLE registries (
@@ -193,7 +194,8 @@ CREATE TABLE node_agents (
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
metadata JSONB, 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 ( CREATE TABLE container_instances (
@@ -2,7 +2,9 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
import { useMemo, useState } from 'react'; import { useMemo, useState } from 'react';
import { Link, useNavigate, useParams } from 'react-router-dom'; import { Link, useNavigate, useParams } from 'react-router-dom';
import { import {
cordonAgent,
deleteAgent, deleteAgent,
drainAgent,
getAgent, getAgent,
getAgentMetrics, getAgentMetrics,
listAgentCommands, listAgentCommands,
@@ -43,6 +45,7 @@ export function NodeDetailPage() {
const [rename, setRename] = useState<string | null>(null); const [rename, setRename] = useState<string | null>(null);
const [confirmDelete, setConfirmDelete] = useState(false); const [confirmDelete, setConfirmDelete] = useState(false);
const [confirmVolumes, setConfirmVolumes] = useState(false); const [confirmVolumes, setConfirmVolumes] = useState(false);
const [confirmDrain, setConfirmDrain] = useState(false);
const agentQuery = useQuery({ const agentQuery = useQuery({
queryKey: ['node-agent', id], queryKey: ['node-agent', id],
@@ -86,6 +89,14 @@ export function NodeDetailPage() {
mutationFn: () => deleteAgent(id), mutationFn: () => deleteAgent(id),
onSuccess: () => navigate('/usage'), 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 agent = agentQuery.data;
const metrics = useMemo(() => metricsQuery.data ?? [], [metricsQuery.data]); const metrics = useMemo(() => metricsQuery.data ?? [], [metricsQuery.data]);
@@ -168,6 +179,11 @@ export function NodeDetailPage() {
: 'border-[var(--error)]/40 text-[var(--error)]'}`}> : 'border-[var(--error)]/40 text-[var(--error)]'}`}>
{agent.status} {agent.status}
</span> </span>
{!agent.schedulable && (
<span className="v-mono text-[11px] px-2 py-1 rounded-md border border-[var(--warning)]/40 text-[var(--warning)]">
cordoned
</span>
)}
{isAdmin && ( {isAdmin && (
<> <>
<label className="flex items-center gap-1.5 v-mono text-[11px] text-[var(--text-tertiary)] cursor-pointer" title="Daily automatic docker prune"> <label className="flex items-center gap-1.5 v-mono text-[11px] text-[var(--text-tertiary)] cursor-pointer" title="Daily automatic docker prune">
@@ -179,6 +195,26 @@ export function NodeDetailPage() {
/> />
auto-prune auto-prune
</label> </label>
<button
onClick={() => cordonMutation.mutate(agent.schedulable)}
disabled={cordonMutation.isPending}
title={agent.schedulable
? 'Block new placements — running replicas stay'
: 'Re-open for placement'}
className="v-mono text-[11px] px-3 py-1.5 rounded-md border border-[var(--border-subtle)] text-[var(--text-secondary)] hover:text-[var(--text-primary)] disabled:opacity-50"
>
{agent.schedulable ? 'cordon' : 'uncordon'}
</button>
<button
onClick={() => setConfirmDrain(true)}
disabled={drainMutation.isPending || !online}
title={online
? 'Cordon + remove all service containers + unpin services (they redeploy locally)'
: 'Node must be online to drain'}
className="v-mono text-[11px] px-3 py-1.5 rounded-md border border-[var(--warning)]/40 text-[var(--warning)] hover:bg-[var(--warning)]/10 disabled:opacity-50"
>
drain
</button>
<button <button
onClick={() => setConfirmVolumes(true)} onClick={() => setConfirmVolumes(true)}
disabled={pruneMutation.isPending} disabled={pruneMutation.isPending}
@@ -302,6 +338,17 @@ export function NodeDetailPage() {
pending={pruneMutation.isPending} pending={pruneMutation.isPending}
/> />
)} )}
{confirmDrain && (
<ConfirmModal
title={`Drain ${agent.name}?`}
body="Every service container on this node is removed and the affected services are unpinned — they redeploy on the local host until re-pinned elsewhere. The node stays cordoned."
confirmLabel="drain node"
danger
onCancel={() => setConfirmDrain(false)}
onConfirm={() => drainMutation.mutate()}
pending={drainMutation.isPending}
/>
)}
{confirmDelete && ( {confirmDelete && (
<ConfirmModal <ConfirmModal
title={`Remove ${agent.name}?`} title={`Remove ${agent.name}?`}
@@ -48,6 +48,7 @@ import {
cloneService, cloneService,
moveService, moveService,
listProjects, listProjects,
listNodeOptions,
} from '@/lib/api-client'; } from '@/lib/api-client';
import { getDemoProjectById, getDemoServiceById, getDemoCronJobsByService } from '@/lib/demo-data'; import { getDemoProjectById, getDemoServiceById, getDemoCronJobsByService } from '@/lib/demo-data';
import { useDemoMode } from '@/lib/demo-mode'; import { useDemoMode } from '@/lib/demo-mode';
@@ -466,6 +467,7 @@ export function ServiceDetailPage() {
onSuccess: () => { onSuccess: () => {
setNetworkForm(null); setNetworkForm(null);
setSleepDraft(null); setSleepDraft(null);
setNodeDraft(null);
invalidateService(); invalidateService();
}, },
}); });
@@ -485,6 +487,13 @@ export function ServiceDetailPage() {
const [domainChecks, setDomainChecks] = useState<Record<string, DomainCheckResult> | null>(null); const [domainChecks, setDomainChecks] = useState<Record<string, DomainCheckResult> | null>(null);
const [accessForm, setAccessForm] = useState<{ maintenance: boolean; basicAuth: string } | null>(null); const [accessForm, setAccessForm] = useState<{ maintenance: boolean; basicAuth: string } | null>(null);
const [sleepDraft, setSleepDraft] = useState<{ id: string; enabled: boolean; minutes: string } | null>(null); const [sleepDraft, setSleepDraft] = useState<{ id: string; enabled: boolean; minutes: string } | null>(null);
const [nodeDraft, setNodeDraft] = useState<string | null>(null);
const nodesQuery = useQuery({
queryKey: ['node-options'],
queryFn: listNodeOptions,
enabled: !isDemoMode,
staleTime: 30_000,
});
const [buildForm, setBuildForm] = useState<{ const [buildForm, setBuildForm] = useState<{
builder: string; cpuReserve: string; memoryReserve: string; staticCmd: string; staticDir: string; builder: string; cpuReserve: string; memoryReserve: string; staticCmd: string; staticDir: string;
} | null>(null); } | null>(null);
@@ -2384,6 +2393,58 @@ export function ServiceDetailPage() {
</div> </div>
)} )}
{!isDemoMode && (nodesQuery.data?.length ?? 0) > 0 && (
<div className="mt-6 pt-6 border-t border-[var(--border-subtle)]">
<div className="mb-3">
<h3 className="text-sm font-medium text-[var(--text-primary)]">Placement</h3>
<p className="text-xs text-[var(--text-tertiary)] mt-0.5">
Which node runs this service's containers. Remote nodes pull the image
from its registry — git builds and domains stay on the local host.
</p>
</div>
<div className="flex items-center gap-3 text-xs">
<select
value={nodeDraft ?? (service.nodeId || 'local')}
onChange={(e) => setNodeDraft(e.target.value)}
className="px-2 py-1.5 rounded-[var(--radius-md)] bg-[var(--surface-muted)] border border-[var(--border-subtle)] text-sm text-[var(--text-primary)]"
>
<option value="local">This server (local Docker)</option>
<option value="auto">Auto — least-loaded online node</option>
{(nodesQuery.data ?? []).map((n) => (
<option key={n.id} value={n.id}>
{n.name}{n.status !== 'online' ? ' (offline)' : ''}
</option>
))}
</select>
{nodeDraft !== null && (
<>
<button
type="button"
disabled={updateServiceMutation.isPending}
onClick={() => updateServiceMutation.mutate({ node_id: nodeDraft })}
className="px-3 py-1.5 rounded-[var(--radius-md)] bg-[var(--accent-primary)] text-[var(--accent-on)] text-xs font-medium disabled:opacity-50"
>
Save
</button>
<button
type="button"
onClick={() => setNodeDraft(null)}
className="text-[var(--text-tertiary)] hover:text-[var(--text-primary)]"
>
Cancel
</button>
</>
)}
{service.nodeName && nodeDraft === null && (
<span className="text-[var(--text-tertiary)]">running on {service.nodeName}</span>
)}
</div>
{updateServiceMutation.isError && (
<p className="mt-2 text-xs text-[var(--error)]">{(updateServiceMutation.error as Error).message}</p>
)}
</div>
)}
{!isDemoMode && ( {!isDemoMode && (
<div className="mt-6 pt-6 border-t border-[var(--border-subtle)]"> <div className="mt-6 pt-6 border-t border-[var(--border-subtle)]">
<div className="flex items-center justify-between mb-4"> <div className="flex items-center justify-between mb-4">
+228
View File
@@ -7977,6 +7977,51 @@ export interface paths {
patch?: never; patch?: never;
trace?: 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": { "/agents": {
parameters: { parameters: {
query?: never; query?: never;
@@ -8514,6 +8559,179 @@ export interface paths {
patch?: never; patch?: never;
trace?: 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": { "/agents/{id}/metrics": {
parameters: { parameters: {
query?: never; query?: never;
@@ -10656,6 +10874,10 @@ export interface components {
sleep_enabled?: boolean; sleep_enabled?: boolean;
/** @description Idle timeout in minutes before sleeping (1-1440, default 15) */ /** @description Idle timeout in minutes before sleeping (1-1440, default 15) */
sleep_idle_minutes?: number; 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) */ /** @description Computed public URL (published port or domain) */
public_url?: string; public_url?: string;
/** /**
@@ -10726,6 +10948,8 @@ export interface components {
static_build_cmd?: string; static_build_cmd?: string;
/** @description Output directory for the static builder (default dist) */ /** @description Output directory for the static builder (default dist) */
static_dir?: string; static_dir?: string;
/** @description Node agent id to pin to, 'auto' = least-loaded online agent; omit/empty = local host */
node_id?: string;
}; };
UpdateServiceRequest: { UpdateServiceRequest: {
/** @description Service name */ /** @description Service name */
@@ -10787,6 +11011,8 @@ export interface components {
sleep_enabled?: boolean; sleep_enabled?: boolean;
/** @description Idle timeout in minutes before sleeping (1-1440) */ /** @description Idle timeout in minutes before sleeping (1-1440) */
sleep_idle_minutes?: number; sleep_idle_minutes?: number;
/** @description Node agent id to pin to, 'auto' = least-loaded online agent, 'local' = run on the Containr host (clears the pin) */
node_id?: string;
}; };
ServiceDomain: { ServiceDomain: {
/** Format: uuid */ /** Format: uuid */
@@ -11904,6 +12130,8 @@ export interface components {
}; };
/** @description Enqueue a bounded docker prune on this node once per day. */ /** @description Enqueue a bounded docker prune on this node once per day. */
auto_prune?: boolean; 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 */ /** Format: date-time */
last_heartbeat?: string; last_heartbeat?: string;
/** Format: date-time */ /** Format: date-time */
+1 -1
View File
@@ -40,7 +40,7 @@ export const bundledDocs: BundledDoc[] = [
"path": "guides/CLI.md", "path": "guides/CLI.md",
"section": "guides", "section": "guides",
"title": "Containr CLI", "title": "Containr CLI",
"markdown": "# Containr CLI\n\nThe CLI controls every part of a Containr instance — the same API surface\nthe web UI uses. Build it with `scripts/build-cli.sh` or\n`go build ./cmd/cli` from `app/backend`.\n\n## Authentication\n\nCreate a personal access token in the web UI under\n**Settings → Personal Access Tokens** (or `containr tokens create`), then:\n\n```bash\ncontainr auth login cnp_... # verify + store\ncontainr auth login cnp_... --url https://containr.example.com\ncontainr auth login cnp_... --name production # named profile\ncontainr auth status # show active profile + identity\ncontainr auth logout\n```\n\nResolution order for every command: `--token`/`--api-url` flags >\n`CONTAINR_TOKEN`/`CONTAINR_API_URL` env > active profile >\n`http://localhost:8080/api/v1`. The active profile is `--profile` >\n`CONTAINR_PROFILE` > `current_profile` in `~/.containr.yaml`.\n\n## Output and scripting\n\n- `--json` prints the raw API response — safe to pipe and parse.\n- `--yes` / `-y` skips confirmation on destructive commands (delete,\n revoke, restore, failover). Non-interactive shells refuse destructive\n commands without it.\n- Exit codes are stable: `0` ok, `1` error, `2` auth/permission,\n `3` not found, `4` validation, `5` unreachable.\n\n## Commands\n\n```\nauth login|logout|status token auth and profiles\ntokens list|create|revoke personal access tokens\nprojects list|get|create|delete\nservices list|get|create|update|delete|start|stop|restart|redeploy\nservices clone|move duplicate config or re-parent a service\nservices domains list|add|remove|default|check multi-domain + DNS preflight\nregistries list|add|update|remove private-image pull credentials\nvolumes list|delete docker volume inventory + cleanup (admin)\ndeploy <service-id> trigger a deployment (--no-cache, --commit-hash, --branch)\ndeployments list|get|logs|rollback|cancel\nlogs <service-id> [--follow] runtime logs (SSE follow)\nexec <service-id> -- <cmd> one-off container command\nvariables list|set|unset env vars (secrets masked; --redeploy applies them)\nvariables shared list|set|unset project shared vars — referenced as ${{shared.KEY}}\ndatabases list|get|create|update|delete|action|backup|restore|download-backup|import-backup\n import-backup <id> <file.tar.gz|-> uploads an archive as a restore\n point; ships offsite too when a backup target is assigned\ndatabases list|get|create|update|delete|action|backup|restore|download-backup\n create --public exposes the port on all interfaces;\n update <id> --public/--public=false toggles it (container recreates, data persists)\ndatabases register <name> register an external database (--type --host --port --database --username --password --ssl; probed before storing)\ndatabases test-connection probe a connection — by id, or ad-hoc flags\nbackup-targets list|add|update|remove|test\n S3-compatible archive destinations; add/update probe the bucket\n first, empty --access-key/--secret-key on update keep stored keys;\n update <db-id> --backup-target assigns a target (\"\" clears it)\ncron list|get|create|delete|trigger|executions\ntemplates list|get|create|delete|deploy|plan|import-compose|deploy-graph\n deploy --name --var K=V; plan dry-resolves expressions;\n import-compose <compose.yml|-> converts a compose file into\n a v2 graph config; deploy-graph <project-id> <config.json|->\n deploys an ad-hoc service graph\nimport compose <project-id> <compose.yml|->\n deploys a docker-compose file straight into a project —\n no template row. DB images become managed databases\n attached under their service name; --var K=V overrides\n template variables\nnodes list|get|delete|tokens node agents + onboarding tokens (admin)\nnodes add <user@host> SSH bootstrap — one session installs the agent, then never needed again (--print for manual runs)\nnodes update|prune|commands rename/auto-prune, bounded docker prune, command history\nmigrate railway pull a Railway project (services, vars, domains) via GraphQL and recreate it — --dry-run prints the plan\nmigrate dflow import a dflow project (services, vars, volumes, domains) via its Payload REST API — needs --url + --api-key, --dry-run prints the plan\nscaling status|policies|services|scale\nha status|alerts|health|policies|enable|disable|failover\nsecurity scan|vulnerabilities|metrics\ngateway services|keys|anLine truncated "markdown": "# Containr CLI\n\nThe CLI controls every part of a Containr instance — the same API surface\nthe web UI uses. Build it with `scripts/build-cli.sh` or\n`go build ./cmd/cli` from `app/backend`.\n\n## Authentication\n\nCreate a personal access token in the web UI under\n**Settings → Personal Access Tokens** (or `containr tokens create`), then:\n\n```bash\ncontainr auth login cnp_... # verify + store\ncontainr auth login cnp_... --url https://containr.example.com\ncontainr auth login cnp_... --name production # named profile\ncontainr auth status # show active profile + identity\ncontainr auth logout\n```\n\nResolution order for every command: `--token`/`--api-url` flags >\n`CONTAINR_TOKEN`/`CONTAINR_API_URL` env > active profile >\n`http://localhost:8080/api/v1`. The active profile is `--profile` >\n`CONTAINR_PROFILE` > `current_profile` in `~/.containr.yaml`.\n\n## Output and scripting\n\n- `--json` prints the raw API response — safe to pipe and parse.\n- `--yes` / `-y` skips confirmation on destructive commands (delete,\n revoke, restore, failover). Non-interactive shells refuse destructive\n commands without it.\n- Exit codes are stable: `0` ok, `1` error, `2` auth/permission,\n `3` not found, `4` validation, `5` unreachable.\n\n## Commands\n\n```\nauth login|logout|status token auth and profiles\ntokens list|create|revoke personal access tokens\nprojects list|get|create|delete\nservices list|get|create|update|delete|start|stop|restart|redeploy\nservices clone|move duplicate config or re-parent a service\nservices domains list|add|remove|default|check multi-domain + DNS preflight\nregistries list|add|update|remove private-image pull credentials\nvolumes list|delete docker volume inventory + cleanup (admin)\ndeploy <service-id> trigger a deployment (--no-cache, --commit-hash, --branch)\ndeployments list|get|logs|rollback|cancel\nlogs <service-id> [--follow] runtime logs (SSE follow)\nexec <service-id> -- <cmd> one-off container command\nvariables list|set|unset env vars (secrets masked; --redeploy applies them)\nvariables shared list|set|unset project shared vars — referenced as ${{shared.KEY}}\ndatabases list|get|create|update|delete|action|backup|restore|download-backup|import-backup\n import-backup <id> <file.tar.gz|-> uploads an archive as a restore\n point; ships offsite too when a backup target is assigned\ndatabases list|get|create|update|delete|action|backup|restore|download-backup\n create --public exposes the port on all interfaces;\n update <id> --public/--public=false toggles it (container recreates, data persists)\ndatabases register <name> register an external database (--type --host --port --database --username --password --ssl; probed before storing)\ndatabases test-connection probe a connection — by id, or ad-hoc flags\nbackup-targets list|add|update|remove|test\n S3-compatible archive destinations; add/update probe the bucket\n first, empty --access-key/--secret-key on update keep stored keys;\n update <db-id> --backup-target assigns a target (\"\" clears it)\ncron list|get|create|delete|trigger|executions\ntemplates list|get|create|delete|deploy|plan|import-compose|deploy-graph\n deploy --name --var K=V; plan dry-resolves expressions;\n import-compose <compose.yml|-> converts a compose file into\n a v2 graph config; deploy-graph <project-id> <config.json|->\n deploys an ad-hoc service graph\nimport compose <project-id> <compose.yml|->\n deploys a docker-compose file straight into a project —\n no template row. DB images become managed databases\n attached under their service name; --var K=V overrides\n template variables\nnodes list|get|delete|tokens node agents + onboarding tokens (admin)\nnodes add <user@host> SSH bootstrap — one session installs the agent, then never needed again (--print for manual runs)\nnodes update|prune|commands rename/auto-prune, bounded docker prune, command history\nmigrate railway pull a Railway project (services, vars, domains) via GraphQL and recreate it — --dry-run prints the plan\nmigrate dflow import a dflow project (services, vars, volumes, domains) via its Payload REST API — needs --url + --api-key, --dry-run prints the plan\nscaling status|policies|services|scale\nha status|alerts|health|policies|enable|disable|failover\nsecurity scan|vulnerabilities|metrics\ngateway services|keys|anLine truncated
}, },
{ {
"path": "guides/CLOUDFLARE_SETUP.md", "path": "guides/CLOUDFLARE_SETUP.md",
+27
View File
@@ -84,9 +84,22 @@ export type ServiceEntity = {
staticDir?: string; staticDir?: string;
sleepEnabled?: boolean; sleepEnabled?: boolean;
sleepIdleMinutes?: number; sleepIdleMinutes?: number;
nodeId?: string;
nodeName?: string;
publicUrl?: string; publicUrl?: string;
}; };
export type NodeOption = {
id: string;
name: string;
status: string;
};
export async function listNodeOptions(): Promise<NodeOption[]> {
const payload = await requestJson<{ nodes?: NodeOption[] }>('/nodes/options');
return payload.nodes ?? [];
}
export type ServiceDomain = { export type ServiceDomain = {
id: string; id: string;
service_id: string; service_id: string;
@@ -343,6 +356,7 @@ export type NodeAgentEntity = {
capabilities: AgentCapabilities; capabilities: AgentCapabilities;
resources: NodeResources; resources: NodeResources;
autoPrune: boolean; autoPrune: boolean;
schedulable: boolean;
metadata: Record<string, unknown>; metadata: Record<string, unknown>;
lastHeartbeat?: string; lastHeartbeat?: string;
createdAt?: string; createdAt?: string;
@@ -680,6 +694,8 @@ function normalizeService(service: RawService): ServiceEntity | null {
staticDir: service.static_dir, staticDir: service.static_dir,
sleepEnabled: service.sleep_enabled, sleepEnabled: service.sleep_enabled,
sleepIdleMinutes: service.sleep_idle_minutes, sleepIdleMinutes: service.sleep_idle_minutes,
nodeId: service.node_id,
nodeName: service.node_name,
publicUrl: service.public_url, publicUrl: service.public_url,
}; };
} }
@@ -1081,6 +1097,7 @@ function normalizeAgent(agent: RawNodeAgent): NodeAgentEntity | null {
}, },
lastHeartbeat: agent.last_heartbeat, lastHeartbeat: agent.last_heartbeat,
autoPrune: agent.auto_prune ?? false, autoPrune: agent.auto_prune ?? false,
schedulable: agent.schedulable ?? true,
metadata: (agent.metadata ?? {}) as Record<string, unknown>, metadata: (agent.metadata ?? {}) as Record<string, unknown>,
createdAt: agent.created_at, createdAt: agent.created_at,
updatedAt: agent.updated_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<void> {
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<DrainResult> {
return requestJson<DrainResult>(`/agents/${id}/drain`, { method: 'POST' });
}
export async function getAgent(id: string): Promise<NodeAgentEntity | null> { export async function getAgent(id: string): Promise<NodeAgentEntity | null> {
const payload = await requestJson<{ agent?: RawNodeAgent }>(`/agents/${id}`); const payload = await requestJson<{ agent?: RawNodeAgent }>(`/agents/${id}`);
return payload.agent ? normalizeAgent(payload.agent) : null; return payload.agent ? normalizeAgent(payload.agent) : null;
+142
View File
@@ -5080,6 +5080,34 @@ paths:
$ref: '#/components/schemas/MessageResponse' $ref: '#/components/schemas/MessageResponse'
# Node agents — management (JWT-authenticated) # 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: /agents:
get: get:
tags: tags:
@@ -5378,6 +5406,103 @@ paths:
schema: schema:
$ref: '#/components/schemas/Error' $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: /agents/{id}/metrics:
get: get:
tags: tags:
@@ -6843,6 +6968,12 @@ components:
sleep_idle_minutes: sleep_idle_minutes:
type: integer type: integer
description: Idle timeout in minutes before sleeping (1-1440, default 15) 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: public_url:
type: string type: string
description: Computed public URL (published port or domain) description: Computed public URL (published port or domain)
@@ -6934,6 +7065,9 @@ components:
static_dir: static_dir:
type: string type: string
description: Output directory for the static builder (default dist) 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: UpdateServiceRequest:
type: object type: object
@@ -7020,6 +7154,9 @@ components:
sleep_idle_minutes: sleep_idle_minutes:
type: integer type: integer
description: Idle timeout in minutes before sleeping (1-1440) 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: ServiceDomain:
type: object type: object
@@ -8900,6 +9037,11 @@ components:
auto_prune: auto_prune:
type: boolean type: boolean
description: Enqueue a bounded docker prune on this node once per day. 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: last_heartbeat:
type: string type: string
format: date-time format: date-time
+24
View File
@@ -132,6 +132,30 @@ redeploy — `containr deploy <id>` to apply immediately.
`volumes list` shows the node's docker volume inventory with in-use `volumes list` shows the node's docker volume inventory with in-use
flags; `volumes delete <name>` removes an unused volume (admin scope). flags; `volumes delete <name>` 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 <id> --name api --image ghcr.io/x/api:v1 --node <agent-id>
containr services update <id> --node auto # least-loaded online node
containr services update <id> --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 <id> # no new pins; running replicas stay
containr nodes uncordon <id>
containr nodes drain <id> # cordon + remove its containers + unpin services
```
## Service domains and access ## Service domains and access
A service can carry multiple hostnames; the default is mirrored to A service can carry multiple hostnames; the default is mirrored to