Consolidate backend: drop vendored APwhy, converge on goose + sqlc

Phase 0 cleanup verified end-to-end:

- Delete vendored APwhy runtime (cmd/apwhy, internal/apwhy) and the
  unwired copy (internal/api/server.go + static bundle + cloned
  internal/{storage,analytics,auth,rbac}) — ~7.5k lines of dead code.
- Keep Containr's native internal/gateway; move APwhy management
  routes under /api/v1/gateway/* so they stop colliding with the
  native /services paths. Fix handlers against the real schema:
  UUID ids via RETURNING, integer enabled flags, plan validation,
  service validate endpoint, unique-violation mapping.
- Port agents.go off GORM onto sqlc (sqlc/queries/agents.sql);
  drop gorm dependencies; fix sqlc schema drift (services.environment_id).
- Replace the SELECT 1 goose baseline with the real consolidated
  schema (52 tables) minus APwhy's parallel auth/RBAC objects;
  idempotent via IF NOT EXISTS / OR REPLACE / guarded DO blocks so it
  replays cleanly over legacy installs. Add a drop migration for the
  dead objects and delete the legacy filename-tracked migrations/
  loader, directory, and Dockerfile copy. Verified via cmd/migrate:
  fresh install 53 tables, legacy upgrade clean to 20260917000000.
- Canvas: add invisible React Flow handles to ServiceNode so inferred
  auto-connections stop emitting warning #008.
- Remove reactflow dep (superseded by @xyflow/react), test harness
  cmds, and legacy 0001_init.sql backups.

Authored By: TDvorak <[email protected]>
This commit is contained in:
Tomas Dvorak committed 2026-09-17 09:40:08 +02:00
1 parent 37a56dc6b6
commit 9e3aa3aa17
153 files changed
+6996 -11135

No files matched your search

+400 -251
View File
@@ -1,7 +1,10 @@
package api
import (
"context"
"crypto/subtle"
"database/sql"
"encoding/json"
"fmt"
"math"
"net/http"
@@ -9,26 +12,29 @@ import (
"strings"
"time"
"containr/internal/database"
"containr/internal/database/sqlcdb"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"gorm.io/gorm"
"github.com/sqlc-dev/pqtype"
)
// NodeAgent represents a container orchestration agent
type NodeAgent struct {
ID string `json:"id" gorm:"primaryKey"`
Name string `json:"name" gorm:"not null"`
Hostname string `json:"hostname" gorm:"not null"`
IPAddress string `json:"ip_address" gorm:"not null"`
Port int `json:"port" gorm:"not null"`
Status string `json:"status" gorm:"default:'offline'"`
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" gorm:"serializer:json"`
Resources NodeResources `json:"resources" gorm:"serializer:json"`
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" gorm:"serializer:json"`
Metadata map[string]interface{} `json:"metadata"`
}
// AgentCapabilities defines what the agent can do
@@ -89,20 +95,20 @@ type BandwidthInfo struct {
// ContainerInstance represents a container running on an agent
type ContainerInstance struct {
ID string `json:"id" gorm:"primaryKey"`
Name string `json:"name" gorm:"not null"`
Image string `json:"image" gorm:"not null"`
ProjectID string `json:"project_id" gorm:"not null"`
ServiceID string `json:"service_id" gorm:"not null"`
NodeAgentID string `json:"node_agent_id" gorm:"not null"`
Status ContainerStatus `json:"status" gorm:"serializer:json"`
Resources ContainerResources `json:"resources" gorm:"serializer:json"`
Ports []PortMapping `json:"ports" gorm:"serializer:json"`
Environment map[string]string `json:"environment" gorm:"serializer:json"`
Volumes []VolumeMount `json:"volumes" gorm:"serializer:json"`
Networks []string `json:"networks" gorm:"serializer:json"`
RestartPolicy RestartPolicy `json:"restart_policy" gorm:"serializer:json"`
HealthCheck *HealthCheck `json:"health_check" gorm:"serializer:json"`
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"`
@@ -155,12 +161,12 @@ type HealthCheck struct {
// AgentCommand represents a command sent to an agent
type AgentCommand struct {
ID string `json:"id" gorm:"primaryKey"`
Type string `json:"type" gorm:"not null"`
NodeAgentID string `json:"node_agent_id" gorm:"not null"`
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" gorm:"serializer:json"`
Status string `json:"status" gorm:"default:'pending'"`
Payload map[string]interface{} `json:"payload"`
Status string `json:"status"`
Result *string `json:"result"`
Error *string `json:"error"`
CreatedAt time.Time `json:"created_at"`
@@ -181,23 +187,6 @@ type AgentHeartbeat struct {
Version string `json:"version"`
}
type AgentHeartbeatRecord struct {
ID string `json:"id" gorm:"primaryKey"`
NodeAgentID string `json:"node_agent_id" gorm:"index;not null"`
Timestamp time.Time `json:"timestamp" gorm:"index;not null"`
Status string `json:"status"`
Resources NodeResources `json:"resources" gorm:"serializer:json"`
ContainerCount int `json:"container_count"`
SystemLoad SystemLoad `json:"system_load" gorm:"serializer:json"`
Uptime int64 `json:"uptime"`
Version string `json:"version"`
CreatedAt time.Time `json:"created_at"`
}
func (AgentHeartbeatRecord) TableName() string {
return "agent_heartbeats"
}
type SystemLoad struct {
Load1M float64 `json:"load_1m"`
Load5M float64 `json:"load_5m"`
@@ -206,13 +195,125 @@ type SystemLoad struct {
// NodeAgentHandler handles agent-related endpoints
type NodeAgentHandler struct {
db *gorm.DB
q *sqlcdb.Queries
}
func NewNodeAgentHandler(db *gorm.DB) *NodeAgentHandler {
return &NodeAgentHandler{db: db}
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 {
@@ -235,85 +336,83 @@ func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) {
return
}
// Check if agent already exists
var existingAgent NodeAgent
if err := h.db.Where("hostname = ? AND ip_address = ?", req.Hostname, req.IPAddress).First(&existingAgent).Error; err == nil {
// Update existing agent
existingAgent.Name = req.Name
existingAgent.Port = req.Port
existingAgent.Capabilities = req.Capabilities
existingAgent.Status = "connecting"
existingAgent.LastHeartbeat = time.Now()
ctx := c.Request.Context()
if err := h.db.Save(&existingAgent).Error; err != nil {
// 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": existingAgent.ID,
"agent_id": updated.ID,
"auth_token": authToken,
"status": "updated",
})
return
}
// Create new agent
agent := NodeAgent{
ID: uuid.New().String(),
Name: req.Name,
Hostname: req.Hostname,
IPAddress: req.IPAddress,
Port: req.Port,
Status: "connecting",
Capabilities: req.Capabilities,
Resources: NodeResources{
CPU: CPUResources{
Cores: 4,
Allocation: 0,
Usage: 0,
},
Memory: MemoryResources{
Total: 8 * 1024 * 1024 * 1024, // 8GB
Allocated: 0,
Used: 0,
Available: 8 * 1024 * 1024 * 1024,
},
Storage: StorageResources{
Total: 100 * 1024 * 1024 * 1024, // 100GB
Allocated: 0,
Used: 0,
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",
},
},
Bandwidth: BandwidthInfo{
Inbound: 0,
Outbound: 0,
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",
},
},
},
LastHeartbeat: time.Now(),
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
Metadata: make(map[string]interface{}),
}
if err := h.db.Create(&agent).Error; err != nil {
_, 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": agent.ID,
"agent_id": agentID,
"auth_token": authToken,
"status": "registered",
})
@@ -321,22 +420,25 @@ func (h *NodeAgentHandler) RegisterAgent(c *gin.Context) {
// GetAgents returns all registered agents
func (h *NodeAgentHandler) GetAgents(c *gin.Context) {
var agents []NodeAgent
if err := h.db.Find(&agents).Error; err != nil {
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) {
id := c.Param("id")
var agent NodeAgent
if err := h.db.First(&agent, "id = ?", id).Error; err != nil {
if err == gorm.ErrRecordNotFound {
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
}
@@ -344,16 +446,17 @@ func (h *NodeAgentHandler) GetAgent(c *gin.Context) {
return
}
c.JSON(http.StatusOK, gin.H{"agent": agent})
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()
var agent NodeAgent
if err := h.db.First(&agent, "id = ?", id).Error; err != nil {
if err == gorm.ErrRecordNotFound {
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
}
@@ -367,19 +470,53 @@ func (h *NodeAgentHandler) UpdateAgent(c *gin.Context) {
return
}
if err := h.db.Model(&agent).Updates(updates).Error; err != nil {
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": agent})
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) {
id := c.Param("id")
if err := h.db.Delete(&NodeAgent{}, "id = ?", id).Error; err != nil {
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
}
@@ -403,9 +540,9 @@ func (h *NodeAgentHandler) SendHeartbeat(c *gin.Context) {
heartbeat.Timestamp = time.Now()
}
var agent NodeAgent
if err := h.db.First(&agent, "id = ?", heartbeat.NodeAgentID).Error; err != nil {
if err == gorm.ErrRecordNotFound {
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
}
@@ -413,35 +550,31 @@ func (h *NodeAgentHandler) SendHeartbeat(c *gin.Context) {
return
}
// Update agent status and resources
agent.Status = heartbeat.Status
agent.Resources = heartbeat.Resources
agent.LastHeartbeat = heartbeat.Timestamp
agent.UpdatedAt = time.Now()
if err := h.db.Save(&agent).Error; err != nil {
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
}
record := AgentHeartbeatRecord{
err := h.q.InsertAgentHeartbeat(ctx, sqlcdb.InsertAgentHeartbeatParams{
ID: uuid.New().String(),
NodeAgentID: heartbeat.NodeAgentID,
Timestamp: heartbeat.Timestamp,
Status: heartbeat.Status,
Resources: heartbeat.Resources,
ContainerCount: heartbeat.ContainerCount,
SystemLoad: heartbeat.SystemLoad,
Resources: rawJSON(heartbeat.Resources).RawMessage,
ContainerCount: int32(heartbeat.ContainerCount),
SystemLoad: rawJSON(heartbeat.SystemLoad).RawMessage,
Uptime: heartbeat.Uptime,
Version: heartbeat.Version,
CreatedAt: time.Now(),
}
if err := h.db.Create(&record).Error; err != nil {
})
if err != nil && !isMissingTableError(err) {
// Keep heartbeat endpoint available even if history table is not yet migrated.
if !isMissingTableError(err) {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to persist heartbeat"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to persist heartbeat"})
return
}
c.Status(http.StatusOK)
@@ -454,17 +587,17 @@ func (h *NodeAgentHandler) GetPendingCommandsForAgent(c *gin.Context) {
return
}
agentID := c.Param("id")
var commands []AgentCommand
if err := h.db.
Where("node_agent_id = ? AND status = ?", agentID, "pending").
Order("created_at ASC").
Limit(25).
Find(&commands).Error; err != nil {
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})
}
@@ -491,45 +624,39 @@ func (h *NodeAgentHandler) CompleteCommand(c *gin.Context) {
return
}
var command AgentCommand
if err := h.db.First(&command, "id = ? AND node_agent_id = ?", commandID, agentID).Error; err != nil {
if err == gorm.ErrRecordNotFound {
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 fetch command"})
return
}
now := time.Now()
command.Status = req.Status
command.UpdatedAt = now
command.CompletedAt = &now
if req.Result != "" {
command.Result = &req.Result
}
if req.Error != "" {
command.Error = &req.Error
}
if err := h.db.Save(&command).Error; err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update command"})
return
}
if command.ContainerID != nil {
h.updateContainerStatusAfterCommand(command)
out := commandFromRow(command)
if out.ContainerID != nil {
h.updateContainerStatusAfterCommand(ctx, out)
}
c.JSON(http.StatusOK, gin.H{"command": command})
c.JSON(http.StatusOK, gin.H{"command": out})
}
func (h *NodeAgentHandler) updateContainerStatusAfterCommand(command AgentCommand) {
var container ContainerInstance
if err := h.db.First(&container, "id = ?", *command.ContainerID).Error; err != nil {
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 {
@@ -537,7 +664,10 @@ func (h *NodeAgentHandler) updateContainerStatusAfterCommand(command AgentComman
}
container.Status.State = "error"
container.Status.Error = &errorMessage
_ = h.db.Save(&container).Error
_ = h.q.UpdateContainerStatus(ctx, sqlcdb.UpdateContainerStatusParams{
ID: container.ID,
Status: rawJSON(container.Status),
})
return
}
@@ -555,20 +685,25 @@ func (h *NodeAgentHandler) updateContainerStatusAfterCommand(command AgentComman
container.Status.State = "removed"
container.Status.FinishedAt = &now
}
container.UpdatedAt = now
_ = h.db.Save(&container).Error
_ = 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) {
agentID := c.Param("id")
var containers []ContainerInstance
if err := h.db.Where("node_agent_id = ?", agentID).Find(&containers).Error; err != nil {
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})
}
@@ -595,10 +730,11 @@ func (h *NodeAgentHandler) CreateContainer(c *gin.Context) {
return
}
ctx := c.Request.Context()
// Verify agent exists
var agent NodeAgent
if err := h.db.First(&agent, "id = ?", agentID).Error; err != nil {
if err == gorm.ErrRecordNotFound {
if _, err := h.q.GetAgent(ctx, agentID); err != nil {
if err == sql.ErrNoRows {
c.JSON(http.StatusNotFound, gin.H{"error": "Agent not found"})
return
}
@@ -606,17 +742,15 @@ func (h *NodeAgentHandler) CreateContainer(c *gin.Context) {
return
}
containerID := uuid.New().String()
container := ContainerInstance{
ID: uuid.New().String(),
Name: req.Name,
Image: req.Image,
ProjectID: req.ProjectID,
ServiceID: req.ServiceID,
NodeAgentID: agentID,
Status: ContainerStatus{
State: "created",
Health: "none",
},
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,
@@ -628,26 +762,36 @@ func (h *NodeAgentHandler) CreateContainer(c *gin.Context) {
UpdatedAt: time.Now(),
}
if err := h.db.Create(&container).Error; err != nil {
_, 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
command := AgentCommand{
_, err = h.q.CreateCommand(ctx, sqlcdb.CreateCommandParams{
ID: uuid.New().String(),
Type: "create_container",
NodeAgentID: agentID,
ContainerID: &container.ID,
Payload: map[string]interface{}{
"container": container,
},
Status: "pending",
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
if err := h.db.Create(&command).Error; err != nil {
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
}
@@ -669,45 +813,44 @@ func (h *NodeAgentHandler) ExecuteCommand(c *gin.Context) {
return
}
command := AgentCommand{
command, err := h.q.CreateCommand(c.Request.Context(), sqlcdb.CreateCommandParams{
ID: uuid.New().String(),
Type: req.Type,
NodeAgentID: agentID,
Payload: req.Payload,
Status: "pending",
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
if err := h.db.Create(&command).Error; err != nil {
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": command})
c.JSON(http.StatusCreated, gin.H{"command": commandFromRow(command)})
}
// GetAgentCommands returns commands for an agent
func (h *NodeAgentHandler) GetAgentCommands(c *gin.Context) {
agentID := c.Param("id")
var commands []AgentCommand
if err := h.db.Where("node_agent_id = ?", agentID).Order("created_at DESC").Find(&commands).Error; err != nil {
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) {
agentID := c.Param("id")
commandID := c.Param("commandId")
var command AgentCommand
if err := h.db.First(&command, "id = ? AND node_agent_id = ?", commandID, agentID).Error; err != nil {
if err == gorm.ErrRecordNotFound {
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
}
@@ -715,7 +858,7 @@ func (h *NodeAgentHandler) GetCommandStatus(c *gin.Context) {
return
}
c.JSON(http.StatusOK, gin.H{"command": command})
c.JSON(http.StatusOK, gin.H{"command": commandFromRow(command)})
}
// ContainerAction handles container lifecycle actions
@@ -739,10 +882,15 @@ func (h *NodeAgentHandler) ContainerAction(c *gin.Context) {
return
}
ctx := c.Request.Context()
// Verify container exists
var container ContainerInstance
if err := h.db.First(&container, "id = ? AND node_agent_id = ?", containerID, agentID).Error; err != nil {
if err == gorm.ErrRecordNotFound {
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
}
@@ -751,22 +899,18 @@ func (h *NodeAgentHandler) ContainerAction(c *gin.Context) {
}
// Create command for the action
command := AgentCommand{
_, err = h.q.CreateCommand(ctx, sqlcdb.CreateCommandParams{
ID: uuid.New().String(),
Type: fmt.Sprintf("%s_container", action),
NodeAgentID: agentID,
ContainerID: &container.ID,
Payload: map[string]interface{}{
ContainerID: sql.NullString{String: containerRow.ID, Valid: true},
Payload: rawJSON(map[string]interface{}{
"container_id": containerID,
"container_name": container.Name,
"docker_name": container.Name,
},
Status: "pending",
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
if err := h.db.Create(&command).Error; err != nil {
"container_name": containerRow.Name,
"docker_name": containerRow.Name,
}),
})
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create command"})
return
}
@@ -794,22 +938,23 @@ func (h *NodeAgentHandler) GetAgentMetrics(c *gin.Context) {
return
}
var agent NodeAgent
if err := h.db.First(&agent, "id = ?", agentID).Error; err != nil {
if err == gorm.ErrRecordNotFound {
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)
var records []AgentHeartbeatRecord
queryErr := h.db.
Where("node_agent_id = ? AND timestamp >= ?", agentID, from).
Order("timestamp ASC").
Find(&records).Error
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
@@ -817,7 +962,11 @@ func (h *NodeAgentHandler) GetAgentMetrics(c *gin.Context) {
metrics := make([]map[string]interface{}, 0, len(records))
for _, record := range records {
metrics = append(metrics, buildMetricPoint(record.Timestamp, record.Resources, record.SystemLoad, record.ContainerCount))
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 {