mirror of
https://github.com/Dvorinka/Containr.git
synced 2026-10-28 15:51:26 +00:00
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>
843 lines
27 KiB
Go
843 lines
27 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"containr/internal/database"
|
|
"containr/internal/deployment"
|
|
"containr/internal/deployqueue"
|
|
"containr/internal/secrets"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// remoteAgentsForService lists every agent holding inventory rows for the
|
|
// service — covers pinned placements, spread replicas, and leftovers from a
|
|
// since-cleared pin.
|
|
func remoteAgentsForService(ctx context.Context, db *database.DB, serviceID uuid.UUID) []string {
|
|
rows, err := db.Query(
|
|
`SELECT DISTINCT node_agent_id FROM container_instances WHERE service_id = $1`,
|
|
serviceID,
|
|
)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
defer rows.Close()
|
|
var ids []string
|
|
for rows.Next() {
|
|
var id string
|
|
if rows.Scan(&id) == nil && id != "" {
|
|
ids = append(ids, id)
|
|
}
|
|
}
|
|
return ids
|
|
}
|
|
|
|
// serviceLifecycle routes a start/stop/restart action to every agent holding
|
|
// the service's replicas, then to the local Docker engine. Both run — a
|
|
// service can't be pinned and local at once, but stale containers from a
|
|
// cleared pin shouldn't survive a stop.
|
|
func serviceLifecycle(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, serviceID uuid.UUID, action string) error {
|
|
for _, agentID := range remoteAgentsForService(ctx, db, serviceID) {
|
|
if err := engine.ControlServiceOnNode(ctx, serviceID.String(), agentID, action); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
switch action {
|
|
case "start":
|
|
return engine.StartService(ctx, serviceID.String())
|
|
case "stop":
|
|
return engine.StopService(ctx, serviceID.String())
|
|
case "restart":
|
|
if err := engine.StopService(ctx, serviceID.String()); err != nil {
|
|
return err
|
|
}
|
|
return engine.StartService(ctx, serviceID.String())
|
|
}
|
|
return fmt.Errorf("unsupported lifecycle action %q", action)
|
|
}
|
|
|
|
// removeServiceRuntime tears down replicas wherever they run — remote nodes
|
|
// (pinned or spread) and the local Docker host.
|
|
func removeServiceRuntime(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, serviceID uuid.UUID) {
|
|
for _, agentID := range remoteAgentsForService(ctx, db, serviceID) {
|
|
_ = engine.RemoveServiceContainersOnNode(ctx, serviceID.String(), agentID)
|
|
}
|
|
_ = engine.RemoveServiceContainers(ctx, serviceID.String())
|
|
}
|
|
|
|
// serviceRuntimeSpec builds the container spec for a service: stored env vars
|
|
// merged with ${{Service.KEY}} references resolved across the project.
|
|
func serviceRuntimeSpec(db *database.DB, service Service) (deployment.RuntimeSpec, error) {
|
|
env, err := resolveServiceEnv(db, service)
|
|
if err != nil {
|
|
return deployment.RuntimeSpec{}, err
|
|
}
|
|
|
|
spec := deployment.RuntimeSpec{
|
|
ProjectID: service.ProjectID.String(),
|
|
ServiceID: service.ID.String(),
|
|
Name: service.Name,
|
|
Image: service.Image,
|
|
Env: env,
|
|
Replicas: service.Replicas,
|
|
Port: int32(service.Port),
|
|
Domain: service.Domain,
|
|
HealthPath: service.HealthCheckPath,
|
|
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
|
|
// silently get ephemeral ports.
|
|
_ = db.QueryRow(`SELECT published_port FROM services WHERE id = $1`, service.ID).Scan(&spec.PublishedPort)
|
|
if cmd := strings.TrimSpace(service.Command); cmd != "" {
|
|
spec.Command = strings.Fields(cmd)
|
|
}
|
|
if spec.Replicas < 1 {
|
|
spec.Replicas = 1
|
|
}
|
|
if spec.RestartPolicy == "" {
|
|
spec.RestartPolicy = "unless-stopped"
|
|
}
|
|
spec.NanoCPUs = parseCPULimit(service.CPU)
|
|
spec.MemoryBytes = parseMemoryLimit(service.Memory)
|
|
spec.MemoryReserve = parseMemoryLimit(service.MemoryReserve)
|
|
spec.CPUShares = parseCPUReserveShares(service.CPUReserve)
|
|
if service.Builder == "static" && spec.Port == 0 {
|
|
// The static builder's image is nginx; expose its default port so
|
|
// Traefik has something to route to without extra config.
|
|
spec.Port = 80
|
|
}
|
|
spec.Volumes = loadServiceVolumes(db, service.ID)
|
|
spec.Domains = serviceDomainNames(db, service.ID, service.Domain)
|
|
spec.Maintenance, spec.BasicAuthUsers = serviceAccess(db, service.ID)
|
|
if spec.Maintenance {
|
|
spec.MaintenanceURL = maintenanceURL(db)
|
|
}
|
|
if spec.Port > 0 {
|
|
env["PORT"] = strconv.Itoa(service.Port)
|
|
}
|
|
return spec, nil
|
|
}
|
|
|
|
var refVarPattern = regexp.MustCompile(`\$\{\{\s*([A-Za-z0-9_-]+)\.([A-Za-z_][A-Za-z0-9_]*)\s*\}\}`)
|
|
|
|
// resolveServiceEnv loads the service's variables (real values, including
|
|
// secrets) and expands ${{service.KEY}} references. KEY may be a variable on
|
|
// the referenced service, or the builtins HOST (service name alias) and PORT.
|
|
func resolveServiceEnv(db *database.DB, service Service) (map[string]string, error) {
|
|
rows, err := db.Query(
|
|
`SELECT key, value FROM environment_variables WHERE service_id = $1`,
|
|
service.ID,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
env := map[string]string{}
|
|
for rows.Next() {
|
|
var k, v string
|
|
if err := rows.Scan(&k, &v); err != nil {
|
|
return nil, err
|
|
}
|
|
env[k] = secrets.Decrypt(v)
|
|
}
|
|
|
|
var refServices []struct{ ID, Name string }
|
|
svcRows, err := db.Query(
|
|
`SELECT id, name FROM services WHERE project_id = $1`,
|
|
service.ProjectID,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer svcRows.Close()
|
|
for svcRows.Next() {
|
|
var s struct{ ID, Name string }
|
|
if err := svcRows.Scan(&s.ID, &s.Name); err != nil {
|
|
return nil, err
|
|
}
|
|
refServices = append(refServices, s)
|
|
}
|
|
|
|
byName := map[string]string{}
|
|
for _, s := range refServices {
|
|
byName[strings.ToLower(s.Name)] = s.ID
|
|
}
|
|
|
|
resolve := func(serviceName, key string) (string, bool) {
|
|
if strings.EqualFold(serviceName, "shared") {
|
|
return resolveSharedVar(db, service.ProjectID, key)
|
|
}
|
|
id, ok := byName[strings.ToLower(serviceName)]
|
|
if !ok {
|
|
return "", false
|
|
}
|
|
switch strings.ToUpper(key) {
|
|
case "HOST":
|
|
return serviceName, true
|
|
case "PORT":
|
|
var port int
|
|
if err := db.QueryRow(`SELECT COALESCE(port, 0) FROM services WHERE id = $1`, id).Scan(&port); err == nil && port > 0 {
|
|
return strconv.Itoa(port), true
|
|
}
|
|
return "", false
|
|
}
|
|
var value string
|
|
err := db.QueryRow(
|
|
`SELECT value FROM environment_variables WHERE service_id = $1 AND key = $2`,
|
|
id, key,
|
|
).Scan(&value)
|
|
if err != nil {
|
|
return "", false
|
|
}
|
|
return secrets.Decrypt(value), true
|
|
}
|
|
|
|
for k, v := range env {
|
|
env[k] = refVarPattern.ReplaceAllStringFunc(v, func(match string) string {
|
|
parts := refVarPattern.FindStringSubmatch(match)
|
|
if len(parts) != 3 {
|
|
return match
|
|
}
|
|
if resolved, ok := resolve(parts[1], parts[2]); ok {
|
|
return resolved
|
|
}
|
|
return match
|
|
})
|
|
}
|
|
return env, nil
|
|
}
|
|
|
|
func parseCPULimit(cpu string) int64 {
|
|
v, err := strconv.ParseFloat(strings.TrimSpace(cpu), 64)
|
|
if err != nil || v <= 0 {
|
|
return 0
|
|
}
|
|
return int64(v * 1e9)
|
|
}
|
|
|
|
// parseCPUReserveShares maps a cpu fraction to Docker's relative share
|
|
// weight: 1024 shares ≈ one CPU.
|
|
func parseCPUReserveShares(cpu string) int64 {
|
|
v, err := strconv.ParseFloat(strings.TrimSpace(cpu), 64)
|
|
if err != nil || v <= 0 {
|
|
return 0
|
|
}
|
|
return int64(v * 1024)
|
|
}
|
|
|
|
// validateCPUSpec and validateMemorySpec reject values that silently parse
|
|
// to zero — a typo like "512mb" must fail loudly instead of dropping the limit.
|
|
func validateCPUSpec(v string) error {
|
|
if s := strings.TrimSpace(v); s != "" && parseCPULimit(s) <= 0 {
|
|
return fmt.Errorf("must be a positive cpu fraction (e.g. 0.5, 2)")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateMemorySpec(v string) error {
|
|
if s := strings.TrimSpace(v); s != "" && parseMemoryLimit(s) <= 0 {
|
|
return fmt.Errorf("must be bytes with unit (e.g. 256Mi, 1g)")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var memUnits = map[string]int64{"k": 1e3, "m": 1e6, "g": 1e9, "ki": 1 << 10, "mi": 1 << 20, "gi": 1 << 30}
|
|
|
|
func parseMemoryLimit(mem string) int64 {
|
|
s := strings.ToLower(strings.TrimSpace(mem))
|
|
for suffix, mult := range memUnits {
|
|
if strings.HasSuffix(s, suffix) {
|
|
if v, err := strconv.ParseFloat(strings.TrimSuffix(s, suffix), 64); err == nil {
|
|
return int64(v * float64(mult))
|
|
}
|
|
}
|
|
}
|
|
if v, err := strconv.ParseInt(s, 10, 64); err == nil {
|
|
return v
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func getRuntimeEngine(c *gin.Context) (*deployment.DeploymentEngine, *database.DB, bool) {
|
|
dbValue, exists := c.Get("db")
|
|
if !exists {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available"})
|
|
return nil, nil, false
|
|
}
|
|
engineValue, exists := c.Get("deployment_engine")
|
|
if !exists || engineValue == nil {
|
|
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Deployment engine unavailable. Docker may not be configured on this server."})
|
|
return nil, nil, false
|
|
}
|
|
return engineValue.(*deployment.DeploymentEngine), dbValue.(*database.DB), true
|
|
}
|
|
|
|
func loadOwnedService(c *gin.Context, db *database.DB) (Service, bool) {
|
|
serviceID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid service ID"})
|
|
return Service{}, false
|
|
}
|
|
userID, exists := c.Get("user_id")
|
|
if !exists {
|
|
c.JSON(http.StatusUnauthorized, gin.H{"error": "User not authenticated"})
|
|
return Service{}, false
|
|
}
|
|
|
|
var service Service
|
|
var owner string
|
|
err = db.QueryRow(
|
|
`SELECT s.id, s.project_id, s.name,
|
|
COALESCE(s.type, s.service_type, ''), COALESCE(s.status, ''),
|
|
COALESCE(s.image, s.image_name, ''), COALESCE(s.command, s.start_command, ''),
|
|
COALESCE(s.environment, ''), COALESCE(s.git_repo, s.source_url, ''),
|
|
COALESCE(s.git_branch, ''), COALESCE(s.build_path, ''),
|
|
COALESCE(s.cpu, ''), COALESCE(s.memory, ''),
|
|
COALESCE(s.replicas, 1), COALESCE(s.port, 0),
|
|
COALESCE(s.domain, ''), COALESCE(s.healthcheck_path, ''),
|
|
COALESCE(s.restart_policy, 'unless-stopped'),
|
|
COALESCE(s.builder, 'auto'), COALESCE(s.cpu_reserve, ''),
|
|
COALESCE(s.memory_reserve, ''), COALESCE(s.static_build_cmd, ''),
|
|
COALESCE(s.static_dir, ''),
|
|
s.created_at, s.updated_at, p.owner_id
|
|
FROM services s JOIN projects p ON s.project_id = p.id
|
|
WHERE s.id = $1`,
|
|
serviceID,
|
|
).Scan(
|
|
&service.ID, &service.ProjectID, &service.Name, &service.Type, &service.Status,
|
|
&service.Image, &service.Command, &service.Environment, &service.GitRepo,
|
|
&service.GitBranch, &service.BuildPath, &service.CPU, &service.Memory,
|
|
&service.Replicas, &service.Port, &service.Domain, &service.HealthCheckPath,
|
|
&service.RestartPolicy, &service.Builder, &service.CPUReserve,
|
|
&service.MemoryReserve, &service.StaticBuildCmd, &service.StaticDir,
|
|
&service.CreatedAt, &service.UpdatedAt, &owner,
|
|
)
|
|
if err != nil {
|
|
c.JSON(http.StatusNotFound, gin.H{"error": "Service not found"})
|
|
return Service{}, false
|
|
}
|
|
if owner != userID.(string) && !isAdminUser(db, userID.(string)) {
|
|
c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"})
|
|
return Service{}, false
|
|
}
|
|
return service, true
|
|
}
|
|
|
|
// loadReadableService loads a service when the caller may read its project:
|
|
// approved projects are public, unapproved ones need owner/member/admin.
|
|
func loadReadableService(c *gin.Context, db *database.DB) (Service, bool) {
|
|
serviceID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid service ID"})
|
|
return Service{}, false
|
|
}
|
|
|
|
var service Service
|
|
var projectID uuid.UUID
|
|
err = db.QueryRow(
|
|
`SELECT s.id, s.project_id, s.name,
|
|
COALESCE(s.type, s.service_type, ''), COALESCE(s.status, ''),
|
|
COALESCE(s.image, s.image_name, ''), COALESCE(s.command, s.start_command, ''),
|
|
COALESCE(s.environment, ''), COALESCE(s.git_repo, s.source_url, ''),
|
|
COALESCE(s.git_branch, ''), COALESCE(s.build_path, ''),
|
|
COALESCE(s.cpu, ''), COALESCE(s.memory, ''),
|
|
COALESCE(s.replicas, 1), COALESCE(s.port, 0),
|
|
COALESCE(s.domain, ''), COALESCE(s.healthcheck_path, ''),
|
|
COALESCE(s.restart_policy, 'unless-stopped'),
|
|
COALESCE(s.builder, 'auto'), COALESCE(s.cpu_reserve, ''),
|
|
COALESCE(s.memory_reserve, ''), COALESCE(s.static_build_cmd, ''),
|
|
COALESCE(s.static_dir, ''),
|
|
s.created_at, s.updated_at, s.project_id
|
|
FROM services s
|
|
WHERE s.id = $1`,
|
|
serviceID,
|
|
).Scan(
|
|
&service.ID, &service.ProjectID, &service.Name, &service.Type, &service.Status,
|
|
&service.Image, &service.Command, &service.Environment, &service.GitRepo,
|
|
&service.GitBranch, &service.BuildPath, &service.CPU, &service.Memory,
|
|
&service.Replicas, &service.Port, &service.Domain, &service.HealthCheckPath,
|
|
&service.RestartPolicy, &service.Builder, &service.CPUReserve,
|
|
&service.MemoryReserve, &service.StaticBuildCmd, &service.StaticDir,
|
|
&service.CreatedAt, &service.UpdatedAt, &projectID,
|
|
)
|
|
if err != nil {
|
|
c.JSON(http.StatusNotFound, gin.H{"error": "Service not found"})
|
|
return Service{}, false
|
|
}
|
|
if _, allowed := projectReadAccess(c, db, projectID); !allowed {
|
|
c.JSON(http.StatusNotFound, gin.H{"error": "Service not found"})
|
|
return Service{}, false
|
|
}
|
|
return service, true
|
|
}
|
|
|
|
// handleServiceRuntime returns live container state for a service.
|
|
func handleGetServiceRuntime(c *gin.Context) {
|
|
engine, db, ok := getRuntimeEngine(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
service, ok := loadReadableService(c, db)
|
|
if !ok {
|
|
return
|
|
}
|
|
state, err := engine.RuntimeState(c.Request.Context(), service.ID.String(), service.Replicas)
|
|
if err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
mergeRemoteRuntime(c.Request.Context(), engine, service.ID.String(), state)
|
|
engine.ProbeServiceHealth(c.Request.Context(), state, service.ProjectID.String(), service.Name, service.Port, service.HealthCheckPath)
|
|
persistPublishedPort(db, service.ID, state.Ports)
|
|
if host := requestHostname(c); host != "" {
|
|
for i, u := range state.URLs {
|
|
state.URLs[i] = rewriteLoopbackURL(u, host)
|
|
}
|
|
}
|
|
c.JSON(http.StatusOK, gin.H{"runtime": state})
|
|
}
|
|
|
|
func reconcileNow(c *gin.Context, engine *deployment.DeploymentEngine, db *database.DB, service Service) bool {
|
|
spec, err := serviceRuntimeSpec(db, service)
|
|
if err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
|
return false
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
state, err := engine.ReconcileService(ctx, spec)
|
|
if err != nil {
|
|
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
|
return false
|
|
}
|
|
status := "stopped"
|
|
if state != nil {
|
|
status = state.Status
|
|
persistPublishedPort(db, service.ID, state.Ports)
|
|
}
|
|
_, _ = db.Exec(`UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, status, time.Now(), service.ID)
|
|
c.JSON(http.StatusOK, gin.H{"runtime": state})
|
|
return true
|
|
}
|
|
|
|
// handleServiceStart starts all stopped replicas.
|
|
func handleServiceStart(c *gin.Context) {
|
|
engine, db, ok := getRuntimeEngine(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
service, ok := loadOwnedService(c, db)
|
|
if !ok {
|
|
return
|
|
}
|
|
if err := serviceLifecycle(c.Request.Context(), db, engine, service.ID, "start"); err != nil {
|
|
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
_, _ = db.Exec(`UPDATE services SET status = 'running', updated_at = $1 WHERE id = $2`, time.Now(), service.ID)
|
|
c.JSON(http.StatusOK, gin.H{"status": "running"})
|
|
}
|
|
|
|
// handleServiceStop stops all replicas without removing them.
|
|
func handleServiceStop(c *gin.Context) {
|
|
engine, db, ok := getRuntimeEngine(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
service, ok := loadOwnedService(c, db)
|
|
if !ok {
|
|
return
|
|
}
|
|
if err := serviceLifecycle(c.Request.Context(), db, engine, service.ID, "stop"); err != nil {
|
|
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
_, _ = db.Exec(`UPDATE services SET status = 'stopped', updated_at = $1 WHERE id = $2`, time.Now(), service.ID)
|
|
c.JSON(http.StatusOK, gin.H{"status": "stopped"})
|
|
}
|
|
|
|
// handleServiceRestart restarts all replicas.
|
|
func handleServiceRestart(c *gin.Context) {
|
|
engine, db, ok := getRuntimeEngine(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
service, ok := loadOwnedService(c, db)
|
|
if !ok {
|
|
return
|
|
}
|
|
ctx := c.Request.Context()
|
|
if err := serviceLifecycle(ctx, db, engine, service.ID, "restart"); err != nil {
|
|
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
_, _ = db.Exec(`UPDATE services SET status = 'running', updated_at = $1 WHERE id = $2`, time.Now(), service.ID)
|
|
c.JSON(http.StatusOK, gin.H{"status": "running"})
|
|
}
|
|
|
|
// handleServiceRedeploy re-applies the current spec: pulls the image for
|
|
// image-sourced services and reconciles containers (env, ports, replicas).
|
|
// Runs through the per-service deploy queue so it never overlaps a build.
|
|
func handleServiceRedeploy(c *gin.Context) {
|
|
engine, db, ok := getRuntimeEngine(c)
|
|
if !ok {
|
|
return
|
|
}
|
|
service, ok := loadOwnedService(c, db)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
image := service.Image
|
|
if image == "" {
|
|
// Git-sourced service: reuse the last successful deployment image.
|
|
err := db.QueryRow(
|
|
`SELECT image_name || ':' || image_tag FROM deployments
|
|
WHERE service_id = $1 AND status = 'deployed' AND image_name <> ''
|
|
ORDER BY created_at DESC LIMIT 1`,
|
|
service.ID,
|
|
).Scan(&image)
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "No deployable image yet — run a deployment first"})
|
|
return
|
|
}
|
|
}
|
|
service.Image = image
|
|
|
|
finished := make(chan struct{})
|
|
var recErr error
|
|
var state *deployment.RuntimeState
|
|
getDeployQueue(c).Enqueue(service.ID, deployqueue.Job{Run: func(jctx context.Context) {
|
|
defer close(finished)
|
|
state, recErr = reconcileServiceJob(jctx, db, engine, service)
|
|
}})
|
|
|
|
select {
|
|
case <-finished:
|
|
if recErr != nil {
|
|
c.JSON(http.StatusBadGateway, gin.H{"error": recErr.Error()})
|
|
return
|
|
}
|
|
c.JSON(http.StatusOK, gin.H{"runtime": state})
|
|
case <-c.Request.Context().Done():
|
|
c.JSON(http.StatusAccepted, gin.H{"status": "queued"})
|
|
}
|
|
}
|
|
|
|
// reconcileServiceJob is the shared queue-job body for redeploy paths:
|
|
// build the spec from current DB state, reconcile containers, persist the
|
|
// resulting status and published port.
|
|
func reconcileServiceJob(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, service Service) (*deployment.RuntimeState, error) {
|
|
spec, err := serviceRuntimeSpec(db, service)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
state, err := engine.ReconcileService(ctx, spec)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if state != nil {
|
|
persistPublishedPort(db, service.ID, state.Ports)
|
|
status := "stopped"
|
|
if state.Status != "" {
|
|
status = state.Status
|
|
}
|
|
_, _ = db.Exec(`UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, status, time.Now(), service.ID)
|
|
}
|
|
return state, nil
|
|
}
|
|
|
|
// enqueueServiceRedeploy resolves a deployable image (service.Image, else
|
|
// the last successful deployment's tag) and queues a reconcile. Returns
|
|
// false when no image is available or the engine is missing.
|
|
func enqueueServiceRedeploy(c *gin.Context, db *database.DB, service Service) bool {
|
|
engineValue, exists := c.Get("deployment_engine")
|
|
if !exists || engineValue == nil {
|
|
return false
|
|
}
|
|
engine, ok := engineValue.(*deployment.DeploymentEngine)
|
|
if !ok {
|
|
return false
|
|
}
|
|
image := service.Image
|
|
if image == "" {
|
|
if err := db.QueryRow(
|
|
`SELECT image_name || ':' || image_tag FROM deployments
|
|
WHERE service_id = $1 AND status = 'deployed' AND image_name <> ''
|
|
ORDER BY created_at DESC LIMIT 1`,
|
|
service.ID,
|
|
).Scan(&image); err != nil {
|
|
return false
|
|
}
|
|
}
|
|
service.Image = image
|
|
getDeployQueue(c).Enqueue(service.ID, deployqueue.Job{Run: func(jctx context.Context) {
|
|
_, _ = reconcileServiceJob(jctx, db, engine, service)
|
|
}})
|
|
return true
|
|
}
|
|
|
|
// mergeRemoteRuntime folds inventory-backed remote replica state into a
|
|
// local runtime view so pinned and spread services report their containers.
|
|
// Remote rows carry deterministic UUIDs, local ones Docker IDs — no overlap.
|
|
func mergeRemoteRuntime(ctx context.Context, engine *deployment.DeploymentEngine, serviceID string, state *deployment.RuntimeState) {
|
|
remote, err := engine.RemoteRuntimeState(ctx, serviceID)
|
|
if err != nil || remote == nil {
|
|
return
|
|
}
|
|
running := 0
|
|
for _, c := range state.Containers {
|
|
if c.State == "running" {
|
|
running++
|
|
}
|
|
}
|
|
for _, rc := range remote.Containers {
|
|
state.Containers = append(state.Containers, rc)
|
|
if rc.State == "running" {
|
|
running++
|
|
}
|
|
}
|
|
switch {
|
|
case len(state.Containers) == 0:
|
|
case running == 0:
|
|
state.Status = "stopped"
|
|
case running < state.Desired:
|
|
state.Status = "degraded"
|
|
default:
|
|
state.Status = "running"
|
|
}
|
|
}
|
|
|
|
// persistPublishedPort records the host port Docker actually bound so later
|
|
// redeploys can reuse it. Idempotent — only writes when the value changed.
|
|
func persistPublishedPort(db *database.DB, serviceID uuid.UUID, ports []uint16) {
|
|
if len(ports) == 0 {
|
|
return
|
|
}
|
|
_, _ = db.Exec(
|
|
`UPDATE services SET published_port = $1 WHERE id = $2 AND published_port <> $1`,
|
|
int(ports[0]), serviceID,
|
|
)
|
|
}
|
|
|
|
// liveServiceStatus reconciles the stored status with real container state.
|
|
// No-op when Docker is unavailable — the stored status is returned as-is.
|
|
func liveServiceStatus(c *gin.Context, db *database.DB, service *Service) {
|
|
engineValue, exists := c.Get("deployment_engine")
|
|
if !exists || engineValue == nil {
|
|
return
|
|
}
|
|
engine := engineValue.(*deployment.DeploymentEngine)
|
|
ctx, cancel := context.WithTimeout(c.Request.Context(), 3*time.Second)
|
|
defer cancel()
|
|
state, err := engine.RuntimeState(ctx, service.ID.String(), service.Replicas)
|
|
if err != nil || state == nil {
|
|
return
|
|
}
|
|
mergeRemoteRuntime(ctx, engine, service.ID.String(), state)
|
|
if len(state.Containers) == 0 {
|
|
return
|
|
}
|
|
if state.Status != service.Status {
|
|
service.Status = state.Status
|
|
_, _ = db.Exec(`UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, state.Status, time.Now(), service.ID)
|
|
}
|
|
persistPublishedPort(db, service.ID, state.Ports)
|
|
if service.Domain != "" {
|
|
service.PublicURL = "https://" + service.Domain
|
|
} else if len(state.URLs) > 0 {
|
|
service.PublicURL = rewriteLoopbackURL(state.URLs[0], requestHostname(c))
|
|
}
|
|
}
|
|
|
|
// runtimeScaleService adjusts replica count both in bookkeeping and reality.
|
|
func runtimeScaleService(c *gin.Context, serviceID string, replicas int) error {
|
|
dbValue, _ := c.Get("db")
|
|
engineValue, _ := c.Get("deployment_engine")
|
|
db, _ := dbValue.(*database.DB)
|
|
engine, _ := engineValue.(*deployment.DeploymentEngine)
|
|
if db == nil || engine == nil {
|
|
return fmt.Errorf("runtime unavailable")
|
|
}
|
|
|
|
var service Service
|
|
var owner string
|
|
err := db.QueryRow(
|
|
`SELECT s.id, s.project_id, s.name,
|
|
COALESCE(s.type, s.service_type, ''), COALESCE(s.status, ''),
|
|
COALESCE(s.image, s.image_name, ''), COALESCE(s.command, s.start_command, ''),
|
|
COALESCE(s.environment, ''), COALESCE(s.git_repo, s.source_url, ''),
|
|
COALESCE(s.git_branch, ''), COALESCE(s.build_path, ''),
|
|
COALESCE(s.cpu, ''), COALESCE(s.memory, ''),
|
|
COALESCE(s.replicas, 1), COALESCE(s.port, 0),
|
|
COALESCE(s.domain, ''), COALESCE(s.healthcheck_path, ''),
|
|
COALESCE(s.restart_policy, 'unless-stopped'),
|
|
COALESCE(s.builder, 'auto'), COALESCE(s.cpu_reserve, ''),
|
|
COALESCE(s.memory_reserve, ''), COALESCE(s.static_build_cmd, ''),
|
|
COALESCE(s.static_dir, ''),
|
|
s.created_at, s.updated_at, p.owner_id
|
|
FROM services s JOIN projects p ON s.project_id = p.id
|
|
WHERE s.id = $1`, serviceID,
|
|
).Scan(
|
|
&service.ID, &service.ProjectID, &service.Name, &service.Type, &service.Status,
|
|
&service.Image, &service.Command, &service.Environment, &service.GitRepo,
|
|
&service.GitBranch, &service.BuildPath, &service.CPU, &service.Memory,
|
|
&service.Replicas, &service.Port, &service.Domain, &service.HealthCheckPath,
|
|
&service.RestartPolicy, &service.Builder, &service.CPUReserve,
|
|
&service.MemoryReserve, &service.StaticBuildCmd, &service.StaticDir,
|
|
&service.CreatedAt, &service.UpdatedAt, &owner,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if service.Image == "" {
|
|
if err := db.QueryRow(
|
|
`SELECT image_name || ':' || image_tag FROM deployments
|
|
WHERE service_id = $1 AND status = 'deployed' AND image_name <> ''
|
|
ORDER BY created_at DESC LIMIT 1`, service.ID,
|
|
).Scan(&service.Image); err != nil {
|
|
return fmt.Errorf("no deployed image to scale")
|
|
}
|
|
}
|
|
|
|
spec, err := serviceRuntimeSpec(db, service)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
spec.Replicas = replicas
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
state, err := engine.ReconcileService(ctx, spec)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if state != nil {
|
|
persistPublishedPort(db, service.ID, state.Ports)
|
|
}
|
|
_, _ = db.Exec(`UPDATE services SET replicas = $1, updated_at = $2 WHERE id = $3`, replicas, time.Now(), service.ID)
|
|
return nil
|
|
}
|
|
|
|
// handleServiceEnvCheck reports environment problems without exposing
|
|
// values: empty vars, ${{service.KEY}} refs that fail to resolve, and
|
|
// secrets whose ciphertext can't be decrypted (key rotation).
|
|
func handleServiceEnvCheck(c *gin.Context) {
|
|
dbVal, exists := c.Get("db")
|
|
if !exists {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available"})
|
|
return
|
|
}
|
|
db := dbVal.(*database.DB)
|
|
service, ok := loadReadableService(c, db)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
rows, err := db.Query(
|
|
`SELECT key, value, COALESCE(is_secret, false) FROM environment_variables WHERE service_id = $1`,
|
|
service.ID,
|
|
)
|
|
if err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
|
|
var unresolved, empty, unreadable []string
|
|
var raw []struct{ key, val string }
|
|
for rows.Next() {
|
|
var k, v string
|
|
var secret bool
|
|
if err := rows.Scan(&k, &v, &secret); err != nil {
|
|
continue
|
|
}
|
|
decrypted := secrets.Decrypt(v)
|
|
if secret && secrets.IsEncrypted(v) && decrypted == v {
|
|
// Ciphertext failed to decrypt — Decrypt returns the input.
|
|
unreadable = append(unreadable, k)
|
|
continue
|
|
}
|
|
if strings.TrimSpace(decrypted) == "" {
|
|
empty = append(empty, k)
|
|
}
|
|
raw = append(raw, struct{ key, val string }{k, decrypted})
|
|
}
|
|
|
|
// Project siblings for ${{name.KEY}} resolution.
|
|
siblingIDs := map[string]string{}
|
|
svcRows, err := db.Query(`SELECT id, name FROM services WHERE project_id = $1`, service.ProjectID)
|
|
if err == nil {
|
|
for svcRows.Next() {
|
|
var id, name string
|
|
if err := svcRows.Scan(&id, &name); err == nil {
|
|
siblingIDs[strings.ToLower(name)] = id
|
|
}
|
|
}
|
|
svcRows.Close()
|
|
}
|
|
|
|
resolvable := func(serviceName, key string) bool {
|
|
id, ok := siblingIDs[strings.ToLower(serviceName)]
|
|
if !ok {
|
|
return false
|
|
}
|
|
if strings.EqualFold(key, "HOST") {
|
|
return true
|
|
}
|
|
if strings.EqualFold(key, "PORT") {
|
|
var port int
|
|
if err := db.QueryRow(`SELECT COALESCE(port, 0) FROM services WHERE id = $1`, id).Scan(&port); err == nil {
|
|
return port > 0
|
|
}
|
|
return false
|
|
}
|
|
var n int
|
|
if err := db.QueryRow(
|
|
`SELECT COUNT(*) FROM environment_variables WHERE service_id = $1 AND key = $2`,
|
|
id, key,
|
|
).Scan(&n); err == nil {
|
|
return n > 0
|
|
}
|
|
return false
|
|
}
|
|
|
|
for _, kv := range raw {
|
|
for _, match := range refVarPattern.FindAllStringSubmatch(kv.val, -1) {
|
|
if len(match) == 3 && !resolvable(match[1], match[2]) {
|
|
unresolved = append(unresolved, fmt.Sprintf("%s → %s", kv.key, match[0]))
|
|
}
|
|
}
|
|
}
|
|
|
|
if unresolved == nil {
|
|
unresolved = []string{}
|
|
}
|
|
if empty == nil {
|
|
empty = []string{}
|
|
}
|
|
if unreadable == nil {
|
|
unreadable = []string{}
|
|
}
|
|
okResult := len(unresolved) == 0 && len(empty) == 0 && len(unreadable) == 0
|
|
c.JSON(http.StatusOK, gin.H{
|
|
"ok": okResult,
|
|
"unresolved": unresolved,
|
|
"empty": empty,
|
|
"unreadable": unreadable,
|
|
})
|
|
}
|