package api
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"sort"
"strings"
"sync"
"time"
"containr/internal/database"
"containr/internal/database/sqlcdb"
"containr/internal/deployment"
"containr/internal/docker"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/network"
"github.com/docker/docker/api/types/registry"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
)
// Sleep mode: services with sleep_enabled are scaled to zero after
// sleep_idle_minutes without inbound traffic. Idleness is measured as
// NetworkIO byte deltas across the service's containers — we have no
// per-request layer in front of Traefik, so counters are the signal.
//
// When a service sleeps its containers are removed and a tiny busybox
// "wake" container takes over the same Traefik host rule at a higher
// priority: it serves a reloading "waking" page and pings the unauthenticated
// internal wake endpoint, which reconciles the service back to running.
// Wake containers carry a dedicated label (not the service label) so
// RuntimeState and reconcile never count them as replicas; the sweeper
// prunes them once the workload is running again.
const (
wakeImage = "busybox:1.36"
wakeLabel = "containr.wake"
wakeForLabel = "containr.wake-for"
wakeSweepInterval = 30 * time.Second
)
type sleepTracker struct {
mu sync.Mutex
activity map[uuid.UUID]sleepActivity
}
type sleepActivity struct {
bytes uint64
idleSince time.Time
}
var tracker = &sleepTracker{activity: make(map[uuid.UUID]sleepActivity)}
// StartSleepSweeper polls enabled services and scales idle ones to zero.
func StartSleepSweeper(ctx context.Context, db *database.DB, dc *docker.Client, engine *deployment.DeploymentEngine, apiPort int) {
if db == nil || dc == nil || engine == nil {
return
}
go func() {
ticker := time.NewTicker(wakeSweepInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
sweepSleeping(ctx, db, dc, engine, apiPort)
pruneWakeContainers(ctx, db, dc, engine)
}
}
}()
log.Printf("sleep sweeper: running (interval %s)", wakeSweepInterval)
}
type sleepCandidate struct {
ID uuid.UUID
ProjectID uuid.UUID
Name string
Status string
IdleMinutes int
}
func sweepSleeping(ctx context.Context, db *database.DB, dc *docker.Client, engine *deployment.DeploymentEngine, apiPort int) {
sweepRows, err := sqlcdb.New(db.DB).ListSleepCandidates(ctx)
if err != nil {
return
}
var candidates []sleepCandidate
for _, r := range sweepRows {
candidates = append(candidates, sleepCandidate{
ID: r.ID, ProjectID: r.ProjectID, Name: r.Name,
Status: r.Status, IdleMinutes: int(r.SleepIdleMinutes),
})
}
q := sqlcdb.New(db.DB)
for _, svc := range candidates {
// Remote replicas live in inventory, not the local docker host —
// idleness comes from heartbeat-reported net counters instead.
if remote := remoteLiveContainers(ctx, q, svc.ID.String()); len(remote) > 0 {
var total uint64
for _, row := range remote {
total += rowNetBytes(row)
}
if markIdle(svc.ID, total, time.Duration(svc.IdleMinutes)*time.Minute) {
putToSleepRemote(ctx, db, q, svc, apiPort)
}
continue
}
containers, err := engine.ListServiceContainers(ctx, svc.ID.String())
if err != nil || len(containers) == 0 {
continue
}
// Ignore wake containers — they're the placeholder, not the workload.
workload := containers[:0]
for _, c := range containers {
if c.Labels[wakeLabel] != "true" {
workload = append(workload, c)
}
}
if len(workload) == 0 {
continue
}
var total uint64
for _, c := range workload {
total += containerNetBytes(ctx, dc, c.ID)
}
if markIdle(svc.ID, total, time.Duration(svc.IdleMinutes)*time.Minute) {
putToSleep(ctx, db, dc, engine, svc, apiPort)
}
}
}
// remoteLiveContainers returns inventory rows whose replicas are alive on
// node agents — the signal that idleness must be measured remotely.
func remoteLiveContainers(ctx context.Context, q *sqlcdb.Queries, serviceID string) []sqlcdb.ListServiceContainersRow {
rows, err := q.ListServiceContainers(ctx, serviceID)
if err != nil {
return nil
}
live := rows[:0]
for _, row := range rows {
var st struct {
State string `json:"state"`
}
if err := json.Unmarshal(row.Status.RawMessage, &st); err == nil &&
st.State != "" && st.State != "removed" && st.State != "sleeping" {
live = append(live, row)
}
}
return live
}
// rowNetBytes reads the cumulative rx+tx the agent reported at its last
// heartbeat — 0 for rows written before the counter existed.
func rowNetBytes(row sqlcdb.ListServiceContainersRow) uint64 {
var res struct {
NetBytes uint64 `json:"net_bytes"`
}
_ = json.Unmarshal(row.Resources.RawMessage, &res)
return res.NetBytes
}
// putToSleepRemote removes the service's remote replicas via their agents,
// marks the service sleeping, and repoints the domain route at the wake
// endpoint so inbound traffic brings it back.
func putToSleepRemote(ctx context.Context, db *database.DB, q *sqlcdb.Queries, svc sleepCandidate, apiPort int) {
runner := newAgentNodeRunner(db)
for _, agentID := range remoteAgentsForService(ctx, db, svc.ID) {
if err := runner.ControlService(ctx, svc.ID.String(), agentID, "remove"); err != nil {
log.Printf("sleep: remote remove on %s failed: %v", agentID, err)
continue
}
_ = q.DeleteServiceContainersOnAgent(ctx, sqlcdb.DeleteServiceContainersOnAgentParams{
NodeAgentID: agentID,
ServiceID: svc.ID.String(),
})
}
tracker.mu.Lock()
delete(tracker.activity, svc.ID)
tracker.mu.Unlock()
setServiceStatusNow(db, svc.ID, "sleeping")
log.Printf("sleep: %s (%s) remote replicas removed after idle timeout", svc.Name, svc.ID)
domain, _ := q.GetServiceDomainText(ctx, svc.ID)
domains := serviceDomainNames(db, svc.ID, domain)
if len(domains) > 0 {
writeWakeIngress(svc.ID.String(), domains, apiPort)
}
}
// markIdle returns true when the network counter stopped growing past the
// timeout. First sighting only sets the baseline — no idle without history.
func markIdle(id uuid.UUID, bytes uint64, timeout time.Duration) bool {
tracker.mu.Lock()
defer tracker.mu.Unlock()
a := tracker.activity[id]
if bytes != a.bytes || a.idleSince.IsZero() {
tracker.activity[id] = sleepActivity{bytes: bytes, idleSince: time.Now()}
return false
}
return time.Since(a.idleSince) >= timeout
}
func containerNetBytes(ctx context.Context, dc *docker.Client, id string) uint64 {
stats, err := dc.GetContainerStats(ctx, id, false)
if err != nil || stats == nil {
return 0
}
defer stats.Body.Close()
var snap struct {
Networks map[string]struct {
RxBytes uint64 `json:"rx_bytes"`
TxBytes uint64 `json:"tx_bytes"`
} `json:"networks"`
}
if err := json.NewDecoder(stats.Body).Decode(&snap); err != nil {
return 0
}
var total uint64
for _, n := range snap.Networks {
total += n.RxBytes + n.TxBytes
}
return total
}
// putToSleep removes the workload containers, marks the service sleeping,
// and installs the wake placeholder when the service has a public domain.
func putToSleep(ctx context.Context, db *database.DB, dc *docker.Client, engine *deployment.DeploymentEngine, svc sleepCandidate, apiPort int) {
containers, err := engine.ListServiceContainers(ctx, svc.ID.String())
if err != nil {
return
}
for _, c := range containers {
if c.Labels[wakeLabel] == "true" {
continue
}
_ = dc.StopContainer(ctx, c.ID, nil)
_ = dc.RemoveContainer(ctx, c.ID, true)
}
tracker.mu.Lock()
delete(tracker.activity, svc.ID)
tracker.mu.Unlock()
setServiceStatusNow(db, svc.ID, "sleeping")
log.Printf("sleep: %s (%s) scaled to zero after idle timeout", svc.Name, svc.ID)
domain, _ := sqlcdb.New(db.DB).GetServiceDomainText(ctx, svc.ID)
domains := serviceDomainNames(db, svc.ID, domain)
if len(domains) == 0 {
return
}
if err := spawnWakeContainer(ctx, db, dc, engine, svc, domains, apiPort); err != nil {
log.Printf("sleep: wake placeholder for %s failed: %v", svc.Name, err)
}
}
// wakeCGIScript is installed as /cgi-bin/wake in the wake container. It
// fires the wake callback on the first reachable candidate URL, in the
// background, then serves the reloading placeholder page. Deliberately
// single-quote-free — the file is written through printf '%s'.
const wakeCGIScript = `#!/bin/sh
for u in $WAKE_URLS; do wget -q -O /dev/null --timeout=20 "$u" && break; done >/dev/null 2>&1 &
printf "Content-Type: text/html\r\n\r\n"
printf "%s" "$WAKE_PAGE"
`
// spawnWakeContainer runs a busybox httpd holding the service's Traefik
// host rule at higher priority. Inbound hits redirect to /cgi-bin/wake,
// which calls the internal wake endpoint and serves a reloading page
// until the reconcile removes the placeholder.
func spawnWakeContainer(ctx context.Context, db *database.DB, dc *docker.Client, engine *deployment.DeploymentEngine, svc sleepCandidate, domains []string, apiPort int) error {
edgeNet := ""
for _, n := range networkList(ctx, dc) {
if n == "containr-network" {
edgeNet = n
}
}
if edgeNet == "" {
return fmt.Errorf("no edge network")
}
router := "svc-" + svc.ID.String()[:8]
sort.Strings(domains)
rules := make([]string, 0, len(domains))
for _, d := range domains {
rules = append(rules, "Host(`"+d+"`)")
}
labels := map[string]string{
deployment.LabelManaged: "true",
deployment.LabelProject: svc.ProjectID.String(),
wakeForLabel: svc.ID.String(),
wakeLabel: "true",
"traefik.enable": "true",
"traefik.docker.network": edgeNet,
"traefik.http.routers." + router + ".rule": strings.Join(rules, " || "),
"traefik.http.routers." + router + ".entrypoints": "web",
"traefik.http.routers." + router + ".priority": "100",
"traefik.http.services." + router + ".loadbalancer.server.port": "80",
}
wakeURLs := wakeEndpointURLs(ctx, dc, apiPort, svc.ID, edgeNet)
page := `Waking up` +
`` +
`Service is waking up
This Containr service was sleeping and is resuming now. The page reloads automatically.
Service is waking up
This Containr service was sleeping and is resuming now. The page reloads automatically.