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>
This commit is contained in:
Tomáš DvořákandDevin authored and GitHub committed 2026-10-09 18:11:19 +02:00
1 parent 9de586f177
commit 4458de56b0
23 files changed
+475 -37

No files matched your search

+32 -1
View File
@@ -42,6 +42,9 @@ type NodeAgent struct {
LastHeartbeat time.Time `json:"last_heartbeat"`
AutoPrune bool `json:"auto_prune"`
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"`
UpdatedAt time.Time `json:"updated_at"`
Metadata map[string]interface{} `json:"metadata"`
@@ -242,10 +245,15 @@ func (h *NodeAgentHandler) ListNodeOptions(c *gin.Context) {
}
options := make([]gin.H, 0, len(rows))
for _, row := range rows {
var tags []string
if len(row.Tags) > 0 {
_ = json.Unmarshal(row.Tags, &tags)
}
options = append(options, gin.H{
"id": row.ID,
"name": row.Name,
"status": row.Status.String,
"tags": tags,
})
}
c.JSON(http.StatusOK, gin.H{"nodes": options})
@@ -262,6 +270,7 @@ func agentFromRow(row sqlcdb.NodeAgent) NodeAgent {
Version: row.Version.String,
AutoPrune: row.AutoPrune,
Schedulable: row.Schedulable,
Tags: []string{},
Metadata: map[string]interface{}{},
}
if row.LastHeartbeat.Valid {
@@ -276,6 +285,9 @@ func agentFromRow(row sqlcdb.NodeAgent) NodeAgent {
unmarshalRaw(row.Capabilities, &agent.Capabilities)
unmarshalRaw(row.Resources, &agent.Resources)
unmarshalRaw(row.Metadata, &agent.Metadata)
if len(row.Tags) > 0 {
_ = json.Unmarshal(row.Tags, &agent.Tags)
}
return agent
}
@@ -546,7 +558,26 @@ func (h *NodeAgentHandler) UpdateAgent(c *gin.Context) {
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.
+37 -4
View File
@@ -4,6 +4,7 @@ import (
"context"
"crypto/sha256"
"database/sql"
"encoding/json"
"fmt"
"strconv"
"strings"
@@ -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
// not a spread target — pin "local" or leave unset for local execution.
func (r *agentNodeRunner) ReconcileSpread(ctx context.Context, spec deployment.RuntimeSpec) (*deployment.RuntimeState, error) {
agents, err := r.q.ListSchedulableAgents(ctx)
tagsRaw, _ := json.Marshal(spec.PlacementTags)
if len(spec.PlacementTags) == 0 {
tagsRaw = []byte("[]")
}
agents, err := r.q.ListSchedulableAgentsMatching(ctx, json.RawMessage(tagsRaw))
if err != nil {
return nil, fmt.Errorf("list schedulable nodes: %w", err)
}
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 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.
// Rows only carry the last dispatched state — the agent doesn't stream
// container events — so "running" means "last create succeeded".
@@ -344,16 +368,25 @@ func (r *agentNodeRunner) enqueueAndWait(ctx context.Context, agentID, container
}
// resolveNodePin validates or resolves a node_id request value. "auto" picks
// the least-loaded online agent; "" / "local" clears the pin.
func resolveNodePin(ctx context.Context, db *database.DB, requested string) (string, error) {
// the least-loaded online agent that carries every required placement tag;
// "" / "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)
if requested == "" || requested == "local" {
return "", nil
}
q := sqlcdb.New(db.DB)
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 len(requiredTags) > 0 {
return "", fmt.Errorf("no online nodes carry placement tags %v", requiredTags)
}
return "", fmt.Errorf("no online nodes available")
}
return id, nil
+1
View File
@@ -94,6 +94,7 @@ func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpe
RestartPolicy: service.RestartPolicy,
NodeID: serviceNodeID(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
// 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:"-"`
// Spread distributes replicas across every online, schedulable agent.
// Mutually exclusive with an explicit NodeID pin.
Spread bool `json:"spread" db:"-"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
Spread bool `json:"spread" db:"-"`
// PlacementTags restrict auto/spread candidates to agents carrying all
// 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
@@ -91,12 +94,51 @@ func serviceSpread(db *database.DB, serviceID uuid.UUID) bool {
return false
}
// loadServiceNode fills the pin + spread + the agent's display name for
// responses.
// servicePlacementTags reads the placement_tags column with the lazy pattern.
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) {
var name sql.NullString
s.NodeID = serviceNodeID(db, s.ID)
s.Spread = serviceSpread(db, s.ID)
s.PlacementTags = servicePlacementTags(db, s.ID)
if s.NodeID == "" {
return
}
@@ -188,6 +230,9 @@ type CreateServiceRequest struct {
// Spread distributes replicas across every online, schedulable agent.
// Mutually exclusive with an explicit node pin.
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
@@ -222,6 +267,10 @@ type UpdateServiceRequest struct {
// Spread toggles multi-node replica distribution; pointer so false is
// a real update, not a missing field.
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
@@ -465,12 +514,17 @@ func handleCreateService(c *gin.Context) {
if req.Spread != nil {
service.Spread = *req.Spread
}
service.PlacementTags = normalizePlacementTags(req.PlacementTags)
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"})
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 {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
return
@@ -488,9 +542,9 @@ func handleCreateService(c *gin.Context) {
git_repo, git_branch, build_path, cpu, memory, replicas, port, domain,
healthcheck_path, restart_policy, volumes,
builder, cpu_reserve, memory_reserve, static_build_cmd, static_dir,
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,
$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,
sourceType, firstNonEmpty(service.GitRepo, service.Image), service.Image,
"", service.Command, service.Type, service.Status, service.Image, service.Command,
@@ -500,6 +554,7 @@ func handleCreateService(c *gin.Context) {
service.StaticBuildCmd, service.StaticDir,
sql.NullString{String: service.NodeID, Valid: service.NodeID != ""},
service.Spread,
tagsJSON(service.PlacementTags),
service.CreatedAt, service.UpdatedAt,
)
@@ -807,13 +862,22 @@ func handleUpdateService(c *gin.Context) {
} else {
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 {
nodeValue := strings.TrimSpace(*req.NodeID)
if existingService.Spread && nodeValue != "" && nodeValue != "auto" && nodeValue != "local" {
c.JSON(http.StatusBadRequest, gin.H{"error": "spread and an explicit node pin are mutually exclusive — set spread=false first", "code": "VALIDATION"})
return
}
nodeID, nodeErr := resolveNodePin(c.Request.Context(), db.(*database.DB), nodeValue)
if 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 {
c.JSON(http.StatusBadRequest, gin.H{"error": nodeErr.Error(), "code": "VALIDATION"})
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"})
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()
@@ -838,7 +906,8 @@ func handleUpdateService(c *gin.Context) {
restart_policy = $15, volumes = COALESCE($17::jsonb, volumes), updated_at = $16,
builder = $19, cpu_reserve = $20, memory_reserve = $21,
static_build_cmd = $22, static_dir = $23,
sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26, spread = $27
sleep_enabled = $24, sleep_idle_minutes = $25, node_id = $26, spread = $27,
placement_tags = $28::jsonb
WHERE id = $18`,
existingService.Name, existingService.Type, existingService.Image, existingService.Command,
existingService.Environment, existingService.GitRepo, existingService.GitBranch,
@@ -851,6 +920,7 @@ func handleUpdateService(c *gin.Context) {
existingService.SleepEnabled, existingService.SleepIdleMinutes,
sql.NullString{String: existingService.NodeID, Valid: existingService.NodeID != ""},
existingService.Spread,
tagsJSON(existingService.PlacementTags),
)
if err != nil {
+14 -2
View File
@@ -2,6 +2,7 @@ package commands
import (
"fmt"
"strings"
"github.com/spf13/cobra"
)
@@ -83,7 +84,7 @@ var nodesDeleteCmd = &cobra.Command{
var nodesUpdateCmd = &cobra.Command{
Use: "update <id>",
Short: "Update a node agent (--name, --auto-prune)",
Short: "Update a node agent (--name, --auto-prune, --tags)",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
c, err := client()
@@ -97,8 +98,18 @@ var nodesUpdateCmd = &cobra.Command{
if cmd.Flags().Changed("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 {
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)
if err != nil {
@@ -292,6 +303,7 @@ var nodeTokensRevokeCmd = &cobra.Command{
func init() {
nodesUpdateCmd.Flags().String("name", "", "node display name")
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().Bool("volumes", false, "also prune volumes (destructive — asks twice)")
nodeTokensCmd.AddCommand(nodeTokensIssueCmd, nodeTokensListCmd, nodeTokensRevokeCmd)
@@ -117,6 +117,15 @@ var servicesCreateCmd = &cobra.Command{
if v, _ := cmd.Flags().GetBool("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)
if err != nil {
return err
@@ -210,6 +219,16 @@ var servicesUpdateCmd = &cobra.Command{
v, _ := cmd.Flags().GetBool("spread")
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") {
v, _ := cmd.Flags().GetInt("sleep-idle")
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("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.String("placement-tags", "", "comma-separated node tags required for remote placement (auto/spread only)")
uf := servicesUpdateCmd.Flags()
uf.String("name", "", "service name")
@@ -529,6 +549,7 @@ func init() {
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.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"} {
verb := a
@@ -75,7 +75,7 @@ INSERT INTO node_agents (
id, name, hostname, ip_address, port, status, version,
capabilities, resources, last_heartbeat, metadata
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, 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 {
@@ -123,6 +123,7 @@ func (q *Queries) CreateAgent(ctx context.Context, arg CreateAgentParams) (NodeA
&i.Metadata,
&i.AutoPrune,
&i.Schedulable,
&i.Tags,
)
return i, err
}
@@ -257,7 +258,7 @@ func (q *Queries) DeleteServiceContainersOnAgent(ctx context.Context, arg Delete
}
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) {
@@ -279,12 +280,13 @@ func (q *Queries) GetAgent(ctx context.Context, id string) (NodeAgent, error) {
&i.Metadata,
&i.AutoPrune,
&i.Schedulable,
&i.Tags,
)
return i, err
}
const getAgentByHostAndIP = `-- name: GetAgentByHostAndIP :one
SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, 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
LIMIT 1
`
@@ -313,6 +315,7 @@ func (q *Queries) GetAgentByHostAndIP(ctx context.Context, arg GetAgentByHostAnd
&i.Metadata,
&i.AutoPrune,
&i.Schedulable,
&i.Tags,
)
return i, err
}
@@ -520,7 +523,7 @@ func (q *Queries) ListAgentHeartbeatsSince(ctx context.Context, arg ListAgentHea
}
const listAgents = `-- name: ListAgents :many
SELECT id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, 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) {
@@ -548,6 +551,7 @@ func (q *Queries) ListAgents(ctx context.Context) ([]NodeAgent, error) {
&i.Metadata,
&i.AutoPrune,
&i.Schedulable,
&i.Tags,
); err != nil {
return nil, err
}
@@ -725,6 +729,42 @@ func (q *Queries) ListSchedulableAgents(ctx context.Context) ([]ListSchedulableA
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 listServiceContainers = `-- name: ListServiceContainers :many
SELECT id, name, node_agent_id, status FROM container_instances
WHERE service_id = $1
@@ -794,6 +834,36 @@ func (q *Queries) PickLeastLoadedAgent(ctx context.Context) (string, error) {
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
UPDATE agent_commands SET payload = $2, updated_at = NOW() WHERE id = $1
`
@@ -822,6 +892,20 @@ func (q *Queries) SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulab
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
UPDATE node_agents SET
name = $2,
@@ -838,7 +922,7 @@ UPDATE node_agents SET
schedulable = $13,
updated_at = NOW()
WHERE id = $1
RETURNING id, name, hostname, ip_address, port, status, version, capabilities, resources, last_heartbeat, created_at, updated_at, metadata, auto_prune, 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 {
@@ -890,6 +974,7 @@ func (q *Queries) UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeA
&i.Metadata,
&i.AutoPrune,
&i.Schedulable,
&i.Tags,
)
return i, err
}
@@ -177,6 +177,7 @@ type NodeAgent struct {
Metadata pqtype.NullRawMessage `json:"metadata"`
AutoPrune bool `json:"auto_prune"`
Schedulable bool `json:"schedulable"`
Tags json.RawMessage `json:"tags"`
}
type Notification struct {
@@ -277,6 +278,7 @@ type Service struct {
StaticDir string `json:"static_dir"`
NodeID sql.NullString `json:"node_id"`
Spread bool `json:"spread"`
PlacementTags json.RawMessage `json:"placement_tags"`
}
type ServiceDomain struct {
@@ -7,6 +7,7 @@ package sqlcdb
import (
"context"
"database/sql"
"encoding/json"
"github.com/google/uuid"
)
@@ -95,6 +96,8 @@ type Querier interface {
// platform admin. Anonymous callers pass user_id = uuid.Nil, is_admin = false.
ListProjectsWithStatsByUser(ctx context.Context, arg ListProjectsWithStatsByUserParams) ([]ListProjectsWithStatsByUserRow, 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)
ListServiceContainers(ctx context.Context, serviceID string) ([]ListServiceContainersRow, error)
ListServiceTemplatesByCategoryForUser(ctx context.Context, arg ListServiceTemplatesByCategoryForUserParams) ([]ServiceTemplate, error)
ListServiceTemplatesForUser(ctx context.Context, ownerID uuid.NullUUID) ([]ServiceTemplate, error)
@@ -107,10 +110,14 @@ type Querier interface {
// Resource-aware: lowest memory utilisation first, then cpu, then container
// count. Agents without telemetry sort last (NULL → worst score).
PickLeastLoadedAgent(ctx context.Context) (string, error)
// 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)
RevokeUserToken(ctx context.Context, arg RevokeUserTokenParams) (int64, error)
ScrubCommandPayload(ctx context.Context, arg ScrubCommandPayloadParams) error
SetAgentSchedulable(ctx context.Context, arg SetAgentSchedulableParams) error
SetAgentTags(ctx context.Context, arg SetAgentTagsParams) error
SetDatabaseBackupRemoteKeyByID(ctx context.Context, arg SetDatabaseBackupRemoteKeyByIDParams) error
SetDatabaseBackupScheduleByIDAndUser(ctx context.Context, arg SetDatabaseBackupScheduleByIDAndUserParams) error
SetDatabaseBackupStatusByID(ctx context.Context, arg SetDatabaseBackupStatusByIDParams) error
@@ -20,6 +20,10 @@ type NodeRunner interface {
// ReconcileSpread distributes the service's replicas across every online,
// schedulable agent — replica i lands on agents[i % len].
ReconcileSpread(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error)
// 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")
// out to every replica of the service on the agent.
ControlService(ctx context.Context, serviceID, agentID, action string) error
@@ -60,6 +60,9 @@ type RuntimeSpec struct {
// Spread distributes replicas across all online, schedulable agents
// instead of a single node. Mutually exclusive with NodeID.
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.
@@ -171,6 +174,14 @@ func (de *DeploymentEngine) ReconcileService(ctx context.Context, spec RuntimeSp
}
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)
}
if spec.Replicas < 1 {
spec.Replicas = 1
}
+3 -3
View File
@@ -53,7 +53,7 @@ func init() {
// Services
register(spec{"containr_services_list", "List services in a project", "GET", "/projects/{project_id}/services", []string{"project_id"}, nil, false})
register(spec{"containr_services_get", "Get a service by id", "GET", "/services/{id}", []string{"id"}, nil, false})
register(spec{"containr_services_create", "Create a service in a project. Args: name (required), type, image, git_repo, git_branch, environment, port, domain, replicas, volumes = [{type: volume|bind, source, target, read_only}], builder = auto|railpack|nixpacks|dockerfile|static, cpu, memory, cpu_reserve, memory_reserve, static_build_cmd, static_dir, 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_start", "Start a service", "POST", "/services/{id}/start", []string{"id"}, nil, false})
register(spec{"containr_services_stop", "Stop a service", "POST", "/services/{id}/stop", []string{"id"}, nil, false})
@@ -66,7 +66,7 @@ func init() {
register(spec{"containr_services_move", "Move a service to another project the caller owns. Args: id, project_id (required).", "POST", "/services/{id}/move", []string{"id"}, []string{"project_id"}, false})
register(spec{"containr_services_logs", "Get service runtime logs. Args: id, tail (default 100).", "GET", "/services/{id}/logs", []string{"id"}, nil, false})
register(spec{"containr_services_exec", "Run a one-off command in the service container (30s ceiling). Args: id, command (required).", "POST", "/services/{id}/exec", []string{"id"}, []string{"command"}, false})
register(spec{"containr_services_update", "Update a service. Args: id, name, image, command, domain, restart_policy, healthcheck_path, cpu, memory, replicas, port, volumes = [{type: volume|bind, source, target, read_only}] — volumes replaces the whole mount list; pass [] to clear. maintenance_mode (bool), basic_auth = [{username, password}] — replaces all creds; [] clears. builder = auto|railpack|nixpacks|dockerfile|static, cpu_reserve, memory_reserve, static_build_cmd, static_dir (empty string clears reserves/static fields). sleep_enabled (bool) + sleep_idle_minutes (1-1440) enable scale-to-zero on idle; a wake placeholder answers traffic while the service resumes. 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
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_agent_tokens_issue", "Issue an agent onboarding token (admin). Args: label.", "POST", "/agent-tokens", nil, []string{"label"}, false})
register(spec{"containr_agent_tokens_list", "List agent onboarding tokens (admin)", "GET", "/agent-tokens", nil, nil, false})
register(spec{"containr_nodes_update", "Update a node agent (admin). Args: id; optional auto_prune, name.", "PUT", "/agents/{id}", []string{"id"}, []string{"auto_prune", "name"}, false})
register(spec{"containr_nodes_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_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})