Files
Containr/app/backend/internal/api/agents.go
T
Tomas DvorakandDevin aab785afc1 Add agent onboarding tokens with Usage-page issuance UI
Per-node onboarding tokens replace the placeholder <token> in the agent
connect command: raw cagt_ tokens are returned once at issuance, only
sha256 hashes persist in agent_auth_tokens, and listing exposes metadata
only. Public agent routes accept env-configured shared tokens or any
active DB-issued token; successful validation stamps last_used_at and
revoked tokens are rejected — verified end-to-end against a real stack
(register, heartbeat, revoke → 401, bogus → 401).

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

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-17 16:46:26 +02:00

1249 lines
37 KiB
Go

package api
import (
"context"
"crypto/rand"
"crypto/sha256"
"crypto/subtle"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"math"
"net/http"
"os"
"strings"
"time"
"containr/internal/database"
"containr/internal/database/sqlcdb"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/sqlc-dev/pqtype"
)
// NodeAgent represents a container orchestration agent
type NodeAgent struct {
ID string `json:"id"`
Name string `json:"name"`
Hostname string `json:"hostname"`
IPAddress string `json:"ip_address"`
Port int `json:"port"`
Status string `json:"status"`
Version string `json:"version"`
Capabilities AgentCapabilities `json:"capabilities"`
Resources NodeResources `json:"resources"`
LastHeartbeat time.Time `json:"last_heartbeat"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
Metadata map[string]interface{} `json:"metadata"`
}
// AgentCapabilities defines what the agent can do
type AgentCapabilities struct {
ContainerRuntimes []string `json:"container_runtimes"`
SupportedArchitectures []string `json:"supported_architectures"`
MaxContainers int `json:"max_containers"`
StorageDriver string `json:"storage_driver"`
NetworkPlugins []string `json:"network_plugins"`
Features []string `json:"features"`
}
// NodeResources represents the agent's available resources
type NodeResources struct {
CPU CPUResources `json:"cpu"`
Memory MemoryResources `json:"memory"`
Storage StorageResources `json:"storage"`
Network NetworkResources `json:"network"`
}
type CPUResources struct {
Cores int `json:"cores"`
Allocation float64 `json:"allocation"` // percentage
Usage float64 `json:"usage"` // current usage percentage
}
type MemoryResources struct {
Total int `json:"total"`
Allocated int `json:"allocated"`
Used int `json:"used"`
Available int `json:"available"`
}
type StorageResources struct {
Total int `json:"total"`
Allocated int `json:"allocated"`
Used int `json:"used"`
Available int `json:"available"`
}
type NetworkResources struct {
Interfaces []NetworkInterface `json:"interfaces"`
Bandwidth BandwidthInfo `json:"bandwidth"`
}
type NetworkInterface struct {
Name string `json:"name"`
IPAddress string `json:"ip_address"`
MACAddress string `json:"mac_address"`
Speed int `json:"speed"`
Status string `json:"status"`
}
type BandwidthInfo struct {
Inbound int `json:"inbound"` // bytes per second
Outbound int `json:"outbound"` // bytes per second
}
// ContainerInstance represents a container running on an agent
type ContainerInstance struct {
ID string `json:"id"`
Name string `json:"name"`
Image string `json:"image"`
ProjectID string `json:"project_id"`
ServiceID string `json:"service_id"`
NodeAgentID string `json:"node_agent_id"`
Status ContainerStatus `json:"status"`
Resources ContainerResources `json:"resources"`
Ports []PortMapping `json:"ports"`
Environment map[string]string `json:"environment"`
Volumes []VolumeMount `json:"volumes"`
Networks []string `json:"networks"`
RestartPolicy RestartPolicy `json:"restart_policy"`
HealthCheck *HealthCheck `json:"health_check"`
CreatedAt time.Time `json:"created_at"`
StartedAt *time.Time `json:"started_at"`
UpdatedAt time.Time `json:"updated_at"`
}
type ContainerStatus struct {
State string `json:"state"`
Health string `json:"health"`
ExitCode *int `json:"exit_code"`
Error *string `json:"error"`
StartedAt *time.Time `json:"started_at"`
FinishedAt *time.Time `json:"finished_at"`
}
type ContainerResources struct {
CPULimit int `json:"cpu_limit"`
CPUReservation int `json:"cpu_reservation"`
MemoryLimit int `json:"memory_limit"`
MemoryReservation int `json:"memory_reservation"`
DiskLimit *int `json:"disk_limit"`
}
type PortMapping struct {
ContainerPort int `json:"container_port"`
HostPort *int `json:"host_port"`
Protocol string `json:"protocol"`
Published bool `json:"published"`
}
type VolumeMount struct {
Name string `json:"name"`
Source string `json:"source"`
Target string `json:"target"`
Type string `json:"type"`
ReadOnly bool `json:"read_only"`
}
type RestartPolicy struct {
Name string `json:"name"`
MaximumRetryCount *int `json:"maximum_retry_count"`
}
type HealthCheck struct {
Test []string `json:"test"`
Interval int `json:"interval"`
Timeout int `json:"timeout"`
Retries int `json:"retries"`
StartPeriod int `json:"start_period"`
}
// AgentCommand represents a command sent to an agent
type AgentCommand struct {
ID string `json:"id"`
Type string `json:"type"`
NodeAgentID string `json:"node_agent_id"`
ContainerID *string `json:"container_id"`
Payload map[string]interface{} `json:"payload"`
Status string `json:"status"`
Result *string `json:"result"`
Error *string `json:"error"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
CompletedAt *time.Time `json:"completed_at"`
}
// AgentHeartbeat represents a heartbeat message from an agent
type AgentHeartbeat struct {
NodeAgentID string `json:"node_agent_id"`
AuthToken string `json:"auth_token,omitempty"`
Timestamp time.Time `json:"timestamp"`
Status string `json:"status"`
Resources NodeResources `json:"resources"`
ContainerCount int `json:"container_count"`
SystemLoad SystemLoad `json:"system_load"`
Uptime int64 `json:"uptime"`
Version string `json:"version"`
}
type SystemLoad struct {
Load1M float64 `json:"load_1m"`
Load5M float64 `json:"load_5m"`
Load15M float64 `json:"load_15m"`
}
// NodeAgentHandler handles agent-related endpoints
type NodeAgentHandler struct {
q *sqlcdb.Queries
}
func NewNodeAgentHandler(db *database.DB) *NodeAgentHandler {
return &NodeAgentHandler{q: sqlcdb.New(db.DB)}
}
// --- row <-> API type conversion helpers ---
func rawJSON(v interface{}) pqtype.NullRawMessage {
if v == nil {
return pqtype.NullRawMessage{}
}
b, err := json.Marshal(v)
if err != nil {
return pqtype.NullRawMessage{}
}
return pqtype.NullRawMessage{RawMessage: b, Valid: true}
}
func unmarshalRaw(raw pqtype.NullRawMessage, dst interface{}) {
if raw.Valid && len(raw.RawMessage) > 0 {
_ = json.Unmarshal(raw.RawMessage, dst)
}
}
func agentFromRow(row sqlcdb.NodeAgent) NodeAgent {
agent := NodeAgent{
ID: row.ID,
Name: row.Name,
Hostname: row.Hostname,
IPAddress: row.IpAddress,
Port: int(row.Port),
Status: row.Status.String,
Version: row.Version.String,
Metadata: map[string]interface{}{},
}
if row.LastHeartbeat.Valid {
agent.LastHeartbeat = row.LastHeartbeat.Time
}
if row.CreatedAt.Valid {
agent.CreatedAt = row.CreatedAt.Time
}
if row.UpdatedAt.Valid {
agent.UpdatedAt = row.UpdatedAt.Time
}
unmarshalRaw(row.Capabilities, &agent.Capabilities)
unmarshalRaw(row.Resources, &agent.Resources)
unmarshalRaw(row.Metadata, &agent.Metadata)
return agent
}
func containerFromRow(row sqlcdb.ContainerInstance) ContainerInstance {
container := ContainerInstance{
ID: row.ID,
Name: row.Name,
Image: row.Image,
ProjectID: row.ProjectID,
ServiceID: row.ServiceID,
NodeAgentID: row.NodeAgentID,
}
if row.CreatedAt.Valid {
container.CreatedAt = row.CreatedAt.Time
}
if row.UpdatedAt.Valid {
container.UpdatedAt = row.UpdatedAt.Time
}
if row.StartedAt.Valid {
t := row.StartedAt.Time
container.StartedAt = &t
}
unmarshalRaw(row.Status, &container.Status)
unmarshalRaw(row.Resources, &container.Resources)
unmarshalRaw(row.Ports, &container.Ports)
unmarshalRaw(row.Environment, &container.Environment)
unmarshalRaw(row.Volumes, &container.Volumes)
unmarshalRaw(row.Networks, &container.Networks)
unmarshalRaw(row.RestartPolicy, &container.RestartPolicy)
if row.HealthCheck.Valid {
var hc HealthCheck
if err := json.Unmarshal(row.HealthCheck.RawMessage, &hc); err == nil {
container.HealthCheck = &hc
}
}
return container
}
func commandFromRow(row sqlcdb.AgentCommand) AgentCommand {
cmd := AgentCommand{
ID: row.ID,
Type: row.Type,
NodeAgentID: row.NodeAgentID,
Status: row.Status.String,
}
if row.ContainerID.Valid {
cmd.ContainerID = &row.ContainerID.String
}
if row.Result.Valid {
cmd.Result = &row.Result.String
}
if row.Error.Valid {
cmd.Error = &row.Error.String
}
if row.CreatedAt.Valid {
cmd.CreatedAt = row.CreatedAt.Time
}
if row.UpdatedAt.Valid {
cmd.UpdatedAt = row.UpdatedAt.Time
}
if row.CompletedAt.Valid {
t := row.CompletedAt.Time
cmd.CompletedAt = &t
}
unmarshalRaw(row.Payload, &cmd.Payload)
return cmd
}
// --- handlers ---
// RegisterAgent handles agent registration
func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) {
var req struct {
Name string `json:"name" binding:"required"`
Hostname string `json:"hostname" binding:"required"`
IPAddress string `json:"ip_address" binding:"required"`
Port int `json:"port" binding:"required"`
Capabilities AgentCapabilities `json:"capabilities" binding:"required"`
AuthToken string `json:"auth_token"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
authToken := firstNonEmpty(req.AuthToken, agentAuthTokenFromRequest(c))
if !h.isValidAgentAuthToken(c.Request.Context(), authToken) {
c.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid auth token"})
return
}
ctx := c.Request.Context()
// Re-registering an existing host updates it in place.
if existing, err := h.q.GetAgentByHostAndIP(ctx, sqlcdb.GetAgentByHostAndIPParams{
Hostname: req.Hostname,
IpAddress: req.IPAddress,
}); err == nil {
updated, err := h.q.UpdateAgent(ctx, sqlcdb.UpdateAgentParams{
ID: existing.ID,
Name: req.Name,
Hostname: existing.Hostname,
IpAddress: existing.IpAddress,
Port: int32(req.Port),
Status: sql.NullString{String: "connecting", Valid: true},
Version: existing.Version,
Capabilities: rawJSON(req.Capabilities),
Resources: existing.Resources,
LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true},
Metadata: existing.Metadata,
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent"})
return
}
c.JSON(http.StatusOK, gin.H{
"agent_id": updated.ID,
"auth_token": authToken,
"status": "updated",
})
return
}
agentID := uuid.New().String()
resources := NodeResources{
CPU: CPUResources{Cores: 4},
Memory: MemoryResources{
Total: 8 * 1024 * 1024 * 1024, // 8GB
Available: 8 * 1024 * 1024 * 1024,
},
Storage: StorageResources{
Total: 100 * 1024 * 1024 * 1024, // 100GB
Available: 100 * 1024 * 1024 * 1024,
},
Network: NetworkResources{
Interfaces: []NetworkInterface{
{
Name: "eth0",
IPAddress: req.IPAddress,
MACAddress: "00:00:00:00:00:00",
Speed: 1000,
Status: "up",
},
},
},
}
_, err := h.q.CreateAgent(ctx, sqlcdb.CreateAgentParams{
ID: agentID,
Name: req.Name,
Hostname: req.Hostname,
IpAddress: req.IPAddress,
Port: int32(req.Port),
Status: sql.NullString{String: "connecting", Valid: true},
Version: sql.NullString{},
Capabilities: rawJSON(req.Capabilities),
Resources: rawJSON(resources),
LastHeartbeat: sql.NullTime{Time: time.Now(), Valid: true},
Metadata: rawJSON(map[string]interface{}{}),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create agent"})
return
}
c.JSON(http.StatusCreated, gin.H{
"agent_id": agentID,
"auth_token": authToken,
"status": "registered",
})
}
// GetAgents returns all registered agents
func (h *NodeAgentHandler) GetAgents(c *gin.Context) {
rows, err := h.q.ListAgents(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch agents"})
return
}
agents := make([]NodeAgent, 0, len(rows))
for _, row := range rows {
agents = append(agents, agentFromRow(row))
}
c.JSON(http.StatusOK, gin.H{"agents": agents})
}
// GetAgent returns a specific agent
func (h *NodeAgentHandler) GetAgent(c *gin.Context) {
row, err := h.q.GetAgent(c.Request.Context(), c.Param("id"))
if err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Agent not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch agent"})
return
}
c.JSON(http.StatusOK, gin.H{"agent": agentFromRow(row)})
}
// UpdateAgent updates an agent's information
func (h *NodeAgentHandler) UpdateAgent(c *gin.Context) {
id := c.Param("id")
ctx := c.Request.Context()
row, err := h.q.GetAgent(ctx, id)
if err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Agent not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch agent"})
return
}
var updates map[string]interface{}
if err := c.ShouldBindJSON(&updates); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
agent := agentFromRow(row)
applyAgentUpdates(&agent, updates)
updated, err := h.q.UpdateAgent(ctx, sqlcdb.UpdateAgentParams{
ID: agent.ID,
Name: agent.Name,
Hostname: agent.Hostname,
IpAddress: agent.IPAddress,
Port: int32(agent.Port),
Status: sql.NullString{String: agent.Status, Valid: agent.Status != ""},
Version: sql.NullString{String: agent.Version, Valid: agent.Version != ""},
Capabilities: rawJSON(agent.Capabilities),
Resources: rawJSON(agent.Resources),
LastHeartbeat: sql.NullTime{Time: agent.LastHeartbeat, Valid: !agent.LastHeartbeat.IsZero()},
Metadata: rawJSON(agent.Metadata),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent"})
return
}
c.JSON(http.StatusOK, gin.H{"agent": agentFromRow(updated)})
}
// applyAgentUpdates merges whitelisted keys from an arbitrary update map.
func applyAgentUpdates(agent *NodeAgent, updates map[string]interface{}) {
remarshal := func(key string, dst interface{}) {
if v, ok := updates[key]; ok {
if b, err := json.Marshal(v); err == nil {
_ = json.Unmarshal(b, dst)
}
}
}
remarshal("name", &agent.Name)
remarshal("hostname", &agent.Hostname)
remarshal("ip_address", &agent.IPAddress)
remarshal("port", &agent.Port)
remarshal("status", &agent.Status)
remarshal("version", &agent.Version)
remarshal("capabilities", &agent.Capabilities)
remarshal("resources", &agent.Resources)
remarshal("metadata", &agent.Metadata)
}
// DeleteAgent removes an agent
func (h *NodeAgentHandler) DeleteAgent(c *gin.Context) {
if err := h.q.DeleteAgent(c.Request.Context(), c.Param("id")); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to delete agent"})
return
}
c.Status(http.StatusNoContent)
}
// SendHeartbeat handles heartbeat messages from agents
func (h *NodeAgentHandler) SendHeartbeat(c *gin.Context) {
var heartbeat AgentHeartbeat
if err := c.ShouldBindJSON(&heartbeat); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
authToken := firstNonEmpty(heartbeat.AuthToken, agentAuthTokenFromRequest(c))
if !h.isValidAgentAuthToken(c.Request.Context(), authToken) {
c.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid auth token"})
return
}
if heartbeat.Timestamp.IsZero() {
heartbeat.Timestamp = time.Now()
}
ctx := c.Request.Context()
if _, err := h.q.GetAgent(ctx, heartbeat.NodeAgentID); err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Agent not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch agent"})
return
}
if err := h.q.UpdateAgentHeartbeat(ctx, sqlcdb.UpdateAgentHeartbeatParams{
ID: heartbeat.NodeAgentID,
Status: sql.NullString{String: heartbeat.Status, Valid: heartbeat.Status != ""},
Resources: rawJSON(heartbeat.Resources),
LastHeartbeat: sql.NullTime{Time: heartbeat.Timestamp, Valid: true},
}); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update agent"})
return
}
err := h.q.InsertAgentHeartbeat(ctx, sqlcdb.InsertAgentHeartbeatParams{
ID: uuid.New().String(),
NodeAgentID: heartbeat.NodeAgentID,
Timestamp: heartbeat.Timestamp,
Status: heartbeat.Status,
Resources: rawJSON(heartbeat.Resources).RawMessage,
ContainerCount: int32(heartbeat.ContainerCount),
SystemLoad: rawJSON(heartbeat.SystemLoad).RawMessage,
Uptime: heartbeat.Uptime,
Version: heartbeat.Version,
})
if err != nil && !isMissingTableError(err) {
// Keep heartbeat endpoint available even if history table is not yet migrated.
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to persist heartbeat"})
return
}
c.Status(http.StatusOK)
}
// GetPendingCommandsForAgent exposes queued commands to token-authenticated node agents.
func (h *NodeAgentHandler) GetPendingCommandsForAgent(c *gin.Context) {
if !h.isValidAgentAuthToken(c.Request.Context(), agentAuthTokenFromRequest(c)) {
c.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid auth token"})
return
}
rows, err := h.q.ListPendingCommands(c.Request.Context(), c.Param("id"))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch commands"})
return
}
commands := make([]AgentCommand, 0, len(rows))
for _, row := range rows {
commands = append(commands, commandFromRow(row))
}
c.JSON(http.StatusOK, gin.H{"commands": commands})
}
// CompleteCommand lets a node agent report command completion or failure.
func (h *NodeAgentHandler) CompleteCommand(c *gin.Context) {
if !h.isValidAgentAuthToken(c.Request.Context(), agentAuthTokenFromRequest(c)) {
c.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid auth token"})
return
}
agentID := c.Param("id")
commandID := c.Param("commandId")
var req struct {
Status string `json:"status" binding:"required"`
Result string `json:"result"`
Error string `json:"error"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
if req.Status != "completed" && req.Status != "failed" {
c.JSON(http.StatusBadRequest, gin.H{"error": "status must be completed or failed"})
return
}
ctx := c.Request.Context()
command, err := h.q.CompleteCommand(ctx, sqlcdb.CompleteCommandParams{
ID: commandID,
NodeAgentID: agentID,
Status: sql.NullString{String: req.Status, Valid: true},
Result: sql.NullString{String: req.Result, Valid: req.Result != ""},
Error: sql.NullString{String: req.Error, Valid: req.Error != ""},
})
if err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Command not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update command"})
return
}
out := commandFromRow(command)
if out.ContainerID != nil {
h.updateContainerStatusAfterCommand(ctx, out)
}
c.JSON(http.StatusOK, gin.H{"command": out})
}
func (h *NodeAgentHandler) updateContainerStatusAfterCommand(ctx context.Context, command AgentCommand) {
row, err := h.q.GetContainer(ctx, *command.ContainerID)
if err != nil {
return
}
container := containerFromRow(row)
if command.Status == "failed" {
errorMessage := ""
if command.Error != nil {
errorMessage = *command.Error
}
container.Status.State = "error"
container.Status.Error = &errorMessage
_ = h.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: container.ID,
Status: rawJSON(container.Status),
})
return
}
now := time.Now()
switch command.Type {
case "create_container", "start_container", "restart_container":
container.Status.State = "running"
container.Status.Health = "unknown"
container.Status.StartedAt = &now
container.Status.Error = nil
case "stop_container":
container.Status.State = "stopped"
container.Status.FinishedAt = &now
case "remove_container":
container.Status.State = "removed"
container.Status.FinishedAt = &now
}
_ = h.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: container.ID,
Status: rawJSON(container.Status),
})
}
// GetAgentContainers returns containers running on a specific agent
func (h *NodeAgentHandler) GetAgentContainers(c *gin.Context) {
rows, err := h.q.ListContainersForAgent(c.Request.Context(), c.Param("id"))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch containers"})
return
}
containers := make([]ContainerInstance, 0, len(rows))
for _, row := range rows {
containers = append(containers, containerFromRow(row))
}
c.JSON(http.StatusOK, gin.H{"containers": containers})
}
// CreateContainer creates a new container on an agent
func (h *NodeAgentHandler) CreateContainer(c *gin.Context) {
agentID := c.Param("id")
var req struct {
Name string `json:"name" binding:"required"`
Image string `json:"image" binding:"required"`
ProjectID string `json:"project_id" binding:"required"`
ServiceID string `json:"service_id" binding:"required"`
Resources ContainerResources `json:"resources" binding:"required"`
Ports []PortMapping `json:"ports"`
Environment map[string]string `json:"environment"`
Volumes []VolumeMount `json:"volumes"`
Networks []string `json:"networks"`
RestartPolicy RestartPolicy `json:"restart_policy"`
HealthCheck *HealthCheck `json:"health_check"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
ctx := c.Request.Context()
// Verify agent exists
if _, err := h.q.GetAgent(ctx, agentID); err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Agent not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch agent"})
return
}
containerID := uuid.New().String()
container := ContainerInstance{
ID: containerID,
Name: req.Name,
Image: req.Image,
ProjectID: req.ProjectID,
ServiceID: req.ServiceID,
NodeAgentID: agentID,
Status: ContainerStatus{State: "created", Health: "none"},
Resources: req.Resources,
Ports: req.Ports,
Environment: req.Environment,
Volumes: req.Volumes,
Networks: req.Networks,
RestartPolicy: req.RestartPolicy,
HealthCheck: req.HealthCheck,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
_, err := h.q.CreateContainer(ctx, sqlcdb.CreateContainerParams{
ID: container.ID,
Name: container.Name,
Image: container.Image,
ProjectID: container.ProjectID,
ServiceID: container.ServiceID,
NodeAgentID: container.NodeAgentID,
Status: rawJSON(container.Status),
Resources: rawJSON(container.Resources),
Ports: rawJSON(container.Ports),
Environment: rawJSON(container.Environment),
Volumes: rawJSON(container.Volumes),
Networks: rawJSON(container.Networks),
RestartPolicy: rawJSON(container.RestartPolicy),
HealthCheck: rawJSON(container.HealthCheck),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create container"})
return
}
// Create command to start container on agent
_, err = h.q.CreateCommand(ctx, sqlcdb.CreateCommandParams{
ID: uuid.New().String(),
Type: "create_container",
NodeAgentID: agentID,
ContainerID: sql.NullString{String: container.ID, Valid: true},
Payload: rawJSON(map[string]interface{}{"container": container}),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create container command"})
return
}
c.JSON(http.StatusCreated, gin.H{"container": container})
}
// ExecuteCommand executes a command on an agent
func (h *NodeAgentHandler) ExecuteCommand(c *gin.Context) {
agentID := c.Param("id")
var req struct {
Type string `json:"type" binding:"required"`
Payload map[string]interface{} `json:"payload"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
command, err := h.q.CreateCommand(c.Request.Context(), sqlcdb.CreateCommandParams{
ID: uuid.New().String(),
Type: req.Type,
NodeAgentID: agentID,
Payload: rawJSON(req.Payload),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create command"})
return
}
c.JSON(http.StatusCreated, gin.H{"command": commandFromRow(command)})
}
// GetAgentCommands returns commands for an agent
func (h *NodeAgentHandler) GetAgentCommands(c *gin.Context) {
rows, err := h.q.ListCommandsForAgent(c.Request.Context(), c.Param("id"))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch commands"})
return
}
commands := make([]AgentCommand, 0, len(rows))
for _, row := range rows {
commands = append(commands, commandFromRow(row))
}
c.JSON(http.StatusOK, gin.H{"commands": commands})
}
// GetCommandStatus returns the status of a specific command
func (h *NodeAgentHandler) GetCommandStatus(c *gin.Context) {
command, err := h.q.GetCommandForAgent(c.Request.Context(), sqlcdb.GetCommandForAgentParams{
ID: c.Param("commandId"),
NodeAgentID: c.Param("id"),
})
if err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Command not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch command"})
return
}
c.JSON(http.StatusOK, gin.H{"command": commandFromRow(command)})
}
// ContainerAction handles container lifecycle actions
func (h *NodeAgentHandler) ContainerAction(c *gin.Context) {
agentID := c.Param("id")
containerID := c.Param("containerId")
action := c.Param("action")
if action == "" && c.Request.Method == http.MethodDelete {
action = "remove"
}
// Validate action
validActions := map[string]bool{
"start": true,
"stop": true,
"restart": true,
"remove": true,
}
if !validActions[action] {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid action"})
return
}
ctx := c.Request.Context()
// Verify container exists
containerRow, err := h.q.GetContainerForAgent(ctx, sqlcdb.GetContainerForAgentParams{
ID: containerID,
NodeAgentID: agentID,
})
if err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Container not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch container"})
return
}
// Create command for the action
_, err = h.q.CreateCommand(ctx, sqlcdb.CreateCommandParams{
ID: uuid.New().String(),
Type: fmt.Sprintf("%s_container", action),
NodeAgentID: agentID,
ContainerID: sql.NullString{String: containerRow.ID, Valid: true},
Payload: rawJSON(map[string]interface{}{
"container_id": containerID,
"container_name": containerRow.Name,
"docker_name": containerRow.Name,
}),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create command"})
return
}
c.JSON(http.StatusOK, gin.H{"message": fmt.Sprintf("Container %s action initiated", action)})
}
// GetAgentMetrics returns metrics for an agent
func (h *NodeAgentHandler) GetAgentMetrics(c *gin.Context) {
agentID := c.Param("id")
timeRange := c.Query("time_range")
if timeRange == "" {
timeRange = "1h" // default to 1 hour
}
// Parse time range
duration, err := time.ParseDuration(timeRange)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid time range"})
return
}
if duration <= 0 || duration > 30*24*time.Hour {
c.JSON(http.StatusBadRequest, gin.H{"error": "time_range must be between 1s and 720h"})
return
}
ctx := c.Request.Context()
agentRow, err := h.q.GetAgent(ctx, agentID)
if err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Agent not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch agent"})
return
}
agent := agentFromRow(agentRow)
from := time.Now().Add(-duration)
records, queryErr := h.q.ListAgentHeartbeatsSince(ctx, sqlcdb.ListAgentHeartbeatsSinceParams{
NodeAgentID: agentID,
Timestamp: from,
})
if queryErr != nil && !isMissingTableError(queryErr) {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch agent metrics"})
return
}
metrics := make([]map[string]interface{}, 0, len(records))
for _, record := range records {
var resources NodeResources
var load SystemLoad
_ = json.Unmarshal(record.Resources, &resources)
_ = json.Unmarshal(record.SystemLoad, &load)
metrics = append(metrics, buildMetricPoint(record.Timestamp, resources, load, int(record.ContainerCount)))
}
if len(metrics) == 0 {
// Fallback to current snapshot when no historical records exist.
metrics = append(metrics, buildMetricPoint(
nonZeroTime(agent.LastHeartbeat, time.Now()),
agent.Resources,
SystemLoad{},
0,
))
}
c.JSON(http.StatusOK, gin.H{"metrics": metrics})
}
func buildMetricPoint(ts time.Time, resources NodeResources, load SystemLoad, containerCount int) map[string]interface{} {
memLimit := maxInt(resources.Memory.Total, 1)
memUsagePercent := math.Min(100, (float64(resources.Memory.Used)/float64(memLimit))*100)
cpuUsage := resources.CPU.Usage
if cpuUsage < 0 {
cpuUsage = 0
}
return map[string]interface{}{
"timestamp": ts.Format(time.RFC3339),
"cpu": map[string]interface{}{
"usage": cpuUsage,
"usage_percent": cpuUsage,
"cores": resources.CPU.Cores,
},
"memory": map[string]interface{}{
"usage": resources.Memory.Used,
"usage_percent": memUsagePercent,
"limit": resources.Memory.Total,
"available": resources.Memory.Available,
},
"system_load": map[string]interface{}{
"load_1m": load.Load1M,
"load_5m": load.Load5M,
"load_15m": load.Load15M,
},
"container_count": containerCount,
}
}
func isValidAgentAuthToken(token string) bool {
token = strings.TrimSpace(token)
if token == "" {
return false
}
for _, candidate := range configuredAgentAuthTokens() {
if subtle.ConstantTimeCompare([]byte(token), []byte(candidate)) == 1 {
return true
}
}
return false
}
// isValidAgentAuthTokenDB accepts env-configured shared tokens or a
// DB-issued onboarding token (matched by sha256 hash, never stored raw).
func (h *NodeAgentHandler) isValidAgentAuthToken(ctx context.Context, token string) bool {
if isValidAgentAuthToken(token) {
return true
}
token = strings.TrimSpace(token)
if token == "" {
return false
}
sum := sha256.Sum256([]byte(token))
hash := hex.EncodeToString(sum[:])
if _, err := h.q.GetActiveAgentAuthTokenByHash(ctx, hash); err != nil {
return false
}
_ = h.q.TouchAgentAuthToken(ctx, hash)
return true
}
func generateAgentAuthToken() (token string, hash string, err error) {
buf := make([]byte, 24)
if _, err = rand.Read(buf); err != nil {
return "", "", err
}
token = "cagt_" + hex.EncodeToString(buf)
sum := sha256.Sum256([]byte(token))
return token, hex.EncodeToString(sum[:]), nil
}
func agentAuthTokenFromRequest(c *gin.Context) string {
if token := strings.TrimSpace(c.GetHeader("X-Containr-Agent-Token")); token != "" {
return token
}
if authHeader := strings.TrimSpace(c.GetHeader("Authorization")); authHeader != "" {
const bearerPrefix = "Bearer "
if strings.HasPrefix(authHeader, bearerPrefix) {
return strings.TrimSpace(strings.TrimPrefix(authHeader, bearerPrefix))
}
}
return strings.TrimSpace(c.Query("auth_token"))
}
func configuredAgentAuthTokens() []string {
candidateCSV := strings.TrimSpace(os.Getenv("CONTAINR_AGENT_AUTH_TOKENS"))
if candidateCSV != "" {
parts := strings.Split(candidateCSV, ",")
out := make([]string, 0, len(parts))
for _, part := range parts {
if trimmed := strings.TrimSpace(part); trimmed != "" {
out = append(out, trimmed)
}
}
if len(out) > 0 {
return out
}
}
if single := strings.TrimSpace(os.Getenv("CONTAINR_AGENT_AUTH_TOKEN")); single != "" {
return []string{single}
}
if strings.EqualFold(strings.TrimSpace(os.Getenv("ENVIRONMENT")), "production") {
return nil
}
// Development fallback for local installs with no explicit secret configured.
return []string{"valid-token"}
}
func isMissingTableError(err error) bool {
if err == nil {
return false
}
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "does not exist") && strings.Contains(msg, "agent_heartbeats")
}
func nonZeroTime(primary, fallback time.Time) time.Time {
if primary.IsZero() {
return fallback
}
return primary
}
func maxInt(a, b int) int {
if a > b {
return a
}
return b
}
// --- onboarding token management ---
func agentAuthTokenJSON(t sqlcdb.AgentAuthToken) gin.H {
row := gin.H{
"id": t.ID,
"label": t.Label,
"revoked": t.RevokedAt.Valid,
}
if t.CreatedAt.Valid {
row["created_at"] = t.CreatedAt.Time
}
if t.LastUsedAt.Valid {
row["last_used_at"] = t.LastUsedAt.Time
}
if t.RevokedAt.Valid {
row["revoked_at"] = t.RevokedAt.Time
}
return row
}
// CreateAgentToken issues a new onboarding token. The raw token is returned
// once and never stored — only its sha256 hash persists.
func (h *NodeAgentHandler) CreateAgentToken(c *gin.Context) {
var req struct {
Label string `json:"label"`
}
_ = c.ShouldBindJSON(&req)
token, hash, err := generateAgentAuthToken()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to generate token"})
return
}
row, err := h.q.CreateAgentAuthToken(c.Request.Context(), sqlcdb.CreateAgentAuthTokenParams{
TokenHash: hash,
Label: strings.TrimSpace(req.Label),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to store token"})
return
}
payload := agentAuthTokenJSON(row)
payload["token"] = token
c.JSON(http.StatusCreated, payload)
}
// ListAgentTokens returns issued onboarding tokens (metadata only, never hashes).
func (h *NodeAgentHandler) ListAgentTokens(c *gin.Context) {
rows, err := h.q.ListAgentAuthTokens(c.Request.Context())
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch tokens"})
return
}
tokens := make([]gin.H, 0, len(rows))
for _, row := range rows {
tokens = append(tokens, agentAuthTokenJSON(row))
}
c.JSON(http.StatusOK, gin.H{"tokens": tokens})
}
// RevokeAgentToken marks an onboarding token as revoked; agents presenting it
// are rejected from then on.
func (h *NodeAgentHandler) RevokeAgentToken(c *gin.Context) {
id, err := uuid.Parse(strings.TrimSpace(c.Param("id")))
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid token id"})
return
}
row, err := h.q.RevokeAgentAuthToken(c.Request.Context(), id)
if err != nil {
status := http.StatusInternalServerError
if errors.Is(err, sql.ErrNoRows) {
status = http.StatusNotFound
}
c.JSON(status, gin.H{"error": "Token not found"})
return
}
c.JSON(http.StatusOK, gin.H{"status": "revoked", "id": row.ID})
}
// SetupPublicRoutes registers token-authenticated agent ingestion routes.
func (h *NodeAgentHandler) SetupPublicRoutes(router *gin.RouterGroup) {
agents := router.Group("/agents")
{
agents.POST("/register", h.RegisterAgent)
agents.POST("/heartbeat", h.SendHeartbeat)
agents.GET("/:id/commands", h.GetPendingCommandsForAgent)
agents.POST("/:id/commands/:commandId/result", h.CompleteCommand)
}
}
// SetupRoutes registers authenticated dashboard/control agent routes.
func (h *NodeAgentHandler) SetupRoutes(router *gin.RouterGroup) {
tokens := router.Group("/agent-tokens")
{
tokens.POST("", h.CreateAgentToken)
tokens.GET("", h.ListAgentTokens)
tokens.DELETE("/:id", h.RevokeAgentToken)
}
agents := router.Group("/agents")
{
agents.GET("", h.GetAgents)
agents.GET("/:id", h.GetAgent)
agents.PUT("/:id", h.UpdateAgent)
agents.DELETE("/:id", h.DeleteAgent)
agents.GET("/:id/containers", h.GetAgentContainers)
agents.POST("/:id/containers", h.CreateContainer)
agents.POST("/:id/containers/:containerId/start", h.ContainerAction)
agents.POST("/:id/containers/:containerId/stop", h.ContainerAction)
agents.POST("/:id/containers/:containerId/restart", h.ContainerAction)
agents.DELETE("/:id/containers/:containerId", h.ContainerAction)
agents.GET("/:id/metrics", h.GetAgentMetrics)
agents.POST("/:id/commands", h.ExecuteCommand)
agents.GET("/:id/commands", h.GetAgentCommands)
agents.GET("/:id/commands/:commandId", h.GetCommandStatus)
}
}