feat: real service runtime — private networks, replicas, lifecycle, rollback

Services now reconcile into actual Docker containers instead of
bookkeeping rows:

- Per-project bridge network with service-name DNS aliases and
  ${{service.KEY}} env expansion (HOST/PORT builtins)
- Replicas, published ports, Traefik domain routing, health checks,
  restart policies persisted on the service spec; spec-hash drift
  recreates containers on redeploy
- Lifecycle API: start/stop/restart/redeploy, live runtime state,
  real image rollback (replaces the fake sleep)
- Manual scale and PUT replicas reconcile live containers; service
  and project delete remove their containers and the project network
- Canvas: real URLs, replica badges, env filter, right-click
  deploy/restart/stop/delete menu, empty state
- Service detail: real lifecycle buttons, live replica/URL strip,
  Networking editor, ${{ }} variable helper
- Fix prebuilt-image pulls racing the tag (drain the pull stream)

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

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Tomas DvorakandDevin committed 2026-09-21 19:31:19 +02:00
1 parent c16b32f56a
commit ff356230fe
20 files changed
+2270 -243

No files matched your search

+126 -22
View File
@@ -153,8 +153,11 @@ func handleCreateDeployment(c *gin.Context) {
var service Service
var projectOwner string
err = db.(*database.DB).QueryRow(
`SELECT s.id, s.project_id, s.name, s.type, s.status, s.image, s.command,
s.environment, s.git_repo, s.git_branch, s.build_path, s.cpu, s.memory,
`SELECT s.id, s.project_id, s.name, s.type, s.status, s.image, s.command,
s.environment, s.git_repo, s.git_branch, s.build_path, s.cpu, s.memory,
COALESCE(s.replicas, 1), COALESCE(s.port, 0),
COALESCE(s.domain, ''), COALESCE(s.healthcheck_path, ''),
COALESCE(s.restart_policy, 'unless-stopped'),
s.created_at, s.updated_at, p.owner_id
FROM services s
JOIN projects p ON s.project_id = p.id
@@ -164,7 +167,8 @@ func handleCreateDeployment(c *gin.Context) {
&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.CreatedAt, &service.UpdatedAt, &projectOwner,
&service.Replicas, &service.Port, &service.Domain, &service.HealthCheckPath,
&service.RestartPolicy, &service.CreatedAt, &service.UpdatedAt, &projectOwner,
)
if err != nil {
@@ -256,6 +260,22 @@ func runDeploymentAndSync(
service Service,
req CreateDeploymentRequest,
userID string,
) {
runDeploymentAndSyncWithImage(parentCtx, db, engine, dbDeployment, service, req, userID, "")
}
// runDeploymentAndSyncWithImage deploys imageOverride as a prebuilt image
// (rollback / redeploy of an existing tag) when set; otherwise builds from
// source or pulls the service image.
func runDeploymentAndSyncWithImage(
parentCtx context.Context,
db *database.DB,
engine *deployment.DeploymentEngine,
dbDeployment *DeploymentModel,
service Service,
req CreateDeploymentRequest,
userID string,
imageOverride string,
) {
ctx, cancel := context.WithTimeout(parentCtx, 30*time.Minute)
defer cancel()
@@ -265,21 +285,44 @@ func runDeploymentAndSync(
sourcePath = "."
}
// Resolve the runtime env: stored service variables (with ${{svc.KEY}}
// references expanded) plus any per-deployment overrides.
env, envErr := resolveServiceEnv(db, service)
if envErr != nil {
env = map[string]string{}
}
for k, v := range req.EnvVars {
env[k] = v
}
var command []string
if cmd := strings.TrimSpace(service.Command); cmd != "" {
command = strings.Fields(cmd)
}
replicas := service.Replicas
if replicas < 1 {
replicas = 1
}
deployReq := &deployment.DeploymentRequest{
ProjectID: service.ProjectID.String(),
ServiceID: service.ID.String(),
Environment: service.Environment,
Config: deployment.ServiceConfig{
Name: service.Name,
Image: service.Image,
Environment: req.EnvVars,
Replicas: 1,
},
BuildConfig: &deployment.BuildConfig{
BuildType: "nixpacks",
SourcePath: sourcePath,
Branch: req.Branch,
Commit: req.CommitHash,
Name: service.Name,
Image: service.Image,
Command: command,
Environment: env,
Replicas: replicas,
PublicPort: int32(service.Port),
Domain: service.Domain,
HealthPath: service.HealthCheckPath,
RestartPolicy: service.RestartPolicy,
Resources: deployment.ResourceLimits{
MemoryBytes: parseMemoryLimit(service.Memory),
CPUQuota: parseCPULimit(service.CPU),
},
},
Trigger: deployment.TriggerConfig{
Type: req.Trigger,
@@ -289,6 +332,26 @@ func runDeploymentAndSync(
},
}
if imageOverride != "" {
deployReq.BuildConfig = &deployment.BuildConfig{
BuildType: "prebuilt",
PrebuiltImage: imageOverride,
}
} else if service.GitRepo == "" {
// Image-sourced service: pull the image instead of building.
deployReq.BuildConfig = &deployment.BuildConfig{
BuildType: "prebuilt",
PrebuiltImage: service.Image,
}
} else {
deployReq.BuildConfig = &deployment.BuildConfig{
BuildType: "nixpacks",
SourcePath: sourcePath,
Branch: req.Branch,
Commit: req.CommitHash,
}
}
engineDeployment, err := engine.Deploy(ctx, deployReq)
if err != nil {
failedAt := time.Now()
@@ -550,17 +613,58 @@ func handleRollbackDeployment(c *gin.Context) {
time.Now(), serviceID,
)
go func() {
time.Sleep(2 * time.Second)
db.(*database.DB).Exec(
`UPDATE deployments SET status = 'deployed', completed_at = $1, updated_at = $1 WHERE id = $2`,
time.Now(), rollbackID,
// Real rollback: redeploy the target deployment's image.
engineValue, _ := c.Get("deployment_engine")
engine, _ := engineValue.(*deployment.DeploymentEngine)
targetImage := targetDeployment.ImageName
if targetDeployment.ImageTag != "" {
targetImage += ":" + targetDeployment.ImageTag
}
if engine == nil || targetImage == "" {
completedAt := time.Now()
reason := "Rollback image unavailable"
if engine == nil {
reason = "Deployment engine unavailable. Docker may not be configured on this server."
}
_, _ = db.(*database.DB).Exec(
`UPDATE deployments SET status = 'failed', error = $1, completed_at = $2, updated_at = $2 WHERE id = $3`,
reason, completedAt, rollbackID,
)
db.(*database.DB).Exec(
`UPDATE services SET status = 'running', updated_at = $1 WHERE id = $2`,
time.Now(), serviceID,
_, _ = db.(*database.DB).Exec(
`UPDATE services SET status = 'failed', updated_at = $1 WHERE id = $2`,
completedAt, serviceID,
)
}()
} else {
var service Service
_ = db.(*database.DB).QueryRow(
`SELECT s.id, s.project_id, s.name, s.type, s.status, s.image, s.command,
s.environment, s.git_repo, s.git_branch, s.build_path, s.cpu, s.memory,
COALESCE(s.replicas, 1), COALESCE(s.port, 0),
COALESCE(s.domain, ''), COALESCE(s.healthcheck_path, ''),
COALESCE(s.restart_policy, 'unless-stopped'),
s.created_at, s.updated_at
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.CreatedAt, &service.UpdatedAt,
)
rollbackReq := CreateDeploymentRequest{
CommitHash: func() string {
if targetDeployment.CommitHash != nil {
return *targetDeployment.CommitHash
}
return ""
}(),
Trigger: "rollback",
}
// Reuse the stored image directly — no rebuild.
go runDeploymentAndSyncWithImage(context.Background(), db.(*database.DB), engine, &rollback, service, rollbackReq, userID.(string), targetImage)
}
c.JSON(http.StatusCreated, gin.H{
"deployment": DeploymentResponse{
+12
View File
@@ -3,6 +3,8 @@ package api
import (
"containr/internal/database"
"containr/internal/database/sqlcdb"
"containr/internal/deployment"
"context"
"database/sql"
"errors"
"net/http"
@@ -310,6 +312,16 @@ func handleDeleteProject(c *gin.Context) {
return
}
// Remove live containers and the project network before dropping rows.
// Best-effort: Docker may be unavailable; DB cleanup proceeds regardless.
if engineValue, exists := c.Get("deployment_engine"); exists && engineValue != nil {
engine := engineValue.(*deployment.DeploymentEngine)
ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second)
_ = engine.RemoveProjectContainers(ctx, projectID.String())
_ = engine.RemoveProjectNetwork(ctx, projectID.String())
cancel()
}
deletedRows, err := queries.DeleteProjectByID(c.Request.Context(), projectID)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to delete project"})
+7
View File
@@ -209,6 +209,13 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg
protected.DELETE("/services/:id", handleDeleteService)
protected.GET("/services/:id/metrics", handleGetServiceMetrics)
// Runtime lifecycle (Railway-style)
protected.GET("/services/:id/runtime", handleGetServiceRuntime)
protected.POST("/services/:id/start", handleServiceStart)
protected.POST("/services/:id/stop", handleServiceStop)
protected.POST("/services/:id/restart", handleServiceRestart)
protected.POST("/services/:id/redeploy", handleServiceRedeploy)
// Deployment routes
protected.GET("/services/:id/deployments", handleGetDeployments)
protected.POST("/services/:id/deployments", handleCreateDeployment)
+438
View File
@@ -0,0 +1,438 @@
package api
import (
"context"
"fmt"
"net/http"
"regexp"
"strconv"
"strings"
"time"
"containr/internal/database"
"containr/internal/deployment"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
)
// 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,
}
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)
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] = 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) {
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 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)
}
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'),
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.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
}
// handleServiceRuntime returns live container state for a service.
func handleGetServiceRuntime(c *gin.Context) {
engine, db, ok := getRuntimeEngine(c)
if !ok {
return
}
service, ok := loadOwnedService(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
}
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
}
_, _ = 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 := engine.StartService(c.Request.Context(), service.ID.String()); 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 := engine.StopService(c.Request.Context(), service.ID.String()); 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 := engine.StopService(ctx, service.ID.String()); err != nil {
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
return
}
if err := engine.StartService(ctx, service.ID.String()); 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).
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
reconcileNow(c, engine, db, service)
}
// 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 || 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)
}
if len(state.URLs) > 0 && service.Domain == "" {
service.PublicURL = state.URLs[0]
}
}
// 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'),
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.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()
if _, err := engine.ReconcileService(ctx, spec); err != nil {
return err
}
_, _ = db.Exec(`UPDATE services SET replicas = $1, updated_at = $2 WHERE id = $3`, replicas, time.Now(), service.ID)
return nil
}
+20 -3
View File
@@ -204,7 +204,16 @@ func (h *ScalingHandler) ManualScale(c *gin.Context) {
event, err := h.autoScaler.ManualScale(c.Request.Context(), serviceID, request.Replicas, strings.TrimSpace(request.Reason))
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
// Service unknown to the autoscaler: the runtime is authoritative —
// reconcile containers directly and persist the replica count.
if runtimeErr := runtimeScaleService(c, serviceID, request.Replicas); runtimeErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error(), "runtime_error": runtimeErr.Error()})
return
}
c.JSON(http.StatusOK, gin.H{
"message": "Service scaled successfully",
"state": gin.H{"service_id": serviceID, "desired_replicas": request.Replicas},
})
return
}
@@ -214,11 +223,19 @@ func (h *ScalingHandler) ManualScale(c *gin.Context) {
return
}
c.JSON(http.StatusOK, gin.H{
// Apply the replica change to live containers (best effort — the
// bookkeeping update above already succeeded).
runtimeErr := runtimeScaleService(c, serviceID, request.Replicas)
resp := gin.H{
"message": "Service scaled successfully",
"event": event,
"state": state,
})
}
if runtimeErr != nil {
resp["runtime_error"] = runtimeErr.Error()
}
c.JSON(http.StatusOK, resp)
}
// GetScalingStatus returns the overall status of the auto-scaler
+142 -47
View File
@@ -2,6 +2,8 @@ package api
import (
"containr/internal/database"
"containr/internal/deployment"
"context"
"net/http"
"time"
@@ -24,37 +26,54 @@ type Service struct {
BuildPath string `json:"build_path" db:"build_path"`
CPU string `json:"cpu" db:"cpu"`
Memory string `json:"memory" db:"memory"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
// Runtime spec
Replicas int `json:"replicas" db:"replicas"`
Port int `json:"port" db:"port"` // container port to expose
Domain string `json:"domain" db:"domain"` // public hostname via Traefik
HealthCheckPath string `json:"healthcheck_path" db:"healthcheck_path"` // probed on Port
RestartPolicy string `json:"restart_policy" db:"restart_policy"`
PublicURL string `json:"public_url,omitempty" db:"-"` // computed at read time
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
}
// CreateServiceRequest represents a request to create a service
type CreateServiceRequest struct {
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name" binding:"required,min=1,max=255"`
Type string `json:"type" binding:"required,oneof=web worker database cron"`
Image string `json:"image"`
Command string `json:"command"`
Environment string `json:"environment" binding:"required,oneof=production preview development"`
GitRepo string `json:"git_repo"`
GitBranch string `json:"git_branch"`
BuildPath string `json:"build_path"`
CPU string `json:"cpu"`
Memory string `json:"memory"`
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name" binding:"required,min=1,max=255"`
Type string `json:"type" binding:"required,oneof=web worker database cron"`
Image string `json:"image"`
Command string `json:"command"`
Environment string `json:"environment" binding:"required,oneof=production preview development"`
GitRepo string `json:"git_repo"`
GitBranch string `json:"git_branch"`
BuildPath string `json:"build_path"`
CPU string `json:"cpu"`
Memory string `json:"memory"`
Replicas int `json:"replicas"`
Port int `json:"port"`
Domain string `json:"domain"`
HealthCheckPath string `json:"healthcheck_path"`
RestartPolicy string `json:"restart_policy"`
}
// UpdateServiceRequest represents a request to update a service
type UpdateServiceRequest struct {
Name string `json:"name" binding:"omitempty,min=1,max=255"`
Type string `json:"type" binding:"omitempty,oneof=web worker database cron"`
Image string `json:"image"`
Command string `json:"command"`
Environment string `json:"environment" binding:"omitempty,oneof=production preview development"`
GitRepo string `json:"git_repo"`
GitBranch string `json:"git_branch"`
BuildPath string `json:"build_path"`
CPU string `json:"cpu"`
Memory string `json:"memory"`
Name string `json:"name" binding:"omitempty,min=1,max=255"`
Type string `json:"type" binding:"omitempty,oneof=web worker database cron"`
Image string `json:"image"`
Command string `json:"command"`
Environment string `json:"environment" binding:"omitempty,oneof=production preview development"`
GitRepo string `json:"git_repo"`
GitBranch string `json:"git_branch"`
BuildPath string `json:"build_path"`
CPU string `json:"cpu"`
Memory string `json:"memory"`
Replicas *int `json:"replicas"`
Port *int `json:"port"`
Domain *string `json:"domain"`
HealthCheckPath *string `json:"healthcheck_path"`
RestartPolicy string `json:"restart_policy"`
}
// handleGetServices retrieves all services for a project
@@ -110,6 +129,9 @@ func handleGetServices(c *gin.Context) {
COALESCE(build_path, ''),
COALESCE(cpu, ''),
COALESCE(memory, ''),
COALESCE(replicas, 1), COALESCE(port, 0),
COALESCE(domain, ''), COALESCE(healthcheck_path, ''),
COALESCE(restart_policy, 'unless-stopped'),
created_at, updated_at
FROM services
WHERE project_id = $1
@@ -129,7 +151,8 @@ func handleGetServices(c *gin.Context) {
&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.CreatedAt, &service.UpdatedAt,
&service.Replicas, &service.Port, &service.Domain, &service.HealthCheckPath,
&service.RestartPolicy, &service.CreatedAt, &service.UpdatedAt,
)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to scan service"})
@@ -138,6 +161,11 @@ func handleGetServices(c *gin.Context) {
services = append(services, service)
}
// Reconcile stored status with live container state (no-op without Docker).
for i := range services {
liveServiceStatus(c, db.(*database.DB), &services[i])
}
c.JSON(http.StatusOK, gin.H{"services": services})
}
@@ -213,21 +241,26 @@ func handleCreateService(c *gin.Context) {
// Create new service
service := Service{
ID: uuid.New(),
ProjectID: req.ProjectID,
Name: req.Name,
Type: req.Type,
Status: "stopped", // Initial status
Image: req.Image,
Command: req.Command,
Environment: req.Environment,
GitRepo: req.GitRepo,
GitBranch: req.GitBranch,
BuildPath: req.BuildPath,
CPU: req.CPU,
Memory: req.Memory,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
ID: uuid.New(),
ProjectID: req.ProjectID,
Name: req.Name,
Type: req.Type,
Status: "stopped", // Initial status
Image: req.Image,
Command: req.Command,
Environment: req.Environment,
GitRepo: req.GitRepo,
GitBranch: req.GitBranch,
BuildPath: req.BuildPath,
CPU: req.CPU,
Memory: req.Memory,
Replicas: req.Replicas,
Port: req.Port,
Domain: req.Domain,
HealthCheckPath: req.HealthCheckPath,
RestartPolicy: req.RestartPolicy,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
}
// Set default values if not provided
@@ -237,6 +270,12 @@ func handleCreateService(c *gin.Context) {
if service.Memory == "" {
service.Memory = "512Mi"
}
if service.Replicas < 1 {
service.Replicas = 1
}
if service.RestartPolicy == "" {
service.RestartPolicy = "unless-stopped"
}
environmentID, err := getProjectEnvironmentID(db.(*database.DB), service.ProjectID, service.Environment)
if err != nil {
@@ -248,15 +287,17 @@ func handleCreateService(c *gin.Context) {
// Insert service into database
_, err = db.(*database.DB).Exec(
`INSERT INTO services
`INSERT INTO services
(id, project_id, name, environment_id, service_type, source_type, source_url, image_name,
build_command, start_command, type, status, image, command, environment,
git_repo, git_branch, build_path, cpu, memory, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22)`,
git_repo, git_branch, build_path, cpu, memory, replicas, port, domain,
healthcheck_path, restart_policy, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27)`,
service.ID, service.ProjectID, service.Name, environmentID, service.Type,
sourceType, firstNonEmpty(service.GitRepo, service.Image), service.Image,
"", service.Command, service.Type, service.Status, service.Image, service.Command,
service.Environment, service.GitRepo, service.GitBranch, service.BuildPath, service.CPU, service.Memory,
service.Replicas, service.Port, service.Domain, service.HealthCheckPath, service.RestartPolicy,
service.CreatedAt, service.UpdatedAt,
)
@@ -324,6 +365,9 @@ func handleGetService(c *gin.Context) {
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'),
s.created_at, s.updated_at
FROM services s
JOIN projects p ON s.project_id = p.id
@@ -333,7 +377,8 @@ func handleGetService(c *gin.Context) {
&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.CreatedAt, &service.UpdatedAt,
&service.Replicas, &service.Port, &service.Domain, &service.HealthCheckPath,
&service.RestartPolicy, &service.CreatedAt, &service.UpdatedAt,
)
if err != nil {
@@ -341,6 +386,7 @@ func handleGetService(c *gin.Context) {
return
}
liveServiceStatus(c, db.(*database.DB), &service)
c.JSON(http.StatusOK, gin.H{"service": service})
}
@@ -386,6 +432,9 @@ func handleUpdateService(c *gin.Context) {
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'),
s.created_at, s.updated_at
FROM services s
JOIN projects p ON s.project_id = p.id
@@ -396,6 +445,8 @@ func handleUpdateService(c *gin.Context) {
&existingService.Status, &existingService.Image, &existingService.Command,
&existingService.Environment, &existingService.GitRepo, &existingService.GitBranch,
&existingService.BuildPath, &existingService.CPU, &existingService.Memory,
&existingService.Replicas, &existingService.Port, &existingService.Domain,
&existingService.HealthCheckPath, &existingService.RestartPolicy,
&existingService.CreatedAt, &existingService.UpdatedAt,
)
@@ -435,18 +486,37 @@ func handleUpdateService(c *gin.Context) {
if req.Memory != "" {
existingService.Memory = req.Memory
}
if req.Replicas != nil {
existingService.Replicas = *req.Replicas
}
if req.Port != nil {
existingService.Port = *req.Port
}
if req.Domain != nil {
existingService.Domain = *req.Domain
}
if req.HealthCheckPath != nil {
existingService.HealthCheckPath = *req.HealthCheckPath
}
if req.RestartPolicy != "" {
existingService.RestartPolicy = req.RestartPolicy
}
existingService.UpdatedAt = time.Now()
// Update service in database
_, err = db.(*database.DB).Exec(
`UPDATE services
`UPDATE services
SET name = $1, type = $2, image = $3, command = $4, environment = $5,
git_repo = $6, git_branch = $7, build_path = $8, cpu = $9, memory = $10, updated_at = $11
WHERE id = $12`,
git_repo = $6, git_branch = $7, build_path = $8, cpu = $9, memory = $10,
replicas = $11, port = $12, domain = $13, healthcheck_path = $14,
restart_policy = $15, updated_at = $16
WHERE id = $17`,
existingService.Name, existingService.Type, existingService.Image, existingService.Command,
existingService.Environment, existingService.GitRepo, existingService.GitBranch,
existingService.BuildPath, existingService.CPU, existingService.Memory,
existingService.Replicas, existingService.Port, existingService.Domain,
existingService.HealthCheckPath, existingService.RestartPolicy,
existingService.UpdatedAt, existingService.ID,
)
@@ -455,6 +525,24 @@ func handleUpdateService(c *gin.Context) {
return
}
// A replica change applies immediately when the service has live
// containers; other spec fields take effect on the next deploy.
if req.Replicas != nil && existingService.Status == "running" {
if engineValue, exists := c.Get("deployment_engine"); exists && engineValue != nil {
engine := engineValue.(*deployment.DeploymentEngine)
serviceCopy := existingService
go func() {
spec, err := serviceRuntimeSpec(db.(*database.DB), serviceCopy)
if err != nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
_, _ = engine.ReconcileService(ctx, spec)
}()
}
}
c.JSON(http.StatusOK, gin.H{"service": existingService})
}
@@ -496,11 +584,18 @@ func handleDeleteService(c *gin.Context) {
}
// Check if user owns the project
if projectOwnerID != userID.(string) {
if projectOwnerID != userID.(string) && !isAdminUser(db.(*database.DB), userID.(string)) {
c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"})
return
}
// Remove runtime containers before dropping the row.
if engineValue, exists := c.Get("deployment_engine"); exists && engineValue != nil {
if engine, ok := engineValue.(*deployment.DeploymentEngine); ok {
_ = engine.RemoveServiceContainers(c.Request.Context(), serviceID.String())
}
}
// Delete service (cascade will handle related records)
_, err = db.(*database.DB).Exec(
"DELETE FROM services WHERE id = $1",
+1 -5
View File
@@ -3,7 +3,6 @@ package api
import (
"context"
"database/sql"
"io"
"log"
"net/http"
"os"
@@ -241,10 +240,7 @@ func applyCloudflaredTunnel(client *docker.Client, token string) {
return
}
if reader, err := client.PullImage(ctx, cloudflaredImage, registry.AuthConfig{}); err == nil {
_, _ = io.Copy(io.Discard, reader)
_ = reader.Close()
} else {
if err := client.PullImageWait(ctx, cloudflaredImage, registry.AuthConfig{}); err != nil {
log.Printf("Failed to pull %s: %v", cloudflaredImage, err)
}
+5 -5
View File
@@ -93,10 +93,10 @@ func (bm *BuildManager) buildPrebuilt(ctx context.Context, req *types.BuildReque
return nil, fmt.Errorf("prebuilt image not specified")
}
// Pull the prebuilt image
// Pull the prebuilt image — must drain the pull stream or the daemon
// aborts it and the tag below races a missing image.
auth := registry.AuthConfig{}
_, err := bm.dockerClient.PullImage(ctx, req.PrebuiltImage, auth)
if err != nil {
if err := bm.dockerClient.PullImageWait(ctx, req.PrebuiltImage, auth); err != nil {
return nil, fmt.Errorf("failed to pull prebuilt image: %w", err)
}
@@ -104,7 +104,7 @@ func (bm *BuildManager) buildPrebuilt(ctx context.Context, req *types.BuildReque
if req.ImageName != "" && req.ImageTag != "" {
targetImage := fmt.Sprintf("%s:%s", req.ImageName, req.ImageTag)
if targetImage != req.PrebuiltImage {
err = bm.dockerClient.TagImage(ctx, req.PrebuiltImage, targetImage)
err := bm.dockerClient.TagImage(ctx, req.PrebuiltImage, targetImage)
if err != nil {
return nil, fmt.Errorf("failed to tag image: %w", err)
}
@@ -114,7 +114,7 @@ func (bm *BuildManager) buildPrebuilt(ctx context.Context, req *types.BuildReque
// Push to registry if specified
if req.RegistryURL != "" {
err = bm.dockerClient.PushImage(ctx, req.PrebuiltImage, req.RegistryURL)
err := bm.dockerClient.PushImage(ctx, req.PrebuiltImage, req.RegistryURL)
if err != nil {
return nil, fmt.Errorf("failed to push image: %w", err)
}
+41 -76
View File
@@ -9,10 +9,6 @@ import (
"containr/internal/build"
"containr/internal/docker"
"containr/internal/types"
"github.com/docker/docker/api/types/mount"
"github.com/docker/docker/api/types/network"
"github.com/docker/go-connections/nat"
)
type DeploymentEngine struct {
@@ -56,6 +52,13 @@ type ServiceConfig struct {
Resources ResourceLimits `json:"resources,omitempty"`
HealthCheck *HealthCheck `json:"health_check,omitempty"`
Replicas int `json:"replicas"`
// PublicPort is the container port exposed to users (published host port,
// or Traefik routing when Domain is set). 0 = internal-only service.
PublicPort int32 `json:"public_port,omitempty"`
// Domain is a public hostname routed to PublicPort via Traefik.
Domain string `json:"domain,omitempty"`
// HealthPath is an HTTP path probed on PublicPort for container health.
HealthPath string `json:"health_path,omitempty"`
}
type PortMapping struct {
@@ -297,91 +300,53 @@ func (de *DeploymentEngine) buildImage(ctx context.Context, deployment *Deployme
return response.ImageName, nil
}
// deployService deploys the service using the built image
// deployService deploys the service using the built image. Containers are
// reconciled onto the project network with a DNS alias per service name —
// other services reach it at http://<name>:<port>.
func (de *DeploymentEngine) deployService(ctx context.Context, deployment *Deployment) error {
// Convert service config to Docker container config
containerConfig := &docker.ContainerConfig{
Name: fmt.Sprintf("containr-%s-%s", deployment.ServiceID, deployment.ID),
Image: deployment.ImageName,
Cmd: deployment.Config.Command,
Labels: deployment.Config.Labels,
Networks: make(map[string]*network.EndpointSettings),
cfg := deployment.Config
spec := RuntimeSpec{
ProjectID: deployment.ProjectID,
ServiceID: deployment.ServiceID,
Name: cfg.Name,
Image: deployment.ImageName,
Command: cfg.Command,
Env: cfg.Environment,
Replicas: cfg.Replicas,
Port: cfg.PublicPort,
Domain: cfg.Domain,
HealthPath: cfg.HealthPath,
RestartPolicy: cfg.RestartPolicy,
MemoryBytes: cfg.Resources.MemoryBytes,
NanoCPUs: cfg.Resources.CPUQuota,
}
// Set environment variables
for k, v := range deployment.Config.Environment {
containerConfig.Env = append(containerConfig.Env, fmt.Sprintf("%s=%s", k, v))
}
// Set restart policy
containerConfig.RestartPolicy = deployment.Config.RestartPolicy
// Configure port mappings
portBindings := make(nat.PortMap)
for _, pm := range deployment.Config.PortMappings {
port := nat.Port(fmt.Sprintf("%d/%s", pm.ContainerPort, pm.Protocol))
if pm.HostPort > 0 {
portBindings[port] = []nat.PortBinding{
{
HostIP: pm.HostIP,
HostPort: fmt.Sprintf("%d", pm.HostPort),
},
}
for _, pm := range cfg.PortMappings {
if spec.Port == 0 {
spec.Port = pm.ContainerPort
}
}
containerConfig.PortBindings = portBindings
// Configure resource limits
if deployment.Config.Resources.MemoryBytes > 0 {
containerConfig.Memory = deployment.Config.Resources.MemoryBytes
}
if deployment.Config.Resources.CPUQuota > 0 {
containerConfig.NanoCPUs = deployment.Config.Resources.CPUQuota
if spec.RestartPolicy == "" {
spec.RestartPolicy = "unless-stopped"
}
// Configure volume mounts
for _, vm := range deployment.Config.VolumeMounts {
mount := mount.Mount{
Type: mount.Type(vm.Type),
Source: vm.Source,
Target: vm.Destination,
ReadOnly: vm.ReadOnly,
}
containerConfig.Mounts = append(containerConfig.Mounts, mount)
// Volume mounts aren't reconciled replicas-aware yet; still applied on the
// container config of every replica via spec extension later if needed.
state, err := de.ReconcileService(ctx, spec)
if err != nil {
return err
}
// Create containers based on replica count
deployment.Containers = make([]ContainerInfo, deployment.Config.Replicas)
for i := 0; i < deployment.Config.Replicas; i++ {
containerName := fmt.Sprintf("%s-%d", containerConfig.Name, i)
// Create container
containerID, err := de.dockerClient.CreateContainer(ctx, *containerConfig)
if err != nil {
return fmt.Errorf("failed to create container %d: %w", i, err)
}
// Start container
err = de.dockerClient.StartContainer(ctx, containerID)
if err != nil {
return fmt.Errorf("failed to start container %d: %w", i, err)
}
// Get container info
_, err = de.dockerClient.GetContainer(ctx, containerID)
if err != nil {
log.Printf("Failed to get container info for %s: %v", containerID, err)
}
deployment.Containers = make([]ContainerInfo, len(state.Containers))
for i, c := range state.Containers {
deployment.Containers[i] = ContainerInfo{
ID: containerID,
Name: containerName,
Status: "running",
ID: c.ID,
Name: c.Name,
Status: c.State,
CreatedAt: time.Now(),
StartedAt: time.Now(),
}
}
return nil
}
+374
View File
@@ -0,0 +1,374 @@
package deployment
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"strings"
"time"
"containr/internal/docker"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/network"
"github.com/docker/go-connections/nat"
)
// Container labels used to associate runtime containers with services.
const (
LabelManaged = "containr.managed"
LabelProject = "containr.project"
LabelService = "containr.service"
LabelReplica = "containr.replica"
LabelSpecHash = "containr.spec-hash"
LabelDeployment = "containr.deployment"
)
// RuntimeSpec is everything needed to run a service's containers.
type RuntimeSpec struct {
ProjectID string
ServiceID string
Name string // DNS alias on the project network
Image string
Command []string
Env map[string]string // already resolved
Replicas int
Port int32 // container port to expose publicly; 0 = none
Domain string // public hostname routed via Traefik; empty = none
HealthPath string // http path probed on Port for container healthcheck
RestartPolicy string
MemoryBytes int64
NanoCPUs int64
}
// RuntimeContainer describes one live replica.
type RuntimeContainer struct {
ID string `json:"id"`
Name string `json:"name"`
State string `json:"state"` // running, exited, ...
Replica int `json:"replica"`
}
// RuntimeState is the live view of a service's containers.
type RuntimeState struct {
Desired int `json:"desired"`
Containers []RuntimeContainer `json:"containers"`
URLs []string `json:"urls"`
Status string `json:"status"` // running, degraded, stopped
}
func projectNetworkName(projectID string) string {
short := strings.ReplaceAll(projectID, "-", "")
if len(short) > 12 {
short = short[:12]
}
return "containr-proj-" + short
}
func serviceNamePrefix(serviceID string) string {
return fmt.Sprintf("containr-%s-", serviceID)
}
// EnsureProjectNetwork creates the per-project bridge network if missing.
// Containers on it resolve each other by service name (private networking).
func (de *DeploymentEngine) EnsureProjectNetwork(ctx context.Context, projectID string) (string, error) {
name := projectNetworkName(projectID)
networks, err := de.dockerClient.ListNetworks(ctx)
if err != nil {
return "", err
}
for _, n := range networks {
if n.Name == name {
return name, nil
}
}
_, err = de.dockerClient.CreateNetwork(ctx, docker.NetworkConfig{
Name: name,
Driver: "bridge",
Labels: map[string]string{
LabelManaged: "true",
LabelProject: projectID,
},
})
if err != nil {
return "", fmt.Errorf("create project network: %w", err)
}
return name, nil
}
// traefikNetworkName finds the shared edge network (the infra stack's
// containr-network where Traefik listens) if it exists.
func (de *DeploymentEngine) traefikNetworkName(ctx context.Context) string {
networks, err := de.dockerClient.ListNetworks(ctx)
if err != nil {
return ""
}
for _, n := range networks {
if n.Name == "containr-network" {
return n.Name
}
}
return ""
}
// ListServiceContainers returns containers managed for a service.
func (de *DeploymentEngine) ListServiceContainers(ctx context.Context, serviceID string) ([]container.Summary, error) {
return de.dockerClient.ListContainersFiltered(ctx, filters.NewArgs(
filters.Arg("label", LabelService+"="+serviceID),
), true)
}
// ReconcileService brings live containers in line with the spec. When the
// spec hash changed (image/env/ports/etc.) all replicas are recreated;
// otherwise only the replica delta is applied.
func (de *DeploymentEngine) ReconcileService(ctx context.Context, spec RuntimeSpec) (*RuntimeState, error) {
if spec.Replicas < 1 {
spec.Replicas = 1
}
networkName, err := de.EnsureProjectNetwork(ctx, spec.ProjectID)
if err != nil {
return nil, err
}
edgeNetwork := ""
if spec.Domain != "" {
edgeNetwork = de.traefikNetworkName(ctx)
}
hash := specHash(spec)
existing, err := de.ListServiceContainers(ctx, spec.ServiceID)
if err != nil {
return nil, err
}
// Deterministic order: scale-down removes the highest replica indices.
sort.Slice(existing, func(i, j int) bool {
return replicaIndex(existing[i]) < replicaIndex(existing[j])
})
drifted := false
for _, c := range existing {
if c.Labels[LabelSpecHash] != hash {
drifted = true
break
}
}
if drifted {
for _, c := range existing {
_ = de.dockerClient.RemoveContainer(ctx, c.ID, true)
}
existing = nil
}
for i := len(existing); i > spec.Replicas; i-- {
victim := existing[i-1]
_ = de.dockerClient.StopContainer(ctx, victim.ID, nil)
_ = de.dockerClient.RemoveContainer(ctx, victim.ID, true)
existing = existing[:i-1]
}
for i := len(existing); i < spec.Replicas; i++ {
if _, err := de.createReplica(ctx, spec, networkName, edgeNetwork, i, hash); err != nil {
return nil, fmt.Errorf("create replica %d: %w", i, err)
}
}
for _, c := range existing {
if c.State != "running" {
_ = de.dockerClient.StartContainer(ctx, c.ID)
}
}
return de.RuntimeState(ctx, spec.ServiceID, spec.Replicas)
}
func (de *DeploymentEngine) createReplica(ctx context.Context, spec RuntimeSpec, projectNet, edgeNet string, index int, hash string) (string, error) {
env := make([]string, 0, len(spec.Env))
for k, v := range spec.Env {
env = append(env, k+"="+v)
}
sort.Strings(env)
labels := map[string]string{
LabelManaged: "true",
LabelProject: spec.ProjectID,
LabelService: spec.ServiceID,
LabelReplica: fmt.Sprintf("%d", index),
LabelSpecHash: hash,
}
endpoints := map[string]*network.EndpointSettings{
projectNet: {Aliases: []string{spec.Name}},
}
if edgeNet != "" {
endpoints[edgeNet] = &network.EndpointSettings{}
labels["traefik.enable"] = "true"
labels["traefik.docker.network"] = edgeNet
labels["traefik.http.routers.svc-"+spec.ServiceID[:8]+".rule"] = "Host(`" + spec.Domain + "`)"
labels["traefik.http.routers.svc-"+spec.ServiceID[:8]+".entrypoints"] = "web"
labels["traefik.http.services.svc-"+spec.ServiceID[:8]+".loadbalancer.server.port"] = fmt.Sprintf("%d", spec.Port)
}
cfg := docker.ContainerConfig{
Name: fmt.Sprintf("%s%d", serviceNamePrefix(spec.ServiceID), index),
Image: spec.Image,
Cmd: spec.Command,
Env: env,
Labels: labels,
RestartPolicy: spec.RestartPolicy,
Memory: spec.MemoryBytes,
NanoCPUs: spec.NanoCPUs,
Networks: endpoints,
}
if spec.Port > 0 {
port := nat.Port(fmt.Sprintf("%d/tcp", spec.Port))
cfg.ExposedPorts = nat.PortSet{port: struct{}{}}
if spec.Domain == "" {
// No edge router: publish an ephemeral host port for direct access.
cfg.PortBindings = nat.PortMap{port: []nat.PortBinding{{HostIP: "0.0.0.0", HostPort: ""}}}
}
if spec.HealthPath != "" {
cfg.Healthcheck = &container.HealthConfig{
Test: []string{"CMD-SHELL", fmt.Sprintf("wget -q -O /dev/null http://127.0.0.1:%d%s || exit 1", spec.Port, spec.HealthPath)},
Interval: 30 * time.Second,
Timeout: 5 * time.Second,
Retries: 3,
StartPeriod: 10 * time.Second,
}
}
}
id, err := de.dockerClient.CreateContainer(ctx, cfg)
if err != nil {
return "", err
}
if err := de.dockerClient.StartContainer(ctx, id); err != nil {
return "", fmt.Errorf("start: %w", err)
}
return id, nil
}
// RuntimeState reads live container state for a service.
func (de *DeploymentEngine) RuntimeState(ctx context.Context, serviceID string, desired int) (*RuntimeState, error) {
containers, err := de.ListServiceContainers(ctx, serviceID)
if err != nil {
return nil, err
}
state := &RuntimeState{Desired: desired, Status: "stopped"}
running := 0
for _, c := range containers {
rc := RuntimeContainer{ID: c.ID, Name: strings.TrimPrefix(c.Names[0], "/"), State: c.State, Replica: replicaIndex(c)}
state.Containers = append(state.Containers, rc)
if c.State == "running" {
running++
}
for _, p := range c.Ports {
if p.PublicPort != 0 {
state.URLs = append(state.URLs, fmt.Sprintf("http://localhost:%d", p.PublicPort))
}
}
}
sort.Slice(state.Containers, func(i, j int) bool { return state.Containers[i].Replica < state.Containers[j].Replica })
sort.Strings(state.URLs)
switch {
case running == 0:
state.Status = "stopped"
case running < desired:
state.Status = "degraded"
default:
state.Status = "running"
}
return state, nil
}
// StopService stops (but keeps) all replicas of a service.
func (de *DeploymentEngine) StopService(ctx context.Context, serviceID string) error {
containers, err := de.ListServiceContainers(ctx, serviceID)
if err != nil {
return err
}
for _, c := range containers {
if c.State == "running" {
if err := de.dockerClient.StopContainer(ctx, c.ID, nil); err != nil {
return err
}
}
}
return nil
}
// StartService starts all stopped replicas.
func (de *DeploymentEngine) StartService(ctx context.Context, serviceID string) error {
containers, err := de.ListServiceContainers(ctx, serviceID)
if err != nil {
return err
}
for _, c := range containers {
if c.State != "running" {
if err := de.dockerClient.StartContainer(ctx, c.ID); err != nil {
return err
}
}
}
return nil
}
// RemoveServiceContainers deletes every replica of a service.
func (de *DeploymentEngine) RemoveServiceContainers(ctx context.Context, serviceID string) error {
containers, err := de.ListServiceContainers(ctx, serviceID)
if err != nil {
return err
}
for _, c := range containers {
_ = de.dockerClient.RemoveContainer(ctx, c.ID, true)
}
return nil
}
// RemoveProjectContainers deletes every managed container in a project.
func (de *DeploymentEngine) RemoveProjectContainers(ctx context.Context, projectID string) error {
containers, err := de.dockerClient.ListContainersFiltered(ctx, filters.NewArgs(
filters.Arg("label", LabelProject+"="+projectID),
), true)
if err != nil {
return err
}
for _, c := range containers {
_ = de.dockerClient.RemoveContainer(ctx, c.ID, true)
}
return nil
}
// RemoveProjectNetwork removes the per-project bridge network if present.
func (de *DeploymentEngine) RemoveProjectNetwork(ctx context.Context, projectID string) error {
return de.dockerClient.RemoveNetwork(ctx, projectNetworkName(projectID))
}
func replicaIndex(c container.Summary) int {
replica := 0
fmt.Sscanf(c.Labels[LabelReplica], "%d", &replica)
return replica
}
func specHash(spec RuntimeSpec) string {
payload, _ := json.Marshal(struct {
Image, Name, Domain, HealthPath, RestartPolicy string
Command []string
Env map[string]string
Port int32
MemoryBytes, NanoCPUs int64
}{spec.Image, spec.Name, spec.Domain, spec.HealthPath, spec.RestartPolicy,
spec.Command, spec.Env, spec.Port, spec.MemoryBytes, spec.NanoCPUs})
sum := sha256.Sum256(payload)
return hex.EncodeToString(sum[:8])
}
+41 -1
View File
@@ -3,6 +3,7 @@ package docker
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"time"
@@ -10,6 +11,7 @@ import (
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/events"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/image"
"github.com/docker/docker/api/types/network"
"github.com/docker/docker/api/types/registry"
@@ -50,6 +52,14 @@ func (c *Client) ListContainers(ctx context.Context, all bool) ([]types.Containe
})
}
// ListContainersFiltered returns containers matching the given filters.
func (c *Client) ListContainersFiltered(ctx context.Context, f filters.Args, all bool) ([]types.Container, error) {
return c.cli.ContainerList(ctx, container.ListOptions{
All: all,
Filters: f,
})
}
// GetContainer returns detailed information about a specific container
func (c *Client) GetContainer(ctx context.Context, containerID string) (types.ContainerJSON, error) {
return c.cli.ContainerInspect(ctx, containerID)
@@ -63,6 +73,7 @@ func (c *Client) CreateContainer(ctx context.Context, config ContainerConfig) (s
Env: config.Env,
Labels: config.Labels,
ExposedPorts: config.ExposedPorts,
Healthcheck: config.Healthcheck,
}
hostConfig := &container.HostConfig{
@@ -168,7 +179,9 @@ func (c *Client) ListImages(ctx context.Context, all bool) ([]image.Summary, err
})
}
// PullImage pulls an image from a registry
// PullImage pulls an image from a registry. The pull runs as the returned
// stream is consumed — callers must drain and close it (or use
// PullImageWait) or the pull may be aborted.
func (c *Client) PullImage(ctx context.Context, ref string, auth registry.AuthConfig) (io.ReadCloser, error) {
authStr, _ := registry.EncodeAuthConfig(auth)
return c.cli.ImagePull(ctx, ref, image.PullOptions{
@@ -176,6 +189,33 @@ func (c *Client) PullImage(ctx context.Context, ref string, auth registry.AuthCo
})
}
// PullImageWait pulls an image and blocks until the pull stream completes.
func (c *Client) PullImageWait(ctx context.Context, ref string, auth registry.AuthConfig) error {
reader, err := c.PullImage(ctx, ref, auth)
if err != nil {
return err
}
defer reader.Close()
// The daemon performs the pull while streaming progress; drain it fully
// and surface any errorDetail the daemon reports in the stream.
dec := json.NewDecoder(reader)
var msg struct {
ErrorDetail *struct {
Message string `json:"message"`
} `json:"errorDetail"`
Error string `json:"error"`
}
for dec.Decode(&msg) == nil {
if msg.ErrorDetail != nil {
return fmt.Errorf("%s", msg.ErrorDetail.Message)
}
if msg.Error != "" {
return fmt.Errorf("%s", msg.Error)
}
}
return nil
}
// BuildImage builds an image from a Dockerfile
func (c *Client) BuildImage(ctx context.Context, buildContext io.Reader, options BuildOptions) (types.ImageBuildResponse, error) {
return c.cli.ImageBuild(ctx, buildContext, types.ImageBuildOptions{
+2
View File
@@ -3,6 +3,7 @@ package docker
import (
"time"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/mount"
"github.com/docker/docker/api/types/network"
@@ -24,6 +25,7 @@ type ContainerConfig struct {
NanoCPUs int64
NetworkMode string
Networks map[string]*network.EndpointSettings
Healthcheck *container.HealthConfig
}
// LogOptions represents options for retrieving container logs