Compare commits

..
4 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
16 changed files with 690 additions and 50 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.
+18 -6
View File
@@ -296,9 +296,22 @@ 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. Tag affinity shipped: `node_agents.tags` Traefik file-provider config per remote service into
`TRAEFIK_DYNAMIC_DIR` (shared volume, `watch=true`) pointing the
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`) + (operator-set via `PUT /agents/:id` / `nodes update --tags`) +
`services.placement_tags` — `auto` and `spread` only consider `services.placement_tags` — `auto` and `spread` only consider
online schedulable agents carrying every required tag, tags are online schedulable agents carrying every required tag, tags are
@@ -307,9 +320,8 @@ and wired refs, one click or one CLI call.
and local + tags are rejected, empty array clears. Surfaces: and local + tags are rejected, empty array clears. Surfaces:
`--placement-tags` on services create/update, `--tags` on `--placement-tags` on services create/update, `--tags` on
`nodes update`, MCP args, OpenAPI, node-detail tag editor + `nodes update`, MCP args, OpenAPI, node-detail tag editor +
service-detail placement-tags input. Remaining: deployment service-detail placement-tags input. Remaining: build-on-node,
status reported back from agent results, spread weighting by remote sleep/wake, spread weighting by real capacity.
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
+27
View File
@@ -198,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 {
@@ -665,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))
}
+76 -25
View File
@@ -19,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
@@ -131,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
@@ -151,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)
} }
@@ -185,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
} }
@@ -296,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)
} }
@@ -314,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}
@@ -336,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
@@ -348,20 +399,20 @@ 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):
} }
} }
+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
+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"})
} }
@@ -765,8 +765,36 @@ func (q *Queries) ListSchedulableAgentsMatching(ctx context.Context, dollar_1 js
return items, nil 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
` `
@@ -775,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) {
@@ -791,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
} }
@@ -1005,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
@@ -98,6 +98,8 @@ type Querier interface {
ListSchedulableAgents(ctx context.Context) ([]ListSchedulableAgentsRow, error) ListSchedulableAgents(ctx context.Context) ([]ListSchedulableAgentsRow, error)
// Only agents carrying every required placement tag (jsonb array containment). // Only agents carrying every required placement tag (jsonb array containment).
ListSchedulableAgentsMatching(ctx context.Context, dollar_1 json.RawMessage) ([]ListSchedulableAgentsMatchingRow, error) 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)
@@ -134,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
@@ -32,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 {
@@ -182,6 +182,12 @@ func (de *DeploymentEngine) ReconcileService(ctx context.Context, spec RuntimeSp
} }
return de.nodeRunner.ReconcileAuto(ctx, spec) 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
} }
+22 -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
@@ -174,7 +180,7 @@ LIMIT 1;
UPDATE node_agents SET tags = $2::jsonb, updated_at = NOW() WHERE id = $1; 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
@@ -197,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;
+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",
+7 -2
View File
@@ -161,8 +161,13 @@ containr nodes update <id> --tags gpu,eu-west
containr services update <id> --placement-tags gpu # only gpu nodes are eligible containr services update <id> --placement-tags gpu # only gpu nodes are eligible
``` ```
Remote nodes run registry-pulled images only — git builds, domains Remote nodes run registry-pulled images only — git builds stay local.
(Traefik lives on the host), and sleep detection stay local. Start, 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: