Compare commits

..
5 Commits
Author SHA1 Message Date
Tomáš Dvořák d57ac72aa5 Merge pull request #59 from Dvorinka/feat/remote-ingress
CI / Changed paths (push) Canceled after 0s
CI / Infra (Compose + Scripts) (push) Canceled after 0s
CI / Backend (Go) (push) Canceled after 0s
CI / Frontend (React) (push) Canceled after 0s
CI / Mobile (Expo) (push) Canceled after 0s
CI / Compose boot smoke (push) Canceled after 0s
CI / Build & publish images (push) Canceled after 0s
CI / Release (push) Canceled after 0s
feat(nodes): remote ingress via Traefik file provider
2026-10-09 19:26:51 +02:00
Tomas DvorakandDevin 59dfd35583 feat(nodes): remote sleep/wake — idle detection + wake-page ingress
Agents report per-container net IO at heartbeat; the sleeper measures
idleness from those counters for remote services, removes replicas via
the agent queue, and repoints the domain route at the backend's new
/internal/wake-page endpoint (replacePath + priority-100 route in the
file provider). First hit serves a reloading page and retriggers the
wake reconcile, which recreates replicas and restores real upstreams.

Manual POST /services/:id/sleep also routes remote; local sleep/wake
flow is unchanged.

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

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-09 18:43:33 +02:00
Tomas DvorakandDevin 4028639239 feat(nodes): agent-reported container liveness at heartbeat
The agent now lists every containr-managed container's docker state at
each heartbeat; the server reconciles container_instances for rows that
node owns, so exits, OOMs, and crash loops surface in runtime state
between reconciles. Tombstoned rows are never resurrected.

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

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-09 18:29:48 +02:00
Tomas DvorakandDevin c3a35081d0 feat(nodes): remote ingress via Traefik file provider
Remote replicas publish an ephemeral host port; the agent reports the
assigned binding back in the create_container result and the backend
renders a per-service file-provider config into TRAEFIK_DYNAMIC_DIR
pointing Traefik at the node's mesh IP + port. Routes are rewritten on
every remote reconcile and removed on teardown or local placement.

Also: RetireService retires remote leftovers when a service returns to
local placement; remove_container is idempotent agent-side; tombstoned
replicas skip remove dispatch; the domains-stay-local ceiling is gone.

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

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-09 18:19:08 +02:00
Tomáš DvořákandDevin 4458de56b0 feat(scheduling): node tags + service placement_tags affinity (#58)
Operator-set node labels (node_agents.tags) gate which nodes a service
may schedule onto: services.placement_tags requires every listed tag.
auto and spread filter eligible agents by tags; tags are re-evaluated
each reconcile so retagging a node re-spreads replicas. Pin + tags and
local + tags rejected; empty array clears the requirement.

Surfaces: --placement-tags on services create/update, nodes update
--tags, MCP args, OpenAPI, node-detail tag editor, service-detail
placement-tags input.

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

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-09 18:11:19 +02:00
29 changed files with 1160 additions and 82 deletions

No files matched your search

+4
View File
@@ -49,6 +49,10 @@ CONTAINR_AGENT_AUTH_TOKEN=replace_with_strong_agent_secret
# CONTAINR_AGENT_AUTH_TOKENS=current_secret,next_secret # CONTAINR_AGENT_AUTH_TOKENS=current_secret,next_secret
# Bundled agent binaries dir served at /api/agents/download/:platform. # Bundled agent binaries dir served at /api/agents/download/:platform.
# CONTAINR_AGENT_BIN_DIR=/agents # CONTAINR_AGENT_BIN_DIR=/agents
# Directory Traefik's file provider watches. The backend writes one
# config per remote service here so domains route to node agents. The
# infra compose wires a shared volume — set only for custom layouts.
# TRAEFIK_DYNAMIC_DIR=/traefik-dynamic
# Optional remote template catalog (another Containr instance). Its # Optional remote template catalog (another Containr instance). Its
# templates merge into GET /templates as source:"registry" and # templates merge into GET /templates as source:"registry" and
# materialize locally on deploy. The bundled catalog works without it. # materialize locally on deploy. The bundled catalog works without it.
+26 -5
View File
@@ -296,11 +296,32 @@ and wired refs, one click or one CLI call.
utilisation, then cpu, then container count, with uninstrumented utilisation, then cpu, then container count, with uninstrumented
agents sorting last. Lifecycle and `GET /runtime` are agents sorting last. Lifecycle and `GET /runtime` are
inventory-driven, so remote replicas show up and tear down wherever inventory-driven, so remote replicas show up and tear down wherever
they run. Remote nodes run registry-pulled they run. Remote ingress shipped: the agent reports assigned host
images only — git builds, Traefik domains, and remote sleep are ports in the `create_container` result, the backend renders a
documented ceilings. Remaining: `scheduling_rules` affinity Traefik file-provider config per remote service into
policies, deployment status reported back from agent `TRAEFIK_DYNAMIC_DIR` (shared volume, `watch=true`) pointing the
results, spread weighting by real capacity. loadbalancer at each replica's mesh IP + published port; route files
are rewritten on every remote reconcile and removed on teardown or
local placement. Agents report managed-container states at every
heartbeat — exits, OOMs, and crash loops land in inventory between
reconciles (tombstones never resurrect). Remote sleep/wake shipped:
heartbeats carry per-container net counters, idle remote services
lose replicas via agent remove + tombstones, and the route file
repoints at `/internal/wake-page/:id` on the backend — first hit
retriggers the wake reconcile, which recreates replicas and restores
real upstreams. Remote nodes run
registry-pulled images only — git builds are the remaining
documented ceiling. Tag affinity shipped: `node_agents.tags`
(operator-set via `PUT /agents/:id` / `nodes update --tags`) +
`services.placement_tags` — `auto` and `spread` only consider
online schedulable agents carrying every required tag, tags are
re-evaluated each reconcile (a tag change re-spreads replicas),
unsatisfiable requirements fail with a clear error, pin + tags
and local + tags are rejected, empty array clears. Surfaces:
`--placement-tags` on services create/update, `--tags` on
`nodes update`, MCP args, OpenAPI, node-detail tag editor +
service-detail placement-tags input. Remaining: build-on-node,
remote sleep/wake, spread weighting by real capacity.
- [x] **Agent self-upgrade** — `POST /agents/:id/upgrade` enqueues a - [x] **Agent self-upgrade** — `POST /agents/:id/upgrade` enqueues a
`self_upgrade` command carrying platform + sha256 + server version; `self_upgrade` command carrying platform + sha256 + server version;
the agent downloads its replacement from `GET /agents/download/ the agent downloads its replacement from `GET /agents/download/
+107 -2
View File
@@ -204,10 +204,92 @@ func sendHeartbeat(ctx context.Context, client *http.Client, cfg agentConfig, ag
"system_load": readSystemLoad(), "system_load": readSystemLoad(),
"uptime": readSystemUptime(), "uptime": readSystemUptime(),
"version": firstEnv("CONTAINR_AGENT_VERSION", version), "version": firstEnv("CONTAINR_AGENT_VERSION", version),
"containers": managedContainerStates(),
} }
return postJSON(ctx, client, cfg, "/api/agents/heartbeat", payload, nil) return postJSON(ctx, client, cfg, "/api/agents/heartbeat", payload, nil)
} }
// managedContainerStates lists every containr-managed container's docker
// state plus cumulative network counters — state keeps runtime honest,
// net bytes let the sleeper detect idle remote services.
func managedContainerStates() []map[string]interface{} {
out, err := runDocker("ps", "-a",
"--filter", "label=containr.managed=true",
"--format", "{{.Names}}\t{{.State}}")
if err != nil {
return nil
}
netIO := containerNetIO()
var states []map[string]interface{}
for _, line := range strings.Split(out, "\n") {
parts := strings.SplitN(line, "\t", 2)
if len(parts) != 2 || parts[0] == "" {
continue
}
entry := map[string]interface{}{"name": parts[0], "state": parts[1]}
if io, ok := netIO[parts[0]]; ok {
entry["rx_bytes"], entry["tx_bytes"] = io[0], io[1]
}
states = append(states, entry)
}
return states
}
// containerNetIO snapshots cumulative rx/tx per running container. Stats
// only covers running containers — stopped replicas report state without
// counters, which is all the sleeper needs.
func containerNetIO() map[string][2]int64 {
out, err := runDocker("stats", "--no-stream",
"--format", "{{.Name}}\t{{.NetIO}}")
if err != nil {
return nil
}
result := map[string][2]int64{}
for _, line := range strings.Split(out, "\n") {
parts := strings.SplitN(line, "\t", 2)
if len(parts) != 2 {
continue
}
io := strings.Split(parts[1], "/")
if len(io) != 2 {
continue
}
result[parts[0]] = [2]int64{parseHumanBytes(io[0]), parseHumanBytes(io[1])}
}
return result
}
// parseHumanBytes converts docker's human units ("1.2kB", "3.4MB") to bytes.
func parseHumanBytes(s string) int64 {
s = strings.TrimSpace(s)
var num float64
var unit string
if _, err := fmt.Sscanf(s, "%f%s", &num, &unit); err != nil {
return 0
}
switch strings.ToUpper(unit) {
case "B":
return int64(num)
case "KB":
return int64(num * 1e3)
case "MB":
return int64(num * 1e6)
case "GB":
return int64(num * 1e9)
case "TB":
return int64(num * 1e12)
case "KIB":
return int64(num * 1024)
case "MIB":
return int64(num * 1024 * 1024)
case "GIB":
return int64(num * 1024 * 1024 * 1024)
case "TIB":
return int64(num * 1024 * 1024 * 1024 * 1024)
}
return int64(num)
}
func processCommands(ctx context.Context, client *http.Client, cfg agentConfig, agentID string) error { func processCommands(ctx context.Context, client *http.Client, cfg agentConfig, agentID string) error {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("%s/api/agents/%s/commands", cfg.APIURL, agentID), nil) req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("%s/api/agents/%s/commands", cfg.APIURL, agentID), nil)
if err != nil { if err != nil {
@@ -272,7 +354,12 @@ func executeCommand(item command) (string, error) {
case "restart_container": case "restart_container":
return runDocker("restart", commandTarget(item.Payload)) return runDocker("restart", commandTarget(item.Payload))
case "remove_container": case "remove_container":
return runDocker("rm", "-f", commandTarget(item.Payload)) out, err := runDocker("rm", "-f", commandTarget(item.Payload))
// A missing container is the desired end state — report success.
if err != nil && strings.Contains(out, "No such container") {
return out, nil
}
return out, err
case "prune": case "prune":
return dockerPrune(item.Payload) return dockerPrune(item.Payload)
case "system_df": case "system_df":
@@ -460,7 +547,25 @@ func createContainer(payload map[string]interface{}) (string, error) {
logs, _ := runDocker("logs", "--tail", "5", name) logs, _ := runDocker("logs", "--tail", "5", name)
return "", fmt.Errorf("container %s is %q after start: %s", name, state, logs) return "", fmt.Errorf("container %s is %q after start: %s", name, state, logs)
} }
return name, nil // Report the real host port bindings — ephemeral publishes pick a port
// the server can't guess, and remote ingress needs the actual mapping.
hostPorts := map[string]string{}
portsJSON, _ := runDocker("inspect", "-f", "{{json .NetworkSettings.Ports}}", name)
var bindings map[string][]struct {
HostIP string `json:"HostIp"`
HostPort string `json:"HostPort"`
}
if json.Unmarshal([]byte(portsJSON), &bindings) == nil {
for cport, addrs := range bindings {
for _, a := range addrs {
if a.HostPort != "" && a.HostPort != "0" {
hostPorts[cport] = a.HostPort
}
}
}
}
out, _ := json.Marshal(map[string]interface{}{"name": name, "host_ports": hostPorts})
return string(out), nil
} }
// registryHostOf mirrors the server's registryHost: first path segment with // registryHostOf mirrors the server's registryHost: first path segment with
+59 -1
View File
@@ -42,6 +42,9 @@ type NodeAgent struct {
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"` Schedulable bool `json:"schedulable"`
// Tags are operator-set placement labels; services with placement_tags
// only schedule onto agents carrying every required tag.
Tags []string `json:"tags"`
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"`
@@ -195,6 +198,17 @@ type AgentHeartbeat struct {
SystemLoad SystemLoad `json:"system_load"` SystemLoad SystemLoad `json:"system_load"`
Uptime int64 `json:"uptime"` Uptime int64 `json:"uptime"`
Version string `json:"version"` Version string `json:"version"`
// Containers reports managed-container liveness — {name, state} per
// containr.managed container on the node — so remote replica status
// reflects reality between reconciles (exits, OOMs, crash loops).
Containers []AgentContainerState `json:"containers,omitempty"`
}
type AgentContainerState struct {
Name string `json:"name"`
State string `json:"state"`
RxBytes int64 `json:"rx_bytes,omitempty"`
TxBytes int64 `json:"tx_bytes,omitempty"`
} }
type SystemLoad struct { type SystemLoad struct {
@@ -242,10 +256,15 @@ func (h *NodeAgentHandler) ListNodeOptions(c *gin.Context) {
} }
options := make([]gin.H, 0, len(rows)) options := make([]gin.H, 0, len(rows))
for _, row := range rows { for _, row := range rows {
var tags []string
if len(row.Tags) > 0 {
_ = json.Unmarshal(row.Tags, &tags)
}
options = append(options, gin.H{ options = append(options, gin.H{
"id": row.ID, "id": row.ID,
"name": row.Name, "name": row.Name,
"status": row.Status.String, "status": row.Status.String,
"tags": tags,
}) })
} }
c.JSON(http.StatusOK, gin.H{"nodes": options}) c.JSON(http.StatusOK, gin.H{"nodes": options})
@@ -262,6 +281,7 @@ func agentFromRow(row sqlcdb.NodeAgent) NodeAgent {
Version: row.Version.String, Version: row.Version.String,
AutoPrune: row.AutoPrune, AutoPrune: row.AutoPrune,
Schedulable: row.Schedulable, Schedulable: row.Schedulable,
Tags: []string{},
Metadata: map[string]interface{}{}, Metadata: map[string]interface{}{},
} }
if row.LastHeartbeat.Valid { if row.LastHeartbeat.Valid {
@@ -276,6 +296,9 @@ func agentFromRow(row sqlcdb.NodeAgent) NodeAgent {
unmarshalRaw(row.Capabilities, &agent.Capabilities) unmarshalRaw(row.Capabilities, &agent.Capabilities)
unmarshalRaw(row.Resources, &agent.Resources) unmarshalRaw(row.Resources, &agent.Resources)
unmarshalRaw(row.Metadata, &agent.Metadata) unmarshalRaw(row.Metadata, &agent.Metadata)
if len(row.Tags) > 0 {
_ = json.Unmarshal(row.Tags, &agent.Tags)
}
return agent return agent
} }
@@ -546,7 +569,26 @@ func (h *NodeAgentHandler) UpdateAgent(c *gin.Context) {
return return
} }
c.JSON(http.StatusOK, gin.H{"agent": agentFromRow(updated)}) // Tags persist through their own column write — UpdateAgent keeps the
// register-path signature untouched.
resp := agentFromRow(updated)
if raw, ok := updates["tags"]; ok {
var tags []string
if b, mErr := json.Marshal(raw); mErr == nil {
_ = json.Unmarshal(b, &tags)
}
tags = normalizePlacementTags(tags)
if err := h.q.SetAgentTags(ctx, sqlcdb.SetAgentTagsParams{
ID: id,
Column2: json.RawMessage(tagsJSON(tags)),
}); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent tags"})
return
}
resp.Tags = tags
}
c.JSON(http.StatusOK, gin.H{"agent": resp})
} }
// applyAgentUpdates merges whitelisted keys from an arbitrary update map. // applyAgentUpdates merges whitelisted keys from an arbitrary update map.
@@ -634,6 +676,22 @@ func (h *NodeAgentHandler) SendHeartbeat(c *gin.Context) {
return return
} }
// Reconcile remote container liveness — exits and crash loops surface
// in inventory between reconciles. Best-effort, never fatal.
for _, rc := range heartbeat.Containers {
if rc.Name == "" || rc.State == "" {
continue
}
stateJSON, _ := json.Marshal(rc.State)
bytesJSON, _ := json.Marshal(rc.RxBytes + rc.TxBytes)
_ = h.q.UpdateContainerStateByName(ctx, sqlcdb.UpdateContainerStateByNameParams{
Name: rc.Name,
NodeAgentID: heartbeat.NodeAgentID,
Column3: stateJSON,
Column4: bytesJSON,
})
}
c.Status(http.StatusOK) c.Status(http.StatusOK)
} }
+212
View File
@@ -0,0 +1,212 @@
package api
import (
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"containr/internal/database"
"containr/internal/database/sqlcdb"
"containr/internal/deployment"
)
// Remote ingress: replicas on node agents aren't on the hub's
// containr-network, so Traefik's Docker provider can't see them. Instead
// the backend renders one file-provider config per service into
// TRAEFIK_DYNAMIC_DIR (a directory Traefik watches), pointing the
// loadbalancer at the node's reachable IP + the ephemeral host port the
// agent reported back after `create_container`.
//
// Empty dir → feature off (bare-metal dev). Files are best-effort: a
// failed write never fails a reconcile.
func ingressDir() string { return os.Getenv("TRAEFIK_DYNAMIC_DIR") }
func ingressFile(serviceID string) string {
return filepath.Join(ingressDir(), "containr-"+serviceID+".yml")
}
// jsonInt coerces JSON-ish numbers (int/int32/int64/float64) to int —
// spec fields arrive as int32, agent payloads unmarshal as float64.
func jsonInt(v interface{}) int {
switch n := v.(type) {
case int:
return n
case int32:
return int(n)
case int64:
return int(n)
case float64:
return int(n)
}
return 0
}
// reportedHostPorts parses the agent's create_container result —
// {"name": ..., "host_ports": {"80/tcp": "32768"}}. Older agents return a
// bare name string; JSON parse failure means no port info.
func reportedHostPorts(result string) map[string]string {
var out struct {
HostPorts map[string]string `json:"host_ports"`
}
if json.Unmarshal([]byte(result), &out) != nil {
return nil
}
return out.HostPorts
}
// nodeIngressIP picks the address the hub can reach the node on. Mesh
// overlays first — the agent operator joined a mesh precisely so the hub
// can reach it — then the reported primary IP as a LAN fallback.
func nodeIngressIP(agent sqlcdb.NodeAgent) string {
var meta struct {
Mesh map[string]string `json:"mesh"`
}
_ = json.Unmarshal(agent.Metadata.RawMessage, &meta)
for _, prefix := range []string{"tailscale", "netbird", "wg", "zt"} {
var keys []string
for name := range meta.Mesh {
keys = append(keys, name)
}
sort.Strings(keys)
for _, name := range keys {
if strings.HasPrefix(name, prefix) && meta.Mesh[name] != "" {
return meta.Mesh[name]
}
}
}
return agent.IpAddress
}
// syncRemoteIngress rewrites the service's file-provider config from the
// live remote replica set. Called after every remote reconcile so replica
// adds/removals and node retargets converge the upstream list.
func syncRemoteIngress(ctx context.Context, db *database.DB, spec deployment.RuntimeSpec) {
if ingressDir() == "" {
return
}
domains := spec.Domains
if len(domains) == 0 && spec.Domain != "" {
domains = []string{spec.Domain}
}
if len(domains) == 0 || spec.Port <= 0 {
removeServiceIngress(spec.ServiceID)
return
}
q := sqlcdb.New(db.DB)
rows, err := q.ListServiceContainers(ctx, spec.ServiceID)
if err != nil {
return
}
type upstream struct{ ip, port string }
var servers []upstream
for _, row := range rows {
var st struct {
State string `json:"state"`
}
_ = json.Unmarshal(row.Status.RawMessage, &st)
if st.State != "running" {
continue
}
var port string
var plist []map[string]interface{}
if err := json.Unmarshal(row.Ports.RawMessage, &plist); err == nil {
for _, p := range plist {
if hp, ok := p["host_port"].(float64); ok && hp > 0 {
port = fmt.Sprintf("%d", int(hp))
}
}
}
if port == "" {
continue
}
agent, err := q.GetAgent(ctx, row.NodeAgentID)
if err != nil {
continue
}
if ip := nodeIngressIP(agent); ip != "" {
servers = append(servers, upstream{ip, port})
}
}
if len(servers) == 0 {
removeServiceIngress(spec.ServiceID)
return
}
router := "svc-" + spec.ServiceID[:8]
rules := make([]string, 0, len(domains))
for _, d := range domains {
rules = append(rules, "Host(`"+d+"`)")
}
var b strings.Builder
b.WriteString("http:\n routers:\n " + router + ":\n")
b.WriteString(" rule: \"" + strings.Join(rules, " || ") + "\"\n")
b.WriteString(" entrypoints: [web]\n")
b.WriteString(" service: " + router + "\n")
b.WriteString(" services:\n " + router + ":\n loadBalancer:\n servers:\n")
for _, s := range servers {
b.WriteString(" - url: \"http://" + s.ip + ":" + s.port + "\"\n")
}
// Atomic-ish: write temp then rename so Traefik's watcher never reads a
// truncated file.
tmp := ingressFile(spec.ServiceID) + ".tmp"
if err := os.WriteFile(tmp, []byte(b.String()), 0o644); err != nil {
return
}
_ = os.Rename(tmp, ingressFile(spec.ServiceID))
}
// removeServiceIngress deletes the service's remote route. Called on
// remote teardown and when a service returns to local placement.
func removeServiceIngress(serviceID string) {
if ingressDir() == "" {
return
}
_ = os.Remove(ingressFile(serviceID))
}
// wakeUpstreamURL is where a sleeping remote service's route points —
// the backend's wake-page endpoint. Defaults to the compose service name;
// TRAEFIK_BACKEND_URL overrides for custom layouts.
func wakeUpstreamURL(apiPort int) string {
if v := strings.TrimSpace(os.Getenv("TRAEFIK_BACKEND_URL")); v != "" {
return strings.TrimRight(v, "/")
}
return fmt.Sprintf("http://containr-backend:%d", apiPort)
}
// writeWakeIngress routes the service's domains at the wake-page endpoint —
// first hit triggers the wake reconcile and serves a reloading page.
func writeWakeIngress(serviceID string, domains []string, apiPort int) {
if ingressDir() == "" {
return
}
sort.Strings(domains)
router := "svc-" + serviceID[:8] + "-wake"
rules := make([]string, 0, len(domains))
for _, d := range domains {
rules = append(rules, "Host(`"+d+"`)")
}
var b strings.Builder
mw := router + "-rw"
b.WriteString("http:\n routers:\n " + router + ":\n")
b.WriteString(" rule: \"" + strings.Join(rules, " || ") + "\"\n")
b.WriteString(" entrypoints: [web]\n")
b.WriteString(" service: " + router + "\n")
b.WriteString(" priority: 100\n")
b.WriteString(" middlewares: [" + mw + "]\n")
b.WriteString(" middlewares:\n " + mw + ":\n")
b.WriteString(" replacePath:\n path: \"/api/v1/internal/wake-page/" + serviceID + "\"\n")
b.WriteString(" services:\n " + router + ":\n loadBalancer:\n servers:\n")
b.WriteString(" - url: \"" + wakeUpstreamURL(apiPort) + "\"\n")
tmp := ingressFile(serviceID) + ".tmp"
if err := os.WriteFile(tmp, []byte(b.String()), 0o644); err != nil {
return
}
_ = os.Rename(tmp, ingressFile(serviceID))
}
+113 -29
View File
@@ -4,6 +4,7 @@ import (
"context" "context"
"crypto/sha256" "crypto/sha256"
"database/sql" "database/sql"
"encoding/json"
"fmt" "fmt"
"strconv" "strconv"
"strings" "strings"
@@ -18,8 +19,8 @@ import (
// agentNodeRunner implements deployment.NodeRunner by dispatching container // agentNodeRunner implements deployment.NodeRunner by dispatching container
// lifecycle commands to a node agent's command queue and waiting for the // lifecycle commands to a node agent's command queue and waiting for the
// reported result. Remote nodes have no Traefik — domains stay local-only // reported result. Domains route through the file provider —
// until per-node ingress lands. // syncRemoteIngress renders per-service configs from live inventory.
type agentNodeRunner struct { type agentNodeRunner struct {
q *sqlcdb.Queries q *sqlcdb.Queries
db *database.DB db *database.DB
@@ -56,11 +57,18 @@ func (r *agentNodeRunner) ReconcileOnNode(ctx context.Context, spec deployment.R
// sorted by id. Replicas wrap when they outnumber nodes. The local host is // sorted by id. Replicas wrap when they outnumber nodes. The local host is
// not a spread target — pin "local" or leave unset for local execution. // not a spread target — pin "local" or leave unset for local execution.
func (r *agentNodeRunner) ReconcileSpread(ctx context.Context, spec deployment.RuntimeSpec) (*deployment.RuntimeState, error) { func (r *agentNodeRunner) ReconcileSpread(ctx context.Context, spec deployment.RuntimeSpec) (*deployment.RuntimeState, error) {
agents, err := r.q.ListSchedulableAgents(ctx) tagsRaw, _ := json.Marshal(spec.PlacementTags)
if len(spec.PlacementTags) == 0 {
tagsRaw = []byte("[]")
}
agents, err := r.q.ListSchedulableAgentsMatching(ctx, json.RawMessage(tagsRaw))
if err != nil { if err != nil {
return nil, fmt.Errorf("list schedulable nodes: %w", err) return nil, fmt.Errorf("list schedulable nodes: %w", err)
} }
if len(agents) == 0 { if len(agents) == 0 {
if len(spec.PlacementTags) > 0 {
return nil, fmt.Errorf("no online schedulable nodes carry placement tags %v — tag a node or relax the requirement", spec.PlacementTags)
}
return nil, fmt.Errorf("no online schedulable nodes — connect an agent or unset spread") return nil, fmt.Errorf("no online schedulable nodes — connect an agent or unset spread")
} }
return r.reconcileOnAgents(ctx, spec, func(i int) string { return r.reconcileOnAgents(ctx, spec, func(i int) string {
@@ -68,6 +76,22 @@ func (r *agentNodeRunner) ReconcileSpread(ctx context.Context, spec deployment.R
}) })
} }
// ReconcileAuto resolves the least-loaded online agent matching the
// service's placement tags and reconciles every replica onto it. Stale
// replicas on other agents retire through reconcileOnAgents — a service
// follows its tags when the matching set changes.
func (r *agentNodeRunner) ReconcileAuto(ctx context.Context, spec deployment.RuntimeSpec) (*deployment.RuntimeState, error) {
tagsRaw, _ := json.Marshal(spec.PlacementTags)
if len(spec.PlacementTags) == 0 {
tagsRaw = []byte("[]")
}
agentID, err := r.q.PickLeastLoadedAgentMatching(ctx, json.RawMessage(tagsRaw))
if err != nil {
return nil, fmt.Errorf("no online nodes carry placement tags %v", spec.PlacementTags)
}
return r.reconcileOnAgents(ctx, spec, func(int) string { return agentID })
}
// RemoteRuntimeState reads the service's remote replicas from inventory. // RemoteRuntimeState reads the service's remote replicas from inventory.
// Rows only carry the last dispatched state — the agent doesn't stream // Rows only carry the last dispatched state — the agent doesn't stream
// container events — so "running" means "last create succeeded". // container events — so "running" means "last create succeeded".
@@ -107,10 +131,6 @@ func (r *agentNodeRunner) RemoteRuntimeState(ctx context.Context, serviceID stri
// targetFor and retires stale replicas on any agent — including leftovers // targetFor and retires stale replicas on any agent — including leftovers
// from a previous pin or a different spread assignment. // from a previous pin or a different spread assignment.
func (r *agentNodeRunner) reconcileOnAgents(ctx context.Context, spec deployment.RuntimeSpec, targetFor func(replica int) string) (*deployment.RuntimeState, error) { func (r *agentNodeRunner) reconcileOnAgents(ctx context.Context, spec deployment.RuntimeSpec, targetFor func(replica int) string) (*deployment.RuntimeState, error) {
if len(spec.Domains) > 0 || spec.Domain != "" {
return nil, fmt.Errorf("domains are routed by the local Traefik — remove the service's domains or run it on the local node")
}
replicas := spec.Replicas replicas := spec.Replicas
if replicas < 1 { if replicas < 1 {
replicas = 1 replicas = 1
@@ -127,7 +147,7 @@ func (r *agentNodeRunner) reconcileOnAgents(ctx context.Context, spec deployment
if keep { if keep {
continue continue
} }
if cmdErr := r.enqueueAndWait(ctx, row.NodeAgentID, "", "remove_container", if _, cmdErr := r.enqueueAndWait(ctx, row.NodeAgentID, "", "remove_container",
map[string]interface{}{"container_name": row.Name}); cmdErr != nil { map[string]interface{}{"container_name": row.Name}); cmdErr != nil {
return nil, fmt.Errorf("remove stale replica %s on node %s: %w", row.Name, row.NodeAgentID, cmdErr) return nil, fmt.Errorf("remove stale replica %s on node %s: %w", row.Name, row.NodeAgentID, cmdErr)
} }
@@ -161,18 +181,41 @@ func (r *agentNodeRunner) reconcileOnAgents(ctx context.Context, spec deployment
HealthCheck: rawJSON(map[string]interface{}{"path": spec.HealthPath}), HealthCheck: rawJSON(map[string]interface{}{"path": spec.HealthPath}),
}) })
if err := r.enqueueAndWait(ctx, agentID, containerID, "create_container", result, err := r.enqueueAndWait(ctx, agentID, containerID, "create_container",
map[string]interface{}{"container": container}); err != nil { map[string]interface{}{"container": container})
if err != nil {
return nil, fmt.Errorf("replica %d on node %s: %w", i, agentID, err) return nil, fmt.Errorf("replica %d on node %s: %w", i, agentID, err)
} }
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{ if assigned := reportedHostPorts(result); len(assigned) > 0 {
ID: containerID, for _, p := range ports {
Status: rawJSON(map[string]interface{}{"state": "running", "health": "none"}), pm, ok := p.(map[string]interface{})
}) if !ok {
continue
}
cport := jsonInt(pm["container_port"])
proto, _ := pm["protocol"].(string)
if hp, ok := assigned[fmt.Sprintf("%d/%s", cport, proto)]; ok {
if n, err := strconv.Atoi(hp); err == nil {
pm["host_port"] = n
}
}
}
_ = r.q.UpdateContainerPorts(ctx, sqlcdb.UpdateContainerPortsParams{
ID: containerID,
Status: rawJSON(map[string]interface{}{"state": "running", "health": "none"}),
Ports: rawJSON(ports),
})
} else {
_ = r.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: containerID,
Status: rawJSON(map[string]interface{}{"state": "running", "health": "none"}),
})
}
state.Containers = append(state.Containers, deployment.RuntimeContainer{ state.Containers = append(state.Containers, deployment.RuntimeContainer{
ID: containerID, Name: name, State: "running", Replica: i, ID: containerID, Name: name, State: "running", Replica: i,
}) })
} }
syncRemoteIngress(ctx, r.db, spec)
return state, nil return state, nil
} }
@@ -272,7 +315,17 @@ func (r *agentNodeRunner) ControlService(ctx context.Context, serviceID, agentID
if row.ServiceID != serviceID { if row.ServiceID != serviceID {
continue continue
} }
if err := r.enqueueAndWait(ctx, agentID, row.ID, cmdType, // Tombstoned replicas are already gone remotely — skip the dispatch
// or a fresh docker rm on a missing name reports a false failure.
if action == "remove" {
var st struct {
State string `json:"state"`
}
if err := json.Unmarshal(row.Status.RawMessage, &st); err == nil && st.State == "removed" {
continue
}
}
if _, err := r.enqueueAndWait(ctx, agentID, row.ID, cmdType,
map[string]interface{}{"container_name": row.Name}); err != nil { map[string]interface{}{"container_name": row.Name}); err != nil {
return fmt.Errorf("%s %s on node: %w", action, row.Name, err) return fmt.Errorf("%s %s on node: %w", action, row.Name, err)
} }
@@ -290,16 +343,38 @@ func (r *agentNodeRunner) RemoveService(ctx context.Context, serviceID, agentID
if err := r.ControlService(ctx, serviceID, agentID, "remove"); err != nil { if err := r.ControlService(ctx, serviceID, agentID, "remove"); err != nil {
return err return err
} }
return r.q.DeleteServiceContainersOnAgent(ctx, sqlcdb.DeleteServiceContainersOnAgentParams{ if err := r.q.DeleteServiceContainersOnAgent(ctx, sqlcdb.DeleteServiceContainersOnAgentParams{
NodeAgentID: agentID, NodeAgentID: agentID,
ServiceID: serviceID, ServiceID: serviceID,
}) }); err != nil {
return err
}
removeServiceIngress(serviceID)
return nil
}
// RetireService removes the service's remote replicas wherever they run —
// called when the service converges to local placement so a cleared pin or
// removed tag requirement doesn't leave containers behind.
func (r *agentNodeRunner) RetireService(ctx context.Context, serviceID string) error {
agentIDs, err := r.q.ListServiceAgents(ctx, serviceID)
if err != nil {
return err
}
for _, agentID := range agentIDs {
if err := r.RemoveService(ctx, serviceID, agentID); err != nil {
return fmt.Errorf("retire on node %s: %w", agentID, err)
}
}
removeServiceIngress(serviceID)
return nil
} }
// enqueueAndWait pushes a command to the agent's queue and polls for the // 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 — // reported result, returning the agent's result string on success. Agents
// it runs inside the deploy queue worker, never on a request path. // poll on their own interval (≥5s), so this blocks — it runs inside the
func (r *agentNodeRunner) enqueueAndWait(ctx context.Context, agentID, containerID, cmdType string, payload map[string]interface{}) error { // deploy queue worker, never on a request path.
func (r *agentNodeRunner) enqueueAndWait(ctx context.Context, agentID, containerID, cmdType string, payload map[string]interface{}) (string, error) {
var cid sql.NullString var cid sql.NullString
if containerID != "" { if containerID != "" {
cid = sql.NullString{String: containerID, Valid: true} cid = sql.NullString{String: containerID, Valid: true}
@@ -312,7 +387,7 @@ func (r *agentNodeRunner) enqueueAndWait(ctx context.Context, agentID, container
Payload: rawJSON(payload), Payload: rawJSON(payload),
}) })
if err != nil { if err != nil {
return err return "", err
} }
deadline := time.Now().Add(8 * time.Minute) // image pulls on a cold node deadline := time.Now().Add(8 * time.Minute) // image pulls on a cold node
@@ -324,36 +399,45 @@ func (r *agentNodeRunner) enqueueAndWait(ctx context.Context, agentID, container
if err == nil { if err == nil {
switch row.Status.String { switch row.Status.String {
case "completed": case "completed":
return nil return row.Result.String, nil
case "failed": case "failed":
if row.Error.Valid && row.Error.String != "" { if row.Error.Valid && row.Error.String != "" {
return fmt.Errorf("agent: %s", row.Error.String) return "", fmt.Errorf("agent: %s", row.Error.String)
} }
return fmt.Errorf("agent reported failure") return "", fmt.Errorf("agent reported failure")
} }
} }
if time.Now().After(deadline) { if time.Now().After(deadline) {
return fmt.Errorf("command %s timed out waiting for the agent", cmdType) return "", fmt.Errorf("command %s timed out waiting for the agent", cmdType)
} }
select { select {
case <-ctx.Done(): case <-ctx.Done():
return ctx.Err() return "", ctx.Err()
case <-time.After(2 * time.Second): case <-time.After(2 * time.Second):
} }
} }
} }
// resolveNodePin validates or resolves a node_id request value. "auto" picks // resolveNodePin validates or resolves a node_id request value. "auto" picks
// the least-loaded online agent; "" / "local" clears the pin. // the least-loaded online agent that carries every required placement tag;
func resolveNodePin(ctx context.Context, db *database.DB, requested string) (string, error) { // "" / "local" clears the pin. An explicit pin bypasses tag filtering —
// the operator is overriding placement on purpose.
func resolveNodePin(ctx context.Context, db *database.DB, requested string, requiredTags []string) (string, error) {
requested = strings.TrimSpace(requested) requested = strings.TrimSpace(requested)
if requested == "" || requested == "local" { if requested == "" || requested == "local" {
return "", nil return "", nil
} }
q := sqlcdb.New(db.DB) q := sqlcdb.New(db.DB)
if requested == "auto" { if requested == "auto" {
id, err := q.PickLeastLoadedAgent(ctx) tagsRaw, _ := json.Marshal(requiredTags)
if len(requiredTags) == 0 {
tagsRaw = []byte("[]")
}
id, err := q.PickLeastLoadedAgentMatching(ctx, json.RawMessage(tagsRaw))
if err != nil { if err != nil {
if len(requiredTags) > 0 {
return "", fmt.Errorf("no online nodes carry placement tags %v", requiredTags)
}
return "", fmt.Errorf("no online nodes available") return "", fmt.Errorf("no online nodes available")
} }
return id, nil return id, nil
+1
View File
@@ -285,6 +285,7 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg
// Wake callback for sleep-mode placeholders. Unauthenticated but // Wake callback for sleep-mode placeholders. Unauthenticated but
// non-destructive: it only reconciles the service back to running. // non-destructive: it only reconciles the service back to running.
public.Any("/internal/wake/:id", handleServiceWake) public.Any("/internal/wake/:id", handleServiceWake)
public.Any("/internal/wake-page/:id", handleServiceWakePage)
} }
// Read routes — public browsing. OptionalAuth resolves a session when // Read routes — public browsing. OptionalAuth resolves a session when
+1
View File
@@ -94,6 +94,7 @@ func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpe
RestartPolicy: service.RestartPolicy, RestartPolicy: service.RestartPolicy,
NodeID: serviceNodeID(db, service.ID), NodeID: serviceNodeID(db, service.ID),
Spread: serviceSpread(db, service.ID), Spread: serviceSpread(db, service.ID),
PlacementTags: servicePlacementTags(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
+81 -11
View File
@@ -55,9 +55,12 @@ type Service struct {
NodeName string `json:"node_name,omitempty" db:"-"` NodeName string `json:"node_name,omitempty" db:"-"`
// Spread distributes replicas across every online, schedulable agent. // Spread distributes replicas across every online, schedulable agent.
// Mutually exclusive with an explicit NodeID pin. // Mutually exclusive with an explicit NodeID pin.
Spread bool `json:"spread" db:"-"` Spread bool `json:"spread" db:"-"`
CreatedAt time.Time `json:"created_at" db:"created_at"` // PlacementTags restrict auto/spread candidates to agents carrying all
UpdatedAt time.Time `json:"updated_at" db:"updated_at"` // listed tags. An explicit pin overrides them — pin + tags is rejected.
PlacementTags []string `json:"placement_tags,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
@@ -91,12 +94,51 @@ func serviceSpread(db *database.DB, serviceID uuid.UUID) bool {
return false return false
} }
// loadServiceNode fills the pin + spread + the agent's display name for // servicePlacementTags reads the placement_tags column with the lazy pattern.
// responses. func servicePlacementTags(db *database.DB, serviceID uuid.UUID) []string {
var raw []byte
if err := db.QueryRow(`SELECT placement_tags FROM services WHERE id = $1`, serviceID).Scan(&raw); err != nil {
return nil
}
var tags []string
if err := json.Unmarshal(raw, &tags); err != nil {
return nil
}
return tags
}
// normalizePlacementTags trims, lowercases, and dedupes tag input. Tags are
// matched verbatim against node_agents.tags.
func normalizePlacementTags(tags []string) []string {
seen := map[string]bool{}
out := make([]string, 0, len(tags))
for _, t := range tags {
t = strings.ToLower(strings.TrimSpace(t))
if t == "" || seen[t] {
continue
}
seen[t] = true
out = append(out, t)
}
return out
}
// tagsJSON marshals a tag list for a jsonb column; empty stays an array, not null.
func tagsJSON(tags []string) []byte {
if len(tags) == 0 {
return []byte("[]")
}
raw, _ := json.Marshal(tags)
return raw
}
// loadServiceNode fills the pin + spread + placement tags + the agent's
// display name for responses.
func loadServiceNode(db *database.DB, s *Service) { func loadServiceNode(db *database.DB, s *Service) {
var name sql.NullString var name sql.NullString
s.NodeID = serviceNodeID(db, s.ID) s.NodeID = serviceNodeID(db, s.ID)
s.Spread = serviceSpread(db, s.ID) s.Spread = serviceSpread(db, s.ID)
s.PlacementTags = servicePlacementTags(db, s.ID)
if s.NodeID == "" { if s.NodeID == "" {
return return
} }
@@ -188,6 +230,9 @@ type CreateServiceRequest struct {
// Spread distributes replicas across every online, schedulable agent. // Spread distributes replicas across every online, schedulable agent.
// Mutually exclusive with an explicit node pin. // Mutually exclusive with an explicit node pin.
Spread *bool `json:"spread"` Spread *bool `json:"spread"`
// PlacementTags restrict auto/spread candidates to agents carrying all
// listed tags. Mutually exclusive with an explicit node pin.
PlacementTags []string `json:"placement_tags"`
} }
// UpdateServiceRequest represents a request to update a service // UpdateServiceRequest represents a request to update a service
@@ -222,6 +267,10 @@ type UpdateServiceRequest struct {
// Spread toggles multi-node replica distribution; pointer so false is // Spread toggles multi-node replica distribution; pointer so false is
// a real update, not a missing field. // a real update, not a missing field.
Spread *bool `json:"spread"` Spread *bool `json:"spread"`
// PlacementTags restrict auto/spread candidates to agents carrying all
// listed tags; pointer so [] clears the requirement. Mutually exclusive
// with an explicit node pin.
PlacementTags *[]string `json:"placement_tags"`
} }
// handleGetServices retrieves all services for a project // handleGetServices retrieves all services for a project
@@ -465,12 +514,17 @@ func handleCreateService(c *gin.Context) {
if req.Spread != nil { if req.Spread != nil {
service.Spread = *req.Spread service.Spread = *req.Spread
} }
service.PlacementTags = normalizePlacementTags(req.PlacementTags)
if req.NodeID != "" { if req.NodeID != "" {
if service.Spread && req.NodeID != "auto" { if service.Spread && req.NodeID != "auto" && req.NodeID != "local" {
c.JSON(http.StatusBadRequest, gin.H{"error": "spread and an explicit node pin are mutually exclusive", "code": "VALIDATION"}) c.JSON(http.StatusBadRequest, gin.H{"error": "spread and an explicit node pin are mutually exclusive", "code": "VALIDATION"})
return return
} }
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), req.NodeID) if len(service.PlacementTags) > 0 && req.NodeID != "auto" {
c.JSON(http.StatusBadRequest, gin.H{"error": "placement tags only apply to remote placement — set node_id=\"auto\" or omit it", "code": "VALIDATION"})
return
}
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), req.NodeID, service.PlacementTags)
if nodeErr != nil { if nodeErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"}) c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
return return
@@ -488,9 +542,9 @@ func handleCreateService(c *gin.Context) {
git_repo, git_branch, build_path, cpu, memory, replicas, port, domain, git_repo, git_branch, build_path, cpu, memory, replicas, port, domain,
healthcheck_path, restart_policy, volumes, healthcheck_path, restart_policy, volumes,
builder, cpu_reserve, memory_reserve, static_build_cmd, static_dir, builder, cpu_reserve, memory_reserve, static_build_cmd, static_dir,
node_id, spread, created_at, updated_at) node_id, spread, placement_tags, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26,
$27, $28, $29, $30, $31, $32, $33, $34, $35)`, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36)`,
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,
@@ -500,6 +554,7 @@ func handleCreateService(c *gin.Context) {
service.StaticBuildCmd, service.StaticDir, service.StaticBuildCmd, service.StaticDir,
sql.NullString{String: service.NodeID, Valid: service.NodeID != ""}, sql.NullString{String: service.NodeID, Valid: service.NodeID != ""},
service.Spread, service.Spread,
tagsJSON(service.PlacementTags),
service.CreatedAt, service.UpdatedAt, service.CreatedAt, service.UpdatedAt,
) )
@@ -807,13 +862,22 @@ func handleUpdateService(c *gin.Context) {
} else { } else {
existingService.Spread = serviceSpread(db.(*database.DB), serviceID) existingService.Spread = serviceSpread(db.(*database.DB), serviceID)
} }
if req.PlacementTags != nil {
existingService.PlacementTags = normalizePlacementTags(*req.PlacementTags)
} else {
existingService.PlacementTags = servicePlacementTags(db.(*database.DB), serviceID)
}
if req.NodeID != nil { if req.NodeID != nil {
nodeValue := strings.TrimSpace(*req.NodeID) nodeValue := strings.TrimSpace(*req.NodeID)
if existingService.Spread && nodeValue != "" && nodeValue != "auto" && nodeValue != "local" { if existingService.Spread && nodeValue != "" && nodeValue != "auto" && nodeValue != "local" {
c.JSON(http.StatusBadRequest, gin.H{"error": "spread and an explicit node pin are mutually exclusive — set spread=false first", "code": "VALIDATION"}) c.JSON(http.StatusBadRequest, gin.H{"error": "spread and an explicit node pin are mutually exclusive — set spread=false first", "code": "VALIDATION"})
return return
} }
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), nodeValue) if len(existingService.PlacementTags) > 0 && nodeValue != "" && nodeValue != "auto" {
c.JSON(http.StatusBadRequest, gin.H{"error": "service has placement tags — clear them (placement_tags=[]) before pinning or going local", "code": "VALIDATION"})
return
}
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), nodeValue, existingService.PlacementTags)
if nodeErr != nil { if nodeErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"}) c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
return return
@@ -826,6 +890,10 @@ func handleUpdateService(c *gin.Context) {
c.JSON(http.StatusBadRequest, gin.H{"error": "service already has an explicit node pin — clear it (node_id=\"local\") before enabling spread", "code": "VALIDATION"}) c.JSON(http.StatusBadRequest, gin.H{"error": "service already has an explicit node pin — clear it (node_id=\"local\") before enabling spread", "code": "VALIDATION"})
return return
} }
if len(existingService.PlacementTags) > 0 && existingService.NodeID != "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "service already has an explicit node pin — clear it (node_id=\"local\") before setting placement tags", "code": "VALIDATION"})
return
}
existingService.UpdatedAt = time.Now() existingService.UpdatedAt = time.Now()
@@ -838,7 +906,8 @@ func handleUpdateService(c *gin.Context) {
restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16, restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16,
builder = $19, cpu_reserve = $20, memory_reserve = $21, builder = $19, cpu_reserve = $20, memory_reserve = $21,
static_build_cmd = $22, static_dir = $23, static_build_cmd = $22, static_dir = $23,
sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26, spread = $27 sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26, spread = $27,
placement_tags = $28::jsonb
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,
@@ -851,6 +920,7 @@ func handleUpdateService(c *gin.Context) {
existingService.SleepEnabled, existingService.SleepIdleMinutes, existingService.SleepEnabled, existingService.SleepIdleMinutes,
sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""}, sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""},
existingService.Spread, existingService.Spread,
tagsJSON(existingService.PlacementTags),
) )
if err != nil { if err != nil {
+108 -10
View File
@@ -13,6 +13,7 @@ import (
"time" "time"
"containr/internal/database" "containr/internal/database"
"containr/internal/database/sqlcdb"
"containr/internal/deployment" "containr/internal/deployment"
"containr/internal/docker" "containr/internal/docker"
@@ -100,7 +101,20 @@ func sweepSleeping(ctx context.Context, db *database.DB, dc *docker.Client, engi
} }
rows.Close() rows.Close()
q := sqlcdb.New(db.DB)
for _, svc := range candidates { for _, svc := range candidates {
// Remote replicas live in inventory, not the local docker host —
// idleness comes from heartbeat-reported net counters instead.
if remote := remoteLiveContainers(ctx, q, svc.ID.String()); len(remote) > 0 {
var total uint64
for _, row := range remote {
total += rowNetBytes(row)
}
if markIdle(svc.ID, total, time.Duration(svc.IdleMinutes)*time.Minute) {
putToSleepRemote(ctx, db, q, svc, apiPort)
}
continue
}
containers, err := engine.ListServiceContainers(ctx, svc.ID.String()) containers, err := engine.ListServiceContainers(ctx, svc.ID.String())
if err != nil || len(containers) == 0 { if err != nil || len(containers) == 0 {
continue continue
@@ -125,6 +139,65 @@ func sweepSleeping(ctx context.Context, db *database.DB, dc *docker.Client, engi
} }
} }
// remoteLiveContainers returns inventory rows whose replicas are alive on
// node agents — the signal that idleness must be measured remotely.
func remoteLiveContainers(ctx context.Context, q *sqlcdb.Queries, serviceID string) []sqlcdb.ListServiceContainersRow {
rows, err := q.ListServiceContainers(ctx, serviceID)
if err != nil {
return nil
}
live := rows[:0]
for _, row := range rows {
var st struct {
State string `json:"state"`
}
if err := json.Unmarshal(row.Status.RawMessage, &st); err == nil &&
st.State != "" && st.State != "removed" && st.State != "sleeping" {
live = append(live, row)
}
}
return live
}
// rowNetBytes reads the cumulative rx+tx the agent reported at its last
// heartbeat — 0 for rows written before the counter existed.
func rowNetBytes(row sqlcdb.ListServiceContainersRow) uint64 {
var res struct {
NetBytes uint64 `json:"net_bytes"`
}
_ = json.Unmarshal(row.Resources.RawMessage, &res)
return res.NetBytes
}
// putToSleepRemote removes the service's remote replicas via their agents,
// marks the service sleeping, and repoints the domain route at the wake
// endpoint so inbound traffic brings it back.
func putToSleepRemote(ctx context.Context, db *database.DB, q *sqlcdb.Queries, svc sleepCandidate, apiPort int) {
runner := newAgentNodeRunner(db)
for _, agentID := range remoteAgentsForService(ctx, db, svc.ID) {
if err := runner.ControlService(ctx, svc.ID.String(), agentID, "remove"); err != nil {
log.Printf("sleep: remote remove on %s failed: %v", agentID, err)
continue
}
_ = q.DeleteServiceContainersOnAgent(ctx, sqlcdb.DeleteServiceContainersOnAgentParams{
NodeAgentID: agentID,
ServiceID: svc.ID.String(),
})
}
tracker.mu.Lock()
delete(tracker.activity, svc.ID)
tracker.mu.Unlock()
_, _ = db.Exec(`UPDATE services SET status = 'sleeping', updated_at = $1 WHERE id = $2`, time.Now(), svc.ID)
log.Printf("sleep: %s (%s) remote replicas removed after idle timeout", svc.Name, svc.ID)
var domain string
_ = db.QueryRow(`SELECT COALESCE(domain,'') FROM services WHERE id = $1`, svc.ID).Scan(&domain)
domains := serviceDomainNames(db, svc.ID, domain)
if len(domains) > 0 {
writeWakeIngress(svc.ID.String(), domains, apiPort)
}
}
// markIdle returns true when the network counter stopped growing past the // markIdle returns true when the network counter stopped growing past the
// timeout. First sighting only sets the baseline — no idle without history. // timeout. First sighting only sets the baseline — no idle without history.
func markIdle(id uuid.UUID, bytes uint64, timeout time.Duration) bool { func markIdle(id uuid.UUID, bytes uint64, timeout time.Duration) bool {
@@ -346,18 +419,40 @@ func pruneWakeContainers(ctx context.Context, db *database.DB, dc *docker.Client
// idempotent and non-destructive, so no token is required; the endpoint // idempotent and non-destructive, so no token is required; the endpoint
// never mutates configuration. // never mutates configuration.
func handleServiceWake(c *gin.Context) { func handleServiceWake(c *gin.Context) {
code, body := triggerServiceWake(c)
if code != 0 {
c.JSON(code, body)
}
}
// handleServiceWakePage is the remote ingress wake target — Traefik routes
// a sleeping remote service's domain here via the file provider. Every hit
// (re)triggers the wake reconcile and returns a reloading page.
func handleServiceWakePage(c *gin.Context) {
triggerServiceWake(c)
c.Header("Content-Type", "text/html; charset=utf-8")
c.String(http.StatusOK, wakePageHTML)
}
const wakePageHTML = `<!doctype html><html><head><meta http-equiv="refresh" content="4"><title>Waking up</title>` +
`<style>body{font-family:system-ui,sans-serif;background:#0b0d12;color:#dde2ea;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0}` +
`.card{text-align:center;max-width:34rem;padding:2rem}h1{font-size:1.25rem;font-weight:600}p{color:#98a2b3}</style></head>` +
`<body><div class="card"><h1>Service is waking up</h1><p>This Containr service was sleeping and is resuming now. The page reloads automatically.</p></div></body></html>`
// triggerServiceWake loads the service and reconciles it back to running
// in the background. Returns (0, nil) when a response was already written
// by an embedded handler path, else the JSON status to send.
func triggerServiceWake(c *gin.Context) (int, gin.H) {
dbVal, _ := c.Get("db") dbVal, _ := c.Get("db")
db := dbVal.(*database.DB) db := dbVal.(*database.DB)
engineVal, _ := c.Get("deployment_engine") engineVal, _ := c.Get("deployment_engine")
engine, _ := engineVal.(*deployment.DeploymentEngine) engine, _ := engineVal.(*deployment.DeploymentEngine)
if engine == nil { if engine == nil {
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "deployment engine unavailable"}) return http.StatusServiceUnavailable, gin.H{"error": "deployment engine unavailable"}
return
} }
id, err := uuid.Parse(c.Param("id")) id, err := uuid.Parse(c.Param("id"))
if err != nil { if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid service id"}) return http.StatusBadRequest, gin.H{"error": "invalid service id"}
return
} }
var service Service var service Service
err = db.QueryRow( err = db.QueryRow(
@@ -385,12 +480,10 @@ func handleServiceWake(c *gin.Context) {
&service.CreatedAt, &service.UpdatedAt, &service.CreatedAt, &service.UpdatedAt,
) )
if err != nil { if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "service not found"}) return http.StatusNotFound, gin.H{"error": "service not found"}
return
} }
if service.Status == "running" || service.Status == "deployed" { if service.Status == "running" || service.Status == "deployed" {
c.JSON(http.StatusOK, gin.H{"status": service.Status}) return http.StatusOK, gin.H{"status": service.Status}
return
} }
go func() { go func() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
@@ -411,7 +504,7 @@ func handleServiceWake(c *gin.Context) {
} }
_, _ = db.Exec(`UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, status, time.Now(), service.ID) _, _ = db.Exec(`UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, status, time.Now(), service.ID)
}() }()
c.JSON(http.StatusAccepted, gin.H{"status": "waking"}) return http.StatusAccepted, gin.H{"status": "waking"}
} }
// handleServiceSleep puts a service to sleep immediately (owner+ only). // handleServiceSleep puts a service to sleep immediately (owner+ only).
@@ -438,7 +531,12 @@ func handleServiceSleep(c *gin.Context) {
Name: service.Name, Name: service.Name,
Status: service.Status, Status: service.Status,
} }
go putToSleep(context.Background(), db, dc, engine, svc, apiPortFromContext(c)) q := sqlcdb.New(db.DB)
if len(remoteLiveContainers(c.Request.Context(), q, service.ID.String())) > 0 {
go putToSleepRemote(context.Background(), db, q, svc, apiPortFromContext(c))
} else {
go putToSleep(context.Background(), db, dc, engine, svc, apiPortFromContext(c))
}
c.JSON(http.StatusAccepted, gin.H{"status": "sleeping"}) c.JSON(http.StatusAccepted, gin.H{"status": "sleeping"})
} }
+14 -2
View File
@@ -2,6 +2,7 @@ package commands
import ( import (
"fmt" "fmt"
"strings"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
@@ -83,7 +84,7 @@ var nodesDeleteCmd = &cobra.Command{
var nodesUpdateCmd = &cobra.Command{ var nodesUpdateCmd = &cobra.Command{
Use: "update <id>", Use: "update <id>",
Short: "Update a node agent (--name, --auto-prune)", Short: "Update a node agent (--name, --auto-prune, --tags)",
Args: cobra.ExactArgs(1), Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
c, err := client() c, err := client()
@@ -97,8 +98,18 @@ var nodesUpdateCmd = &cobra.Command{
if cmd.Flags().Changed("auto-prune") { if cmd.Flags().Changed("auto-prune") {
body["auto_prune"], _ = cmd.Flags().GetBool("auto-prune") body["auto_prune"], _ = cmd.Flags().GetBool("auto-prune")
} }
if cmd.Flags().Changed("tags") {
raw, _ := cmd.Flags().GetString("tags")
tags := []string{}
for _, t := range strings.Split(raw, ",") {
if t = strings.TrimSpace(t); t != "" {
tags = append(tags, t)
}
}
body["tags"] = tags
}
if len(body) == 0 { if len(body) == 0 {
return &APIError{Message: "nothing to update — pass --name or --auto-prune", ExitCode: ExitError} return &APIError{Message: "nothing to update — pass --name, --auto-prune or --tags", ExitCode: ExitError}
} }
data, err := c.Do("PUT", "/agents/"+args[0], body) data, err := c.Do("PUT", "/agents/"+args[0], body)
if err != nil { if err != nil {
@@ -292,6 +303,7 @@ var nodeTokensRevokeCmd = &cobra.Command{
func init() { func init() {
nodesUpdateCmd.Flags().String("name", "", "node display name") nodesUpdateCmd.Flags().String("name", "", "node display name")
nodesUpdateCmd.Flags().Bool("auto-prune", false, "enqueue a daily bounded docker prune") nodesUpdateCmd.Flags().Bool("auto-prune", false, "enqueue a daily bounded docker prune")
nodesUpdateCmd.Flags().String("tags", "", "comma-separated placement tags (services with placement_tags only schedule onto matching nodes)")
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)
@@ -117,6 +117,15 @@ var servicesCreateCmd = &cobra.Command{
if v, _ := cmd.Flags().GetBool("spread"); v { if v, _ := cmd.Flags().GetBool("spread"); v {
body["spread"] = v body["spread"] = v
} }
if v, _ := cmd.Flags().GetString("placement-tags"); v != "" {
tags := []string{}
for _, t := range strings.Split(v, ",") {
if t = strings.TrimSpace(t); t != "" {
tags = append(tags, t)
}
}
body["placement_tags"] = tags
}
data, err := c.Do("POST", "/projects/"+args[0]+"/services", body) data, err := c.Do("POST", "/projects/"+args[0]+"/services", body)
if err != nil { if err != nil {
return err return err
@@ -210,6 +219,16 @@ var servicesUpdateCmd = &cobra.Command{
v, _ := cmd.Flags().GetBool("spread") v, _ := cmd.Flags().GetBool("spread")
body["spread"] = v body["spread"] = v
} }
if cmd.Flags().Changed("placement-tags") {
v, _ := cmd.Flags().GetString("placement-tags")
tags := []string{}
for _, t := range strings.Split(v, ",") {
if t = strings.TrimSpace(t); t != "" {
tags = append(tags, t)
}
}
body["placement_tags"] = tags
}
if cmd.Flags().Changed("sleep-idle") { if cmd.Flags().Changed("sleep-idle") {
v, _ := cmd.Flags().GetInt("sleep-idle") v, _ := cmd.Flags().GetInt("sleep-idle")
body["sleep_idle_minutes"] = v body["sleep_idle_minutes"] = v
@@ -503,6 +522,7 @@ func init() {
f.String("static-dir", "", "output dir for the static builder (default dist)") f.String("static-dir", "", "output dir for the static builder (default dist)")
f.String("node", "", "pin to a node agent id, or 'auto' for least-loaded (image services only)") f.String("node", "", "pin to a node agent id, or 'auto' for least-loaded (image services only)")
f.Bool("spread", false, "distribute replicas across all online schedulable nodes") f.Bool("spread", false, "distribute replicas across all online schedulable nodes")
f.String("placement-tags", "", "comma-separated node tags required for remote placement (auto/spread only)")
uf := servicesUpdateCmd.Flags() uf := servicesUpdateCmd.Flags()
uf.String("name", "", "service name") uf.String("name", "", "service name")
@@ -529,6 +549,7 @@ func init() {
uf.Int("sleep-idle", 0, "idle minutes before sleeping (1-1440)") uf.Int("sleep-idle", 0, "idle minutes before sleeping (1-1440)")
uf.String("node", "", "pin to a node agent id, 'auto' for least-loaded, 'local' to clear") uf.String("node", "", "pin to a node agent id, 'auto' for least-loaded, 'local' to clear")
uf.Bool("spread", false, "spread replicas across all online schedulable nodes (--spread=false to disable)") uf.Bool("spread", false, "spread replicas across all online schedulable nodes (--spread=false to disable)")
uf.String("placement-tags", "", "comma-separated node tags required for remote placement (empty clears)")
for _, a := range []string{"start", "stop", "restart", "redeploy", "sleep", "wake"} { for _, a := range []string{"start", "stop", "restart", "redeploy", "sleep", "wake"} {
verb := a verb := a
@@ -75,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, schedulable RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable, tags
` `
type CreateAgentParams struct { type CreateAgentParams struct {
@@ -123,6 +123,7 @@ func (q *Queries) CreateAgent(ctx context.Context, arg CreateAgentParams) (NodeA
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable, &i.Schedulable,
&i.Tags,
) )
return i, err return i, err
} }
@@ -257,7 +258,7 @@ func (q *Queries) DeleteServiceContainersOnAgent(ctx context.Context, arg Delete
} }
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, schedulable 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, tags 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) {
@@ -279,12 +280,13 @@ func (q *Queries) GetAgent(ctx context.Context, id string) (NodeAgent, error) {
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable, &i.Schedulable,
&i.Tags,
) )
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, schedulable FROM node_agents SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable, tags FROM node_agents
WHERE hostname = $1 AND ip_address = $2 WHERE hostname = $1 AND ip_address = $2
LIMIT 1 LIMIT 1
` `
@@ -313,6 +315,7 @@ func (q *Queries) GetAgentByHostAndIP(ctx context.Context, arg GetAgentByHostAnd
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable, &i.Schedulable,
&i.Tags,
) )
return i, err return i, err
} }
@@ -520,7 +523,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, schedulable 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, tags 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) {
@@ -548,6 +551,7 @@ func (q *Queries) ListAgents(ctx context.Context) ([]NodeAgent, error) {
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable, &i.Schedulable,
&i.Tags,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@@ -725,8 +729,72 @@ func (q *Queries) ListSchedulableAgents(ctx context.Context) ([]ListSchedulableA
return items, nil return items, nil
} }
const listSchedulableAgentsMatching = `-- name: ListSchedulableAgentsMatching :many
SELECT id, name FROM node_agents
WHERE status = 'online' AND schedulable
AND tags @> $1::jsonb
ORDER BY id
`
type ListSchedulableAgentsMatchingRow struct {
ID string `json:"id"`
Name string `json:"name"`
}
// Only agents carrying every required placement tag (jsonb array containment).
func (q *Queries) ListSchedulableAgentsMatching(ctx context.Context, dollar_1 json.RawMessage) ([]ListSchedulableAgentsMatchingRow, error) {
rows, err := q.db.QueryContext(ctx, listSchedulableAgentsMatching, dollar_1)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListSchedulableAgentsMatchingRow{}
for rows.Next() {
var i ListSchedulableAgentsMatchingRow
if err := rows.Scan(&i.ID, &i.Name); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listServiceAgents = `-- name: ListServiceAgents :many
SELECT DISTINCT node_agent_id FROM container_instances WHERE service_id = $1
`
// Distinct agents holding inventory rows for a service.
func (q *Queries) ListServiceAgents(ctx context.Context, serviceID string) ([]string, error) {
rows, err := q.db.QueryContext(ctx, listServiceAgents, serviceID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []string{}
for rows.Next() {
var node_agent_id string
if err := rows.Scan(&node_agent_id); err != nil {
return nil, err
}
items = append(items, node_agent_id)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listServiceContainers = `-- name: ListServiceContainers :many const listServiceContainers = `-- name: ListServiceContainers :many
SELECT id, name, node_agent_id, status FROM container_instances SELECT id, name, node_agent_id, status, ports, resources FROM container_instances
WHERE service_id = $1 WHERE service_id = $1
` `
@@ -735,6 +803,8 @@ type ListServiceContainersRow struct {
Name string `json:"name"` Name string `json:"name"`
NodeAgentID string `json:"node_agent_id"` NodeAgentID string `json:"node_agent_id"`
Status pqtype.NullRawMessage `json:"status"` Status pqtype.NullRawMessage `json:"status"`
Ports pqtype.NullRawMessage `json:"ports"`
Resources pqtype.NullRawMessage `json:"resources"`
} }
func (q *Queries) ListServiceContainers(ctx context.Context, serviceID string) ([]ListServiceContainersRow, error) { func (q *Queries) ListServiceContainers(ctx context.Context, serviceID string) ([]ListServiceContainersRow, error) {
@@ -751,6 +821,8 @@ func (q *Queries) ListServiceContainers(ctx context.Context, serviceID string) (
&i.Name, &i.Name,
&i.NodeAgentID, &i.NodeAgentID,
&i.Status, &i.Status,
&i.Ports,
&i.Resources,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@@ -794,6 +866,36 @@ func (q *Queries) PickLeastLoadedAgent(ctx context.Context) (string, error) {
return id, err return id, err
} }
const pickLeastLoadedAgentMatching = `-- name: PickLeastLoadedAgentMatching :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
AND a.tags @> $1::jsonb
ORDER BY
COALESCE(
(a.resources->'memory'->>'used')::double precision
/ NULLIF((a.resources->'memory'->>'total')::double precision, 0),
1.0
) ASC,
COALESCE((a.resources->'cpu'->>'usage')::double precision, 100.0) ASC,
COALESCE(h.container_count, 0) ASC,
a.created_at ASC
LIMIT 1
`
// Same resource-aware ordering as PickLeastLoadedAgent, restricted to agents
// carrying every required placement tag.
func (q *Queries) PickLeastLoadedAgentMatching(ctx context.Context, dollar_1 json.RawMessage) (string, error) {
row := q.db.QueryRowContext(ctx, pickLeastLoadedAgentMatching, dollar_1)
var id string
err := row.Scan(&id)
return id, err
}
const scrubCommandPayload = `-- name: ScrubCommandPayload :exec const scrubCommandPayload = `-- name: ScrubCommandPayload :exec
UPDATE agent_commands SET payload = $2, updated_at = NOW() WHERE id = $1 UPDATE agent_commands SET payload = $2, updated_at = NOW() WHERE id = $1
` `
@@ -822,6 +924,20 @@ func (q *Queries) SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulab
return err return err
} }
const setAgentTags = `-- name: SetAgentTags :exec
UPDATE node_agents SET tags = $2::jsonb, updated_at = NOW() WHERE id = $1
`
type SetAgentTagsParams struct {
ID string `json:"id"`
Column2 json.RawMessage `json:"column_2"`
}
func (q *Queries) SetAgentTags(ctx context.Context, arg SetAgentTagsParams) error {
_, err := q.db.ExecContext(ctx, setAgentTags, arg.ID, arg.Column2)
return err
}
const updateAgent = `-- name: UpdateAgent :one const updateAgent = `-- name: UpdateAgent :one
UPDATE node_agents SET UPDATE node_agents SET
name = $2, name = $2,
@@ -838,7 +954,7 @@ UPDATE node_agents SET
schedulable = $13, 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, schedulable RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, schedulable, tags
` `
type UpdateAgentParams struct { type UpdateAgentParams struct {
@@ -890,6 +1006,7 @@ func (q *Queries) UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeA
&i.Metadata, &i.Metadata,
&i.AutoPrune, &i.AutoPrune,
&i.Schedulable, &i.Schedulable,
&i.Tags,
) )
return i, err return i, err
} }
@@ -920,6 +1037,53 @@ func (q *Queries) UpdateAgentHeartbeat(ctx context.Context, arg UpdateAgentHeart
return err return err
} }
const updateContainerPorts = `-- name: UpdateContainerPorts :exec
UPDATE container_instances SET status = $2, ports = $3, updated_at = NOW()
WHERE id = $1
`
type UpdateContainerPortsParams struct {
ID string `json:"id"`
Status pqtype.NullRawMessage `json:"status"`
Ports pqtype.NullRawMessage `json:"ports"`
}
// Agent reports the real host bindings after create — persist them with
// the status flip so ingress can route to the assigned port.
func (q *Queries) UpdateContainerPorts(ctx context.Context, arg UpdateContainerPortsParams) error {
_, err := q.db.ExecContext(ctx, updateContainerPorts, arg.ID, arg.Status, arg.Ports)
return err
}
const updateContainerStateByName = `-- name: UpdateContainerStateByName :exec
UPDATE container_instances SET
status = jsonb_set(status, '{state}', $3::jsonb),
resources = jsonb_set(COALESCE(resources, '{}'::jsonb), '{net_bytes}', $4::jsonb),
updated_at = NOW()
WHERE name = $1 AND node_agent_id = $2
AND status->>'state' IS DISTINCT FROM 'removed'
`
type UpdateContainerStateByNameParams struct {
Name string `json:"name"`
NodeAgentID string `json:"node_agent_id"`
Column3 json.RawMessage `json:"column_3"`
Column4 json.RawMessage `json:"column_4"`
}
// Agent heartbeat reports real docker state + net counters; reconcile
// inventory for rows this node owns. Tombstones win — never resurrect a
// removed replica. $4 is the cumulative rx+tx byte counter.
func (q *Queries) UpdateContainerStateByName(ctx context.Context, arg UpdateContainerStateByNameParams) error {
_, err := q.db.ExecContext(ctx, updateContainerStateByName,
arg.Name,
arg.NodeAgentID,
arg.Column3,
arg.Column4,
)
return err
}
const updateContainerStatus = `-- name: UpdateContainerStatus :exec const updateContainerStatus = `-- name: UpdateContainerStatus :exec
UPDATE container_instances SET status = $2, updated_at = NOW() UPDATE container_instances SET status = $2, updated_at = NOW()
WHERE id = $1 WHERE id = $1
@@ -177,6 +177,7 @@ type NodeAgent struct {
Metadata pqtype.NullRawMessage `json:"metadata"` Metadata pqtype.NullRawMessage `json:"metadata"`
AutoPrune bool `json:"auto_prune"` AutoPrune bool `json:"auto_prune"`
Schedulable bool `json:"schedulable"` Schedulable bool `json:"schedulable"`
Tags json.RawMessage `json:"tags"`
} }
type Notification struct { type Notification struct {
@@ -277,6 +278,7 @@ type Service struct {
StaticDir string `json:"static_dir"` StaticDir string `json:"static_dir"`
NodeID sql.NullString `json:"node_id"` NodeID sql.NullString `json:"node_id"`
Spread bool `json:"spread"` Spread bool `json:"spread"`
PlacementTags json.RawMessage `json:"placement_tags"`
} }
type ServiceDomain struct { type ServiceDomain struct {
@@ -7,6 +7,7 @@ package sqlcdb
import ( import (
"context" "context"
"database/sql" "database/sql"
"encoding/json"
"github.com/google/uuid" "github.com/google/uuid"
) )
@@ -95,6 +96,10 @@ type Querier interface {
// platform admin. Anonymous callers pass user_id = uuid.Nil, is_admin = false. // platform admin. Anonymous callers pass user_id = uuid.Nil, is_admin = false.
ListProjectsWithStatsByUser(ctx context.Context, arg ListProjectsWithStatsByUserParams) ([]ListProjectsWithStatsByUserRow, error) ListProjectsWithStatsByUser(ctx context.Context, arg ListProjectsWithStatsByUserParams) ([]ListProjectsWithStatsByUserRow, error)
ListSchedulableAgents(ctx context.Context) ([]ListSchedulableAgentsRow, error) ListSchedulableAgents(ctx context.Context) ([]ListSchedulableAgentsRow, error)
// Only agents carrying every required placement tag (jsonb array containment).
ListSchedulableAgentsMatching(ctx context.Context, dollar_1 json.RawMessage) ([]ListSchedulableAgentsMatchingRow, error)
// Distinct agents holding inventory rows for a service.
ListServiceAgents(ctx context.Context, serviceID string) ([]string, error)
ListServiceContainers(ctx context.Context, serviceID string) ([]ListServiceContainersRow, error) ListServiceContainers(ctx context.Context, serviceID string) ([]ListServiceContainersRow, error)
ListServiceTemplatesByCategoryForUser(ctx context.Context, arg ListServiceTemplatesByCategoryForUserParams) ([]ServiceTemplate, error) ListServiceTemplatesByCategoryForUser(ctx context.Context, arg ListServiceTemplatesByCategoryForUserParams) ([]ServiceTemplate, error)
ListServiceTemplatesForUser(ctx context.Context, ownerID uuid.NullUUID) ([]ServiceTemplate, error) ListServiceTemplatesForUser(ctx context.Context, ownerID uuid.NullUUID) ([]ServiceTemplate, error)
@@ -107,10 +112,14 @@ type Querier interface {
// Resource-aware: lowest memory utilisation first, then cpu, then container // Resource-aware: lowest memory utilisation first, then cpu, then container
// count. Agents without telemetry sort last (NULL → worst score). // count. Agents without telemetry sort last (NULL → worst score).
PickLeastLoadedAgent(ctx context.Context) (string, error) PickLeastLoadedAgent(ctx context.Context) (string, error)
// Same resource-aware ordering as PickLeastLoadedAgent, restricted to agents
// carrying every required placement tag.
PickLeastLoadedAgentMatching(ctx context.Context, dollar_1 json.RawMessage) (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 ScrubCommandPayload(ctx context.Context, arg ScrubCommandPayloadParams) error
SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulableParams) error SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulableParams) error
SetAgentTags(ctx context.Context, arg SetAgentTagsParams) 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 +136,13 @@ type Querier interface {
UpdateAgentHeartbeat(ctx context.Context, arg UpdateAgentHeartbeatParams) error UpdateAgentHeartbeat(ctx context.Context, arg UpdateAgentHeartbeatParams) error
UpdateBackupTargetByIDAndUser(ctx context.Context, arg UpdateBackupTargetByIDAndUserParams) error UpdateBackupTargetByIDAndUser(ctx context.Context, arg UpdateBackupTargetByIDAndUserParams) error
UpdateBanner(ctx context.Context, arg UpdateBannerParams) (Banner, error) UpdateBanner(ctx context.Context, arg UpdateBannerParams) (Banner, error)
// Agent reports the real host bindings after create — persist them with
// the status flip so ingress can route to the assigned port.
UpdateContainerPorts(ctx context.Context, arg UpdateContainerPortsParams) error
// Agent heartbeat reports real docker state + net counters; reconcile
// inventory for rows this node owns. Tombstones win — never resurrect a
// removed replica. $4 is the cumulative rx+tx byte counter.
UpdateContainerStateByName(ctx context.Context, arg UpdateContainerStateByNameParams) error
UpdateContainerStatus(ctx context.Context, arg UpdateContainerStatusParams) error UpdateContainerStatus(ctx context.Context, arg UpdateContainerStatusParams) error
UpdateDatabaseServiceNameAndPlanByIDAndUser(ctx context.Context, arg UpdateDatabaseServiceNameAndPlanByIDAndUserParams) error UpdateDatabaseServiceNameAndPlanByIDAndUser(ctx context.Context, arg UpdateDatabaseServiceNameAndPlanByIDAndUserParams) error
UpdateDatabaseServiceNameByIDAndUser(ctx context.Context, arg UpdateDatabaseServiceNameByIDAndUserParams) error UpdateDatabaseServiceNameByIDAndUser(ctx context.Context, arg UpdateDatabaseServiceNameByIDAndUserParams) error
@@ -20,6 +20,10 @@ type NodeRunner interface {
// ReconcileSpread distributes the service's replicas across every online, // ReconcileSpread distributes the service's replicas across every online,
// schedulable agent — replica i lands on agents[i % len]. // schedulable agent — replica i lands on agents[i % len].
ReconcileSpread(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error) ReconcileSpread(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error)
// ReconcileAuto picks the least-loaded online agent carrying every tag in
// spec.PlacementTags and reconciles all replicas onto it. Re-resolved on
// every reconcile — the service follows the tag, not a stored node.
ReconcileAuto(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error)
// ControlService fans a lifecycle action ("start"|"stop"|"restart"|"remove") // ControlService fans a lifecycle action ("start"|"stop"|"restart"|"remove")
// out to every replica of the service on the agent. // out to every replica of the service on the agent.
ControlService(ctx context.Context, serviceID, agentID, action string) error ControlService(ctx context.Context, serviceID, agentID, action string) error
@@ -28,6 +32,9 @@ type NodeRunner interface {
// RemoteRuntimeState reads the service's remote replicas from inventory // RemoteRuntimeState reads the service's remote replicas from inventory
// across all agents. nil/empty means nothing runs remotely. // across all agents. nil/empty means nothing runs remotely.
RemoteRuntimeState(ctx context.Context, serviceID string) (*RuntimeState, error) RemoteRuntimeState(ctx context.Context, serviceID string) (*RuntimeState, error)
// RetireService removes every remote replica of the service across all
// agents — used when a service returns to local placement.
RetireService(ctx context.Context, serviceID string) error
} }
type DeploymentEngine struct { type DeploymentEngine struct {
@@ -60,6 +60,9 @@ type RuntimeSpec struct {
// Spread distributes replicas across all online, schedulable agents // Spread distributes replicas across all online, schedulable agents
// instead of a single node. Mutually exclusive with NodeID. // instead of a single node. Mutually exclusive with NodeID.
Spread bool Spread bool
// PlacementTags restricts auto/spread candidates to agents carrying
// every listed tag. An explicit NodeID pin bypasses tag filtering.
PlacementTags []string
} }
// RuntimeContainer describes one live replica. // RuntimeContainer describes one live replica.
@@ -171,6 +174,20 @@ func (de *DeploymentEngine) ReconcileService(ctx context.Context, spec RuntimeSp
} }
return de.nodeRunner.ReconcileOnNode(ctx, spec) return de.nodeRunner.ReconcileOnNode(ctx, spec)
} }
// Placement tags without a pin/spread resolve to the least-loaded
// matching agent on every reconcile — soft affinity, re-evaluated.
if len(spec.PlacementTags) > 0 {
if de.nodeRunner == nil {
return nil, fmt.Errorf("service has placement tags but remote dispatch is unavailable")
}
return de.nodeRunner.ReconcileAuto(ctx, spec)
}
// Local placement: retire any remote leftovers from a previous pin,
// spread, or tag affinity before reconciling local replicas.
if de.nodeRunner != nil {
_ = de.nodeRunner.RetireService(ctx, spec.ServiceID)
}
if spec.Replicas < 1 { if spec.Replicas < 1 {
spec.Replicas = 1 spec.Replicas = 1
} }
+3 -3
View File
@@ -53,7 +53,7 @@ func init() {
// Services // Services
register(spec{"containr_services_list", "List services in a project", "GET", "/projects/{project_id}/services", []string{"project_id"}, nil, false}) register(spec{"containr_services_list", "List services in a project", "GET", "/projects/{project_id}/services", []string{"project_id"}, nil, false})
register(spec{"containr_services_get", "Get a service by id", "GET", "/services/{id}", []string{"id"}, nil, false}) register(spec{"containr_services_get", "Get a service by id", "GET", "/services/{id}", []string{"id"}, nil, false})
register(spec{"containr_services_create", "Create a service in a project. Args: name (required), type, image, git_repo, git_branch, environment, port, domain, replicas, volumes = [{type: volume|bind, source, target, read_only}], builder = auto|railpack|nixpacks|dockerfile|static, cpu, memory, cpu_reserve, memory_reserve, static_build_cmd, static_dir, node_id (agent id | 'auto' = least-loaded online node; image services only — remote nodes can't run git builds), spread (bool — distribute replicas across every online schedulable node; mutually exclusive with a node pin).", "POST", "/projects/{project_id}/services", []string{"project_id"}, nil, false}) register(spec{"containr_services_create", "Create a service in a project. Args: name (required), type, image, git_repo, git_branch, environment, port, domain, replicas, volumes = [{type: volume|bind, source, target, read_only}], builder = auto|railpack|nixpacks|dockerfile|static, cpu, memory, cpu_reserve, memory_reserve, static_build_cmd, static_dir, node_id (agent id | 'auto' = least-loaded online node; image services only — remote nodes can't run git builds), spread (bool — distribute replicas across every online schedulable node; mutually exclusive with a node pin), placement_tags (string array — remote placement only onto nodes carrying every tag; re-resolved per reconcile, pin overrides).", "POST", "/projects/{project_id}/services", []string{"project_id"}, nil, false})
register(spec{"containr_services_delete", "Delete a service. Requires confirm:true.", "DELETE", "/services/{id}", []string{"id"}, nil, true}) register(spec{"containr_services_delete", "Delete a service. Requires confirm:true.", "DELETE", "/services/{id}", []string{"id"}, nil, true})
register(spec{"containr_services_start", "Start a service", "POST", "/services/{id}/start", []string{"id"}, nil, false}) register(spec{"containr_services_start", "Start a service", "POST", "/services/{id}/start", []string{"id"}, nil, false})
register(spec{"containr_services_stop", "Stop a service", "POST", "/services/{id}/stop", []string{"id"}, nil, false}) register(spec{"containr_services_stop", "Stop a service", "POST", "/services/{id}/stop", []string{"id"}, nil, false})
@@ -66,7 +66,7 @@ func init() {
register(spec{"containr_services_move", "Move a service to another project the caller owns. Args: id, project_id (required).", "POST", "/services/{id}/move", []string{"id"}, []string{"project_id"}, false}) register(spec{"containr_services_move", "Move a service to another project the caller owns. Args: id, project_id (required).", "POST", "/services/{id}/move", []string{"id"}, []string{"project_id"}, false})
register(spec{"containr_services_logs", "Get service runtime logs. Args: id, tail (default 100).", "GET", "/services/{id}/logs", []string{"id"}, nil, false}) register(spec{"containr_services_logs", "Get service runtime logs. Args: id, tail (default 100).", "GET", "/services/{id}/logs", []string{"id"}, nil, false})
register(spec{"containr_services_exec", "Run a one-off command in the service container (30s ceiling). Args: id, command (required).", "POST", "/services/{id}/exec", []string{"id"}, []string{"command"}, false}) register(spec{"containr_services_exec", "Run a one-off command in the service container (30s ceiling). Args: id, command (required).", "POST", "/services/{id}/exec", []string{"id"}, []string{"command"}, false})
register(spec{"containr_services_update", "Update a service. Args: id, name, image, command, domain, restart_policy, healthcheck_path, cpu, memory, replicas, port, volumes = [{type: volume|bind, source, target, read_only}] — volumes replaces the whole mount list; pass [] to clear. maintenance_mode (bool), basic_auth = [{username, password}] — replaces all creds; [] clears. builder = auto|railpack|nixpacks|dockerfile|static, cpu_reserve, memory_reserve, static_build_cmd, static_dir (empty string clears reserves/static fields). sleep_enabled (bool) + sleep_idle_minutes (1-1440) enable scale-to-zero on idle; a wake placeholder answers traffic while the service resumes. node_id pins placement to a node agent ('auto' = least-loaded, 'local' = run on the Containr host). spread (bool) distributes replicas across every online schedulable node; mutually exclusive with a pin — clear node_id to 'local' first if pinned.", "PUT", "/services/{id}", []string{"id"}, nil, false}) register(spec{"containr_services_update", "Update a service. Args: id, name, image, command, domain, restart_policy, healthcheck_path, cpu, memory, replicas, port, volumes = [{type: volume|bind, source, target, read_only}] — volumes replaces the whole mount list; pass [] to clear. maintenance_mode (bool), basic_auth = [{username, password}] — replaces all creds; [] clears. builder = auto|railpack|nixpacks|dockerfile|static, cpu_reserve, memory_reserve, static_build_cmd, static_dir (empty string clears reserves/static fields). sleep_enabled (bool) + sleep_idle_minutes (1-1440) enable scale-to-zero on idle; a wake placeholder answers traffic while the service resumes. node_id pins placement to a node agent ('auto' = least-loaded, 'local' = run on the Containr host). spread (bool) distributes replicas across every online schedulable node; mutually exclusive with a pin — clear node_id to 'local' first if pinned. placement_tags (string array; [] clears) restricts auto/spread placement to nodes carrying every tag; mutually exclusive with an explicit pin — clear the pin first.", "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})
@@ -173,7 +173,7 @@ func init() {
register(spec{"containr_nodes_get", "Get a node agent (admin)", "GET", "/agents/{id}", []string{"id"}, nil, false}) register(spec{"containr_nodes_get", "Get a node agent (admin)", "GET", "/agents/{id}", []string{"id"}, nil, 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_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, tags (string array — placement labels services can require via placement_tags; [] clears).", "PUT", "/agents/{id}", []string{"id"}, []string{"auto_prune", "name", "tags"}, 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_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_uncordon", "Re-open a node for placement (admin). Args: id.", "POST", "/agents/{id}/uncordon", []string{"id"}, nil, false})
register(spec{"containr_nodes_drain", "Drain a node: cordon + remove all service containers via the agent + unpin affected services so they redeploy locally. Requires the node online (admin). Args: id.", "POST", "/agents/{id}/drain", []string{"id"}, nil, true}) register(spec{"containr_nodes_drain", "Drain a node: cordon + remove all service containers via the agent + unpin affected services so they redeploy locally. Requires the node online (admin). Args: id.", "POST", "/agents/{id}/drain", []string{"id"}, nil, true})
@@ -0,0 +1,11 @@
-- +goose Up
-- Placement tags: operators label node_agents (tags jsonb array of strings),
-- services require a tag set (placement_tags). "auto" and spread placement
-- only consider online schedulable agents carrying every required tag.
-- Explicit node_id pins bypass tag filtering — the pin is an operator override.
ALTER TABLE services ADD COLUMN IF NOT EXISTS placement_tags jsonb NOT NULL DEFAULT '[]';
ALTER TABLE node_agents ADD COLUMN IF NOT EXISTS tags jsonb NOT NULL DEFAULT '[]';
-- +goose Down
ALTER TABLE services DROP COLUMN IF EXISTS placement_tags;
ALTER TABLE node_agents DROP COLUMN IF EXISTS tags;
+54 -1
View File
@@ -110,6 +110,12 @@ RETURNING *;
UPDATE container_instances SET status = $2, updated_at = NOW() UPDATE container_instances SET status = $2, updated_at = NOW()
WHERE id = $1; WHERE id = $1;
-- name: UpdateContainerPorts :exec
-- Agent reports the real host bindings after create — persist them with
-- the status flip so ingress can route to the assigned port.
UPDATE container_instances SET status = $2, ports = $3, updated_at = NOW()
WHERE id = $1;
-- name: GetLastAgentCommandByType :one -- name: GetLastAgentCommandByType :one
SELECT * FROM agent_commands SELECT * FROM agent_commands
WHERE node_agent_id = $1 AND type = $2 WHERE node_agent_id = $1 AND type = $2
@@ -141,8 +147,40 @@ SELECT id, name FROM node_agents
WHERE status = 'online' AND schedulable WHERE status = 'online' AND schedulable
ORDER BY id; ORDER BY id;
-- name: ListSchedulableAgentsMatching :many
-- Only agents carrying every required placement tag (jsonb array containment).
SELECT id, name FROM node_agents
WHERE status = 'online' AND schedulable
AND tags @> $1::jsonb
ORDER BY id;
-- name: PickLeastLoadedAgentMatching :one
-- Same resource-aware ordering as PickLeastLoadedAgent, restricted to agents
-- carrying every required placement tag.
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
AND a.tags @> $1::jsonb
ORDER BY
COALESCE(
(a.resources->'memory'->>'used')::double precision
/ NULLIF((a.resources->'memory'->>'total')::double precision, 0),
1.0
) ASC,
COALESCE((a.resources->'cpu'->>'usage')::double precision, 100.0) ASC,
COALESCE(h.container_count, 0) ASC,
a.created_at ASC
LIMIT 1;
-- name: SetAgentTags :exec
UPDATE node_agents SET tags = $2::jsonb, updated_at = NOW() WHERE id = $1;
-- name: ListServiceContainers :many -- name: ListServiceContainers :many
SELECT id, name, node_agent_id, status FROM container_instances SELECT id, name, node_agent_id, status, ports, resources FROM container_instances
WHERE service_id = $1; WHERE service_id = $1;
-- name: SetAgentSchedulable :exec -- name: SetAgentSchedulable :exec
@@ -165,5 +203,20 @@ ON CONFLICT (id) DO UPDATE SET
-- name: DeleteServiceContainersOnAgent :exec -- name: DeleteServiceContainersOnAgent :exec
DELETE FROM container_instances WHERE node_agent_id = $1 AND service_id = $2; DELETE FROM container_instances WHERE node_agent_id = $1 AND service_id = $2;
-- name: UpdateContainerStateByName :exec
-- Agent heartbeat reports real docker state + net counters; reconcile
-- inventory for rows this node owns. Tombstones win — never resurrect a
-- removed replica. $4 is the cumulative rx+tx byte counter.
UPDATE container_instances SET
status = jsonb_set(status, '{state}', $3::jsonb),
resources = jsonb_set(COALESCE(resources, '{}'::jsonb), '{net_bytes}', $4::jsonb),
updated_at = NOW()
WHERE name = $1 AND node_agent_id = $2
AND status->>'state' IS DISTINCT FROM 'removed';
-- name: ListServiceAgents :many
-- Distinct agents holding inventory rows for a service.
SELECT DISTINCT node_agent_id FROM container_instances WHERE service_id = $1;
-- name: ScrubCommandPayload :exec -- name: ScrubCommandPayload :exec
UPDATE agent_commands SET payload = $2, updated_at = NOW() WHERE id = $1; UPDATE agent_commands SET payload = $2, updated_at = NOW() WHERE id = $1;
+4 -2
View File
@@ -68,7 +68,8 @@ CREATE TABLE services (
static_build_cmd VARCHAR(255) NOT NULL DEFAULT '', static_build_cmd VARCHAR(255) NOT NULL DEFAULT '',
static_dir VARCHAR(255) NOT NULL DEFAULT '', static_dir VARCHAR(255) NOT NULL DEFAULT '',
node_id VARCHAR(255), node_id VARCHAR(255),
spread BOOLEAN NOT NULL DEFAULT false spread BOOLEAN NOT NULL DEFAULT false,
placement_tags JSONB NOT NULL DEFAULT '[]'
); );
CREATE TABLE registries ( CREATE TABLE registries (
@@ -196,7 +197,8 @@ CREATE TABLE node_agents (
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 schedulable BOOLEAN NOT NULL DEFAULT true,
tags JSONB NOT NULL DEFAULT '[]'
); );
CREATE TABLE container_instances ( CREATE TABLE container_instances (
@@ -47,6 +47,7 @@ export function NodeDetailPage() {
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 [confirmDrain, setConfirmDrain] = useState(false);
const [tagDraft, setTagDraft] = useState<string | null>(null);
const agentQuery = useQuery({ const agentQuery = useQuery({
queryKey: ['node-agent', id], queryKey: ['node-agent', id],
@@ -79,8 +80,8 @@ export function NodeDetailPage() {
queryClient.invalidateQueries({ queryKey: ['usage-agents'] }); queryClient.invalidateQueries({ queryKey: ['usage-agents'] });
}; };
const updateMutation = useMutation({ const updateMutation = useMutation({
mutationFn: (body: { name?: string; auto_prune?: boolean }) => updateAgent(id, body), mutationFn: (body: { name?: string; auto_prune?: boolean; tags?: string[] }) => updateAgent(id, body),
onSuccess: () => { setRename(null); invalidate(); }, onSuccess: () => { setRename(null); setTagDraft(null); invalidate(); },
}); });
const pruneMutation = useMutation({ const pruneMutation = useMutation({
mutationFn: (volumes: boolean) => pruneAgent(id, { volumes }), mutationFn: (volumes: boolean) => pruneAgent(id, { volumes }),
@@ -177,6 +178,46 @@ export function NodeDetailPage() {
mesh: {meshAddresses(agent).map(([iface, ip]) => `${iface}=${ip}`).join(' · ')} mesh: {meshAddresses(agent).map(([iface, ip]) => `${iface}=${ip}`).join(' · ')}
</p> </p>
)} )}
<p className="v-mono mt-1 text-[11px] text-[var(--text-tertiary)] flex items-center gap-1.5 flex-wrap">
tags:
{tagDraft === null ? (
<>
{agent.tags.length > 0
? agent.tags.map((t) => (
<span key={t} className="px-1.5 py-0.5 rounded border border-[var(--border-subtle)] text-[var(--text-secondary)]">{t}</span>
))
: <span className="text-[var(--text-muted)]">none</span>}
{isAdmin && (
<button
onClick={() => setTagDraft(agent.tags.join(', '))}
className="text-[var(--text-muted)] hover:text-[var(--text-primary)]"
title="Edit placement tags — services can require them via placement_tags"
>
<Pencil size={11} />
</button>
)}
</>
) : (
<>
<input
value={tagDraft}
onChange={(e) => setTagDraft(e.target.value)}
onKeyDown={(e) => {
if (e.key === 'Enter') updateMutation.mutate({ tags: tagDraft.split(',').map((t) => t.trim()).filter(Boolean) });
if (e.key === 'Escape') setTagDraft(null);
}}
autoFocus
placeholder="eu-west, gpu"
className="rounded-md border border-[var(--border-subtle)] bg-[var(--bg-tertiary)] px-2 py-0.5 text-[11px] text-[var(--text-primary)] w-52"
/>
<button
onClick={() => updateMutation.mutate({ tags: tagDraft.split(',').map((t) => t.trim()).filter(Boolean) })}
className="text-[var(--accent-primary)] hover:underline"
>save</button>
<button onClick={() => setTagDraft(null)} className="text-[var(--text-muted)] hover:underline">cancel</button>
</>
)}
</p>
</div> </div>
<div className="flex items-center gap-3"> <div className="flex items-center gap-3">
<span className={`v-mono text-[11px] px-2 py-1 rounded-md border ${online <span className={`v-mono text-[11px] px-2 py-1 rounded-md border ${online
@@ -2452,6 +2452,30 @@ export function ServiceDetailPage() {
<span className="text-[var(--text-tertiary)]">running on {service.nodeName}</span> <span className="text-[var(--text-tertiary)]">running on {service.nodeName}</span>
)} )}
</div> </div>
{!service.nodeId && (
<div className="mt-3 flex items-center gap-2 text-xs">
<label className="text-[var(--text-tertiary)]" htmlFor="placement-tags">
required node tags
</label>
<input
id="placement-tags"
type="text"
placeholder="e.g. eu-west, gpu — empty = any node"
defaultValue={(service.placementTags ?? []).join(', ')}
disabled={updateServiceMutation.isPending}
title="Remote placement only targets nodes carrying every tag. Re-resolved each reconcile — the service follows the tag, not a stored node."
onBlur={(e) => {
const next = e.target.value.split(',').map((t) => t.trim()).filter(Boolean);
const prev = service.placementTags ?? [];
if (next.join('|') !== prev.join('|')) {
updateServiceMutation.mutate({ placement_tags: next });
}
}}
onKeyDown={(e) => { if (e.key === 'Enter') (e.target as HTMLInputElement).blur(); }}
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)] w-72"
/>
</div>
)}
{updateServiceMutation.isError && ( {updateServiceMutation.isError && (
<p className="mt-2 text-xs text-[var(--error)]">{(updateServiceMutation.error as Error).message}</p> <p className="mt-2 text-xs text-[var(--error)]">{(updateServiceMutation.error as Error).message}</p>
)} )}
+10 -1
View File
@@ -8008,6 +8008,7 @@ export interface paths {
id?: string; id?: string;
name?: string; name?: string;
status?: string; status?: string;
tags?: string[];
}[]; }[];
}; };
}; };
@@ -8104,7 +8105,7 @@ export interface paths {
}; };
/** /**
* Update node agent * Update node agent
* @description Partial update — accepts any subset of agent fields (name, status, labels, etc.). * @description Partial update — accepts any subset of agent fields (name, status, auto_prune, tags). `tags` sets the node's placement labels; services with placement_tags only schedule onto agents carrying every required tag.
*/ */
put: { put: {
parameters: { parameters: {
@@ -10944,6 +10945,8 @@ export interface components {
node_name?: string; node_name?: string;
/** @description Replicas are distributed across every online, schedulable node agent (replica i → agents[i mod n]). Mutually exclusive with an explicit node_id pin. */ /** @description Replicas are distributed across every online, schedulable node agent (replica i → agents[i mod n]). Mutually exclusive with an explicit node_id pin. */
spread?: boolean; spread?: boolean;
/** @description Node tags required for remote placement — auto/spread only consider agents carrying every tag; re-resolved each reconcile. Mutually exclusive with an explicit node_id pin (read-only on responses when a pin is set). */
placement_tags?: string[];
/** @description Computed public URL (published port or domain) */ /** @description Computed public URL (published port or domain) */
public_url?: string; public_url?: string;
/** /**
@@ -11018,6 +11021,8 @@ export interface components {
node_id?: string; node_id?: string;
/** @description Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin */ /** @description Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin */
spread?: boolean; spread?: boolean;
/** @description Restrict remote placement to agents carrying every listed tag; mutually exclusive with an explicit pin or 'local' */
placement_tags?: string[];
}; };
UpdateServiceRequest: { UpdateServiceRequest: {
/** @description Service name */ /** @description Service name */
@@ -11083,6 +11088,8 @@ export interface components {
node_id?: string; node_id?: string;
/** @description Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin — send 'local' first to clear a pin */ /** @description Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin — send 'local' first to clear a pin */
spread?: boolean; spread?: boolean;
/** @description Restrict remote placement to agents carrying every listed tag; [] clears. Mutually exclusive with a pin or 'local' — clear the pin first */
placement_tags?: string[];
}; };
ServiceDomain: { ServiceDomain: {
/** Format: uuid */ /** Format: uuid */
@@ -12202,6 +12209,8 @@ export interface components {
auto_prune?: boolean; auto_prune?: boolean;
/** @description False on a cordoned/drained node — `auto` placement skips it and new explicit pins are rejected. */ /** @description False on a cordoned/drained node — `auto` placement skips it and new explicit pins are rejected. */
schedulable?: boolean; schedulable?: boolean;
/** @description Operator-set placement labels; services with placement_tags require all of them on the target node. */
tags?: string[];
/** 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",
+6 -1
View File
@@ -87,6 +87,7 @@ export type ServiceEntity = {
nodeId?: string; nodeId?: string;
nodeName?: string; nodeName?: string;
spread?: boolean; spread?: boolean;
placementTags?: string[];
publicUrl?: string; publicUrl?: string;
}; };
@@ -94,6 +95,7 @@ export type NodeOption = {
id: string; id: string;
name: string; name: string;
status: string; status: string;
tags?: string[];
}; };
export async function listNodeOptions(): Promise<NodeOption[]> { export async function listNodeOptions(): Promise<NodeOption[]> {
@@ -358,6 +360,7 @@ export type NodeAgentEntity = {
resources: NodeResources; resources: NodeResources;
autoPrune: boolean; autoPrune: boolean;
schedulable: boolean; schedulable: boolean;
tags: string[];
metadata: Record<string, unknown>; metadata: Record<string, unknown>;
lastHeartbeat?: string; lastHeartbeat?: string;
createdAt?: string; createdAt?: string;
@@ -698,6 +701,7 @@ function normalizeService(service: RawService): ServiceEntity | null {
nodeId: service.node_id, nodeId: service.node_id,
nodeName: service.node_name, nodeName: service.node_name,
spread: service.spread, spread: service.spread,
placementTags: service.placement_tags,
publicUrl: service.public_url, publicUrl: service.public_url,
}; };
} }
@@ -1100,6 +1104,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, schedulable: agent.schedulable ?? true,
tags: Array.isArray(agent.tags) ? agent.tags : [],
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,
@@ -1139,7 +1144,7 @@ export async function listAgents(): Promise<NodeAgentEntity[]> {
return normalizeAgentArray(payload.agents); return normalizeAgentArray(payload.agents);
} }
export async function updateAgent(id: string, body: { name?: string; auto_prune?: boolean }): Promise<NodeAgentEntity | null> { export async function updateAgent(id: string, body: { name?: string; auto_prune?: boolean; tags?: string[] }): Promise<NodeAgentEntity | null> {
const payload = await requestJson<{ agent?: RawNodeAgent }>(`/agents/${id}`, { const payload = await requestJson<{ agent?: RawNodeAgent }>(`/agents/${id}`, {
method: 'PUT', method: 'PUT',
body: JSON.stringify(body), body: JSON.stringify(body),
+20 -1
View File
@@ -5107,6 +5107,9 @@ paths:
type: string type: string
status: status:
type: string type: string
tags:
type: array
items: { type: string }
/agents: /agents:
get: get:
@@ -5157,7 +5160,7 @@ paths:
tags: tags:
- Agents - Agents
summary: Update node agent summary: Update node agent
description: Partial update — accepts any subset of agent fields (name, status, labels, etc.). description: Partial update — accepts any subset of agent fields (name, status, auto_prune, tags). `tags` sets the node's placement labels; services with placement_tags only schedule onto agents carrying every required tag.
security: security:
- bearerAuth: [] - bearerAuth: []
parameters: parameters:
@@ -7017,6 +7020,10 @@ components:
spread: spread:
type: boolean type: boolean
description: Replicas are distributed across every online, schedulable node agent (replica i → agents[i mod n]). Mutually exclusive with an explicit node_id pin. description: Replicas are distributed across every online, schedulable node agent (replica i → agents[i mod n]). Mutually exclusive with an explicit node_id pin.
placement_tags:
type: array
items: { type: string }
description: Node tags required for remote placement — auto/spread only consider agents carrying every tag; re-resolved each reconcile. Mutually exclusive with an explicit node_id pin (read-only on responses when a pin is set).
public_url: public_url:
type: string type: string
description: Computed public URL (published port or domain) description: Computed public URL (published port or domain)
@@ -7114,6 +7121,10 @@ components:
spread: spread:
type: boolean type: boolean
description: Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin description: Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin
placement_tags:
type: array
items: { type: string }
description: Restrict remote placement to agents carrying every listed tag; mutually exclusive with an explicit pin or 'local'
UpdateServiceRequest: UpdateServiceRequest:
type: object type: object
@@ -7206,6 +7217,10 @@ components:
spread: spread:
type: boolean type: boolean
description: Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin — send 'local' first to clear a pin description: Distribute replicas across every online, schedulable node agent; mutually exclusive with an explicit node_id pin — send 'local' first to clear a pin
placement_tags:
type: array
items: { type: string }
description: Restrict remote placement to agents carrying every listed tag; [] clears. Mutually exclusive with a pin or 'local' — clear the pin first
ServiceDomain: ServiceDomain:
type: object type: object
@@ -9091,6 +9106,10 @@ components:
description: >- description: >-
False on a cordoned/drained node — `auto` placement skips it and False on a cordoned/drained node — `auto` placement skips it and
new explicit pins are rejected. new explicit pins are rejected.
tags:
type: array
items: { type: string }
description: Operator-set placement labels; services with placement_tags require all of them on the target node.
last_heartbeat: last_heartbeat:
type: string type: string
format: date-time format: date-time
+16 -2
View File
@@ -152,8 +152,22 @@ containr services update <id> --replicas 4 --spread # replica i → node[i %
containr services update <id> --spread=false # back to single placement containr services update <id> --spread=false # back to single placement
``` ```
Remote nodes run registry-pulled images only — git builds, domains Constrain `auto`/`spread` placement to nodes carrying operator tags —
(Traefik lives on the host), and sleep detection stay local. Start, `nodes update` labels a node, `--placement-tags` requires every listed tag
(mutually exclusive with a pin; an empty value clears the requirement):
```bash
containr nodes update <id> --tags gpu,eu-west
containr services update <id> --placement-tags gpu # only gpu nodes are eligible
```
Remote nodes run registry-pulled images only — git builds stay local.
Domains work remotely: the agent publishes the service port and the
backend writes a Traefik file-provider route to the node's mesh IP
(TRAEFIK_DYNAMIC_DIR — wired by the infra compose). Sleep mode works
too: heartbeats carry per-container net counters, idle remote services
lose their replicas via the agent, and the route repoints at the
backend's wake page — the next hit reconciles the service back. Start,
stop, restart, redeploy, and delete all fan out to the pinned node's stop, restart, redeploy, and delete all fan out to the pinned node's
agent. `services get` shows the resolved `node_name`. agent. `services get` shows the resolved `node_name`.
+9 -2
View File
@@ -7,7 +7,8 @@ services:
- "--api.insecure=${TRAEFIK_API_INSECURE:-true}" - "--api.insecure=${TRAEFIK_API_INSECURE:-true}"
- "--providers.docker=true" - "--providers.docker=true"
- "--providers.docker.exposedbydefault=false" - "--providers.docker.exposedbydefault=false"
- "--providers.file.filename=/etc/traefik/traefik-dynamic.yml" - "--providers.file.directory=/traefik-dynamic"
- "--providers.file.watch=true"
- "--entrypoints.web.address=:80" - "--entrypoints.web.address=:80"
- "--entrypoints.websecure.address=:443" - "--entrypoints.websecure.address=:443"
- "--entrypoints.traefik.address=:8080" - "--entrypoints.traefik.address=:8080"
@@ -26,7 +27,8 @@ services:
volumes: volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro - /var/run/docker.sock:/var/run/docker.sock:ro
- ../data/letsencrypt:/letsencrypt - ../data/letsencrypt:/letsencrypt
- ./traefik-dynamic.yml:/etc/traefik/traefik-dynamic.yml:ro - traefik_dynamic:/traefik-dynamic
- ./traefik-dynamic.yml:/traefik-dynamic/00-base.yml:ro
networks: networks:
- containr-network - containr-network
restart: unless-stopped restart: unless-stopped
@@ -100,6 +102,8 @@ services:
- JWT_SECRET=${JWT_SECRET:-dev_jwt_secret_key_change_in_production} - JWT_SECRET=${JWT_SECRET:-dev_jwt_secret_key_change_in_production}
- CONTAINR_AGENT_AUTH_TOKEN=${CONTAINR_AGENT_AUTH_TOKEN:-} - CONTAINR_AGENT_AUTH_TOKEN=${CONTAINR_AGENT_AUTH_TOKEN:-}
- CONTAINR_AGENT_AUTH_TOKENS=${CONTAINR_AGENT_AUTH_TOKENS:-} - CONTAINR_AGENT_AUTH_TOKENS=${CONTAINR_AGENT_AUTH_TOKENS:-}
- TRAEFIK_DYNAMIC_DIR=/traefik-dynamic
- TRAEFIK_BACKEND_URL=http://containr-backend:8080
- COOKIE_SECURE=${COOKIE_SECURE:-true} - COOKIE_SECURE=${COOKIE_SECURE:-true}
- TRUSTED_PROXY_CIDR=${TRUSTED_PROXY_CIDR:-172.20.0.0/16} - TRUSTED_PROXY_CIDR=${TRUSTED_PROXY_CIDR:-172.20.0.0/16}
- CORS_ORIGINS=${CORS_ORIGINS:-http://localhost,http://localhost:3000} - CORS_ORIGINS=${CORS_ORIGINS:-http://localhost,http://localhost:3000}
@@ -136,6 +140,7 @@ services:
- GITEA_BASE_URL=${GITEA_BASE_URL:-https://gitea.example.com} - GITEA_BASE_URL=${GITEA_BASE_URL:-https://gitea.example.com}
volumes: volumes:
- /var/run/docker.sock:/var/run/docker.sock - /var/run/docker.sock:/var/run/docker.sock
- traefik_dynamic:/traefik-dynamic
group_add: group_add:
- "${DOCKER_GID:-0}" - "${DOCKER_GID:-0}"
networks: networks:
@@ -213,6 +218,8 @@ volumes:
driver: local driver: local
dragonfly_data: dragonfly_data:
driver: local driver: local
traefik_dynamic:
driver: local
networks: networks:
containr-network: containr-network: