diff --git a/agent/internal/agent/reporting.go b/agent/internal/agent/reporting.go index 6dc3fadf..50201cf9 100644 --- a/agent/internal/agent/reporting.go +++ b/agent/internal/agent/reporting.go @@ -65,7 +65,7 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport log.Printf("[metrics] failed to collect container stats: %v", err) return } - if err := a.MetricsSender.SendContainerStats(containerStats, collectedAt); err != nil { + if err := a.MetricsSender.SendContainerStats(containerStats, time.Now()); err != nil { log.Printf("[metrics] failed to send container stats: %v", err) } }() diff --git a/agent/internal/container/stats.go b/agent/internal/container/stats.go index 960a2051..46133268 100644 --- a/agent/internal/container/stats.go +++ b/agent/internal/container/stats.go @@ -1,13 +1,17 @@ package container import ( + "bufio" "bytes" - "encoding/json" + "context" "fmt" + "log" "math" "os/exec" "strconv" "strings" + "sync" + "time" "unicode" ) @@ -16,20 +20,52 @@ type ResourceStats struct { ServiceID string DeploymentID string CPUUsagePercent float64 + CPUUsageValid bool MemoryUsagePercent float64 + MemoryUsageValid bool MemoryUsedBytes float64 + MemoryUsedValid bool NetworkReceiveBytes float64 NetworkTransmitBytes float64 } +type podmanStatsSample struct { + containerID string + cpuNano uint64 + systemNano uint64 + cpuCountersValid bool + memoryUsage string + memoryUsagePercent string + networkIO string +} + +const podmanStatsFormat = "{{.ContainerID}}\t{{.CPUNano}}\t{{.SystemNano}}\t{{.MemUsage}}\t{{.MemPerc}}\t{{.NetIO}}" + +var previousResourceSamples = struct { + sync.Mutex + byContainer map[string]podmanStatsSample +}{byContainer: make(map[string]podmanStatsSample)} + +// resourceStatsCollectionMu ensures overlapping periodic and requested reports +// compare CPU counters from snapshots collected in order. +var resourceStatsCollectionMu sync.Mutex + func CollectResourceStats() ([]ResourceStats, error) { + resourceStatsCollectionMu.Lock() + defer resourceStatsCollectionMu.Unlock() + containers, err := List() if err != nil { return nil, err } running := make([]Container, 0, len(containers)) - args := []string{"stats", "--no-stream", "--format", "json"} + args := []string{ + "stats", + "--no-stream", + "--no-trunc", + "--format", podmanStatsFormat, + } for _, c := range containers { if c.State != "running" || c.ServiceID == "" || c.DeploymentID == "" { continue @@ -41,80 +77,42 @@ func CollectResourceStats() ([]ResourceStats, error) { return nil, nil } - cmd := exec.Command("podman", args...) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, "podman", args...) var stderr bytes.Buffer cmd.Stderr = &stderr output, err := cmd.Output() if err != nil { return nil, fmt.Errorf("failed to collect container stats: %s: %w", stderr.String(), err) } - - return parsePodmanStatsOutput(output, running) -} - -func parsePodmanStatsOutput(output []byte, containers []Container) ([]ResourceStats, error) { - rows, err := parseStatsRows(output) + samples, err := parsePodmanStatsSamples(output) if err != nil { return nil, err } - stats := make([]ResourceStats, 0, len(rows)) - for _, row := range rows { - containerID := firstRowString(row, "ID", "Id", "id", "ContainerID", "Container") - container := findStatsContainerByID(containerID, containers) - if container == nil { - name := firstRowString(row, "Name", "Names", "name") - container = findStatsContainerByName(name, containers) + previousResourceSamples.Lock() + defer previousResourceSamples.Unlock() + nextSamples := make(map[string]podmanStatsSample, len(samples)) + for _, container := range running { + if previous, ok := previousResourceSamples.byContainer[container.ID]; ok { + nextSamples[container.ID] = previous } + } + stats := make([]ResourceStats, 0, len(samples)) + for _, sample := range samples { + container := findStatsContainerByID(sample.containerID, running) if container == nil { continue } - - rx, tx := parseNetIO(firstRowString(row, "NetIO", "NetIOBytes", "net_io")) - stats = append(stats, ResourceStats{ - ContainerID: container.ID, - ServiceID: container.ServiceID, - DeploymentID: container.DeploymentID, - CPUUsagePercent: parsePercent(firstRowString(row, "CPUPerc", "CPU", "cpu_percent")), - MemoryUsagePercent: parsePercent(firstRowString(row, "MemPerc", "MEMPerc", "mem_percent")), - MemoryUsedBytes: parseMemUsed(firstRowString(row, "MemUsage", "MemUse", "mem_usage")), - NetworkReceiveBytes: rx, - NetworkTransmitBytes: tx, - }) + previous := previousResourceSamples.byContainer[container.ID] + stats = append(stats, resourceStatsFromSamples(*container, previous, sample)) + nextSamples[container.ID] = sample } - + previousResourceSamples.byContainer = nextSamples return stats, nil } -func parseStatsRows(output []byte) ([]map[string]interface{}, error) { - trimmed := strings.TrimSpace(string(output)) - if trimmed == "" { - return nil, nil - } - - if strings.HasPrefix(trimmed, "[") { - var rows []map[string]interface{} - if err := json.Unmarshal([]byte(trimmed), &rows); err != nil { - return nil, fmt.Errorf("failed to parse podman stats JSON array: %w", err) - } - return rows, nil - } - - var rows []map[string]interface{} - for _, line := range strings.Split(trimmed, "\n") { - line = strings.TrimSpace(line) - if line == "" { - continue - } - var row map[string]interface{} - if err := json.Unmarshal([]byte(line), &row); err != nil { - return nil, fmt.Errorf("failed to parse podman stats JSON row: %w", err) - } - rows = append(rows, row) - } - return rows, nil -} - func findStatsContainerByID(value string, containers []Container) *Container { value = strings.TrimSpace(value) if value == "" { @@ -133,59 +131,98 @@ func findStatsContainerByID(value string, containers []Container) *Container { return nil } -func findStatsContainerByName(value string, containers []Container) *Container { - value = strings.TrimPrefix(strings.TrimSpace(value), "/") - if value == "" { - return nil +func parsePodmanStatsSamples(output []byte) ([]podmanStatsSample, error) { + samples := make([]podmanStatsSample, 0) + skipped := 0 + scanner := bufio.NewScanner(bytes.NewReader(output)) + for scanner.Scan() { + if strings.TrimSpace(scanner.Text()) == "" { + continue + } + sample, err := parsePodmanStatsSample(scanner.Text()) + if err != nil { + skipped++ + continue + } + samples = append(samples, sample) + } + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("failed to read container stats: %w", err) } + if skipped > 0 { + log.Printf("[metrics] skipped %d malformed container stats rows", skipped) + } + return samples, nil +} - for i := range containers { - containerName := strings.TrimPrefix(strings.TrimSpace(containers[i].Name), "/") - if value == containerName { - return &containers[i] - } +func parsePodmanStatsSample(line string) (podmanStatsSample, error) { + parts := strings.Split(line, "\t") + if len(parts) != 6 { + return podmanStatsSample{}, fmt.Errorf("failed to parse podman stats row: expected 6 fields, got %d", len(parts)) } - return nil + cpuNano, cpuErr := strconv.ParseUint(strings.TrimSpace(parts[1]), 10, 64) + systemNano, systemErr := strconv.ParseUint(strings.TrimSpace(parts[2]), 10, 64) + return podmanStatsSample{ + containerID: strings.TrimSpace(parts[0]), + cpuNano: cpuNano, + systemNano: systemNano, + cpuCountersValid: cpuErr == nil && systemErr == nil, + memoryUsage: parts[3], + memoryUsagePercent: parts[4], + networkIO: parts[5], + }, nil } -func firstRowString(row map[string]interface{}, keys ...string) string { - for _, key := range keys { - value, ok := row[key] - if !ok || value == nil { - continue - } - switch v := value.(type) { - case string: - return v - case []interface{}: - if len(v) > 0 { - return fmt.Sprint(v[0]) - } - default: - return fmt.Sprint(v) - } +func resourceStatsFromSamples(container Container, previous, current podmanStatsSample) ResourceStats { + cpuUsagePercent := 0.0 + // Podman SystemNano is a wall-clock timestamp, so CPU nanoseconds divided + // by its delta yields used cores; the metrics sender converts percent to cores. + cpuUsageValid := + previous.cpuCountersValid && + current.cpuCountersValid && + current.cpuNano >= previous.cpuNano && + current.systemNano > previous.systemNano + if cpuUsageValid { + cpuUsagePercent = 100 * float64(current.cpuNano-previous.cpuNano) / + float64(current.systemNano-previous.systemNano) + cpuUsageValid = isFinite(cpuUsagePercent) + } + memoryUsagePercent, memoryUsageValid := parsePercent(current.memoryUsagePercent) + memoryUsedBytes, memoryUsedValid := parseMemUsed(current.memoryUsage) + rx, tx := parseNetIO(current.networkIO) + return ResourceStats{ + ContainerID: container.ID, + ServiceID: container.ServiceID, + DeploymentID: container.DeploymentID, + CPUUsagePercent: cpuUsagePercent, + CPUUsageValid: cpuUsageValid, + MemoryUsagePercent: memoryUsagePercent, + MemoryUsageValid: memoryUsageValid, + MemoryUsedBytes: memoryUsedBytes, + MemoryUsedValid: memoryUsedValid, + NetworkReceiveBytes: rx, + NetworkTransmitBytes: tx, } - return "" } -func parsePercent(value string) float64 { +func parsePercent(value string) (float64, bool) { value = strings.TrimSpace(strings.TrimSuffix(value, "%")) if value == "" || value == "--" { - return 0 + return 0, false } parsed, err := strconv.ParseFloat(value, 64) if err != nil || !isFinite(parsed) { - return 0 + return 0, false } - return parsed + return parsed, true } -func parseMemUsed(value string) float64 { +func parseMemUsed(value string) (float64, bool) { parts := strings.Split(value, "/") if len(parts) == 0 { - return 0 + return 0, false } - return parseByteQuantity(parts[0]) + return parseByteQuantityValue(parts[0]) } func parseNetIO(value string) (float64, float64) { @@ -193,13 +230,20 @@ func parseNetIO(value string) (float64, float64) { if len(parts) != 2 { return 0, 0 } - return parseByteQuantity(parts[0]), parseByteQuantity(parts[1]) + rx, _ := parseByteQuantityValue(parts[0]) + tx, _ := parseByteQuantityValue(parts[1]) + return rx, tx } func parseByteQuantity(value string) float64 { + parsed, _ := parseByteQuantityValue(value) + return parsed +} + +func parseByteQuantityValue(value string) (float64, bool) { value = strings.TrimSpace(value) if value == "" || value == "--" { - return 0 + return 0, false } compact := strings.ReplaceAll(value, " ", "") @@ -215,22 +259,22 @@ func parseByteQuantity(value string) float64 { unit := strings.ToLower(compact[splitAt:]) parsed, err := strconv.ParseFloat(numberText, 64) if err != nil || !isFinite(parsed) { - return 0 + return 0, false } switch unit { case "", "b": - return parsed + return parsed, true case "kb", "k", "kib", "ki": - return parsed * unitMultiplier(unit, 1) + return parsed * unitMultiplier(unit, 1), true case "mb", "m", "mib", "mi": - return parsed * unitMultiplier(unit, 2) + return parsed * unitMultiplier(unit, 2), true case "gb", "g", "gib", "gi": - return parsed * unitMultiplier(unit, 3) + return parsed * unitMultiplier(unit, 3), true case "tb", "t", "tib", "ti": - return parsed * unitMultiplier(unit, 4) + return parsed * unitMultiplier(unit, 4), true default: - return parsed + return 0, false } } diff --git a/agent/internal/container/stats_test.go b/agent/internal/container/stats_test.go index 0f9844f5..d9a6fe00 100644 --- a/agent/internal/container/stats_test.go +++ b/agent/internal/container/stats_test.go @@ -1,90 +1,94 @@ package container import ( - "reflect" + "math" + "os" + "path/filepath" + "strconv" + "strings" "testing" ) -func TestParsePodmanStatsOutputArray(t *testing.T) { - containers := []Container{ - { - ID: "abcdef1234567890", - ServiceID: "svc_1", - DeploymentID: "dep_1", - }, +func TestParsePodmanStatsSample(t *testing.T) { + sample, err := parsePodmanStatsSample("abcdef123456\t1000000000\t2000000000\t64MiB / 512MiB\t12.50%\t1.5MB / 2.5MB") + if err != nil { + t.Fatalf("parse sample: %v", err) } + if sample.containerID != "abcdef123456" || sample.cpuNano != 1_000_000_000 || sample.systemNano != 2_000_000_000 || !sample.cpuCountersValid { + t.Fatalf("unexpected sample: %#v", sample) + } +} - stats, err := parsePodmanStatsOutput([]byte(`[ - { - "ID": "abcdef123456", - "Name": "api", - "CPUPerc": "12.34%", - "MemUsage": "64MiB / 512MiB", - "MemPerc": "12.50%", - "NetIO": "1.5MB / 2.5MB" - } - ]`), containers) +func TestParsePodmanStatsSamplesSkipsMalformedRows(t *testing.T) { + containerID := strings.Repeat("e", 64) + samples, err := parsePodmanStatsSamples([]byte("malformed\n" + statsLine(containerID, 100, 1000))) if err != nil { - t.Fatalf("parse stats: %v", err) - } - want := []ResourceStats{{ - ContainerID: "abcdef1234567890", - ServiceID: "svc_1", - DeploymentID: "dep_1", - CPUUsagePercent: 12.34, - MemoryUsagePercent: 12.5, - MemoryUsedBytes: 64 * 1024 * 1024, - NetworkReceiveBytes: 1.5 * 1000 * 1000, - NetworkTransmitBytes: 2.5 * 1000 * 1000, - }} - if !reflect.DeepEqual(stats, want) { - t.Fatalf("stats = %#v, want %#v", stats, want) + t.Fatalf("parse samples: %v", err) + } + if len(samples) != 1 || samples[0].containerID != containerID { + t.Fatalf("expected valid row to survive malformed peer, got %+v", samples) } } -func TestParsePodmanStatsOutputJSONLines(t *testing.T) { - containers := []Container{ - {ID: "1234567890abcdef", ServiceID: "svc_2", DeploymentID: "dep_2"}, +func TestResourceStatsFromSamplesUsesRecentCPUInterval(t *testing.T) { + container := Container{ID: "container-1", ServiceID: "service-1", DeploymentID: "deployment-1"} + previous := podmanStatsSample{cpuNano: 1_000_000_000, systemNano: 10_000_000_000, cpuCountersValid: true} + current := podmanStatsSample{ + cpuNano: 1_500_000_000, + systemNano: 11_000_000_000, + cpuCountersValid: true, + memoryUsage: "64MiB / 512MiB", + memoryUsagePercent: "12.50%", + networkIO: "1.5MB / 2.5MB", } - stats, err := parsePodmanStatsOutput([]byte(`{"ContainerID":"1234567890","CPUPerc":"0%","MemUsage":"128MB / 1GB","MemPerc":"10%","NetIO":"0B / 32kB"}`), containers) - if err != nil { - t.Fatalf("parse stats: %v", err) + stats := resourceStatsFromSamples(container, previous, current) + if !stats.CPUUsageValid || math.Abs(stats.CPUUsagePercent-50) > 0.001 { + t.Fatalf("CPU stats = %f, valid=%v", stats.CPUUsagePercent, stats.CPUUsageValid) } - if len(stats) != 1 { - t.Fatalf("expected 1 stat, got %d", len(stats)) + if !stats.MemoryUsedValid || stats.MemoryUsedBytes != 64*1024*1024 { + t.Fatalf("memory bytes = %f, valid=%v", stats.MemoryUsedBytes, stats.MemoryUsedValid) } - if stats[0].MemoryUsedBytes != 128*1000*1000 { - t.Fatalf("memory bytes = %f", stats[0].MemoryUsedBytes) + if !stats.MemoryUsageValid || stats.MemoryUsagePercent != 12.5 { + t.Fatalf("memory percent = %f, valid=%v", stats.MemoryUsagePercent, stats.MemoryUsageValid) } - if stats[0].NetworkTransmitBytes != 32*1000 { - t.Fatalf("tx bytes = %f", stats[0].NetworkTransmitBytes) + if stats.NetworkReceiveBytes != 1.5*1000*1000 || stats.NetworkTransmitBytes != 2.5*1000*1000 { + t.Fatalf("network stats = %f/%f", stats.NetworkReceiveBytes, stats.NetworkTransmitBytes) } } -func TestParsePodmanStatsOutputNameFallbackDoesNotUseIDPrefix(t *testing.T) { - containers := []Container{ - {ID: "api1234567890", Name: "backend", ServiceID: "wrong", DeploymentID: "wrong_dep"}, - {ID: "fedcba987654", Name: "api", ServiceID: "svc_3", DeploymentID: "dep_3"}, +func TestResourceStatsFromSamplesKeepsInvalidValuesMissing(t *testing.T) { + container := Container{ID: "container-1", ServiceID: "service-1", DeploymentID: "deployment-1"} + stats := resourceStatsFromSamples(container, + podmanStatsSample{cpuNano: 2, systemNano: 2, cpuCountersValid: true}, + podmanStatsSample{ + cpuNano: 1, + systemNano: 3, + cpuCountersValid: true, + memoryUsage: "-- / 512MiB", + memoryUsagePercent: "NaN%", + networkIO: "-- / --", + }, + ) + if stats.CPUUsageValid || stats.MemoryUsageValid || stats.MemoryUsedValid { + t.Fatalf("invalid observations marked valid: %#v", stats) } +} - stats, err := parsePodmanStatsOutput([]byte(`[ - { - "Name": "api", - "CPUPerc": "7%", - "MemUsage": "1MiB / 128MiB", - "MemPerc": "1%", - "NetIO": "0B / 0B" - } - ]`), containers) - if err != nil { - t.Fatalf("parse stats: %v", err) - } - if len(stats) != 1 { - t.Fatalf("expected 1 stat, got %d", len(stats)) - } - if stats[0].ServiceID != "svc_3" || stats[0].DeploymentID != "dep_3" { - t.Fatalf("unexpected attribution: %#v", stats[0]) +func TestResourceStatsFromSamplesKeepsGenuineZeroValid(t *testing.T) { + container := Container{ID: "container-1", ServiceID: "service-1", DeploymentID: "deployment-1"} + stats := resourceStatsFromSamples(container, + podmanStatsSample{cpuNano: 1, systemNano: 1, cpuCountersValid: true}, + podmanStatsSample{ + cpuNano: 1, + systemNano: 2, + cpuCountersValid: true, + memoryUsage: "0B / 512MiB", + memoryUsagePercent: "0%", + }, + ) + if !stats.CPUUsageValid || !stats.MemoryUsageValid || !stats.MemoryUsedValid { + t.Fatalf("zero observations marked invalid: %#v", stats) } } @@ -105,3 +109,181 @@ func TestParseByteQuantity(t *testing.T) { } } } + +func TestCollectResourceStatsUsesCounterDeltas(t *testing.T) { + statsOutput := installFakeStatsPodman(t, []string{strings.Repeat("a", 64)}) + resetPreviousResourceSamples(t) + containerID := strings.Repeat("a", 64) + + writeStatsOutput(t, statsOutput, statsLine(containerID, 1_000_000_000, 10_000_000_000)) + first, err := CollectResourceStats() + if err != nil { + t.Fatalf("first collection failed: %v", err) + } + if len(first) != 1 { + t.Fatalf("expected one stat, got %d", len(first)) + } + if first[0].CPUUsageValid { + t.Fatal("expected first CPU sample to be invalid without a baseline") + } + if !first[0].MemoryUsageValid || !first[0].MemoryUsedValid { + t.Fatal("expected memory values to remain valid on first sample") + } + + writeStatsOutput(t, statsOutput, statsLine(containerID, 2_000_000_000, 12_000_000_000)) + second, err := CollectResourceStats() + if err != nil { + t.Fatalf("second collection failed: %v", err) + } + if !second[0].CPUUsageValid || second[0].CPUUsagePercent != 50 { + t.Fatalf("expected 50%% CPU from counter delta, got %+v", second[0]) + } +} + +func TestCollectResourceStatsRejectsInvalidCounterDeltas(t *testing.T) { + statsOutput := installFakeStatsPodman(t, []string{strings.Repeat("b", 64)}) + containerID := strings.Repeat("b", 64) + tests := []struct { + name string + firstCPU uint64 + firstTime uint64 + secondCPU uint64 + secondTime uint64 + }{ + {name: "CPU counter reset", firstCPU: 200, firstTime: 2000, secondCPU: 100, secondTime: 3000}, + {name: "system counter unchanged", firstCPU: 100, firstTime: 2000, secondCPU: 200, secondTime: 2000}, + {name: "system counter reset", firstCPU: 100, firstTime: 2000, secondCPU: 200, secondTime: 1000}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resetPreviousResourceSamples(t) + writeStatsOutput(t, statsOutput, statsLine(containerID, tt.firstCPU, tt.firstTime)) + if _, err := CollectResourceStats(); err != nil { + t.Fatalf("first collection failed: %v", err) + } + writeStatsOutput(t, statsOutput, statsLine(containerID, tt.secondCPU, tt.secondTime)) + stats, err := CollectResourceStats() + if err != nil { + t.Fatalf("second collection failed: %v", err) + } + if stats[0].CPUUsageValid { + t.Fatal("expected invalid CPU sample") + } + }) + } +} + +func TestCollectResourceStatsPreservesBaselinesAndPrunesStoppedContainers(t *testing.T) { + firstID := strings.Repeat("c", 64) + missingID := strings.Repeat("d", 64) + statsOutput := installFakeStatsPodman(t, []string{firstID, missingID}) + resetPreviousResourceSamples(t) + + writeStatsOutput(t, statsOutput, statsLine(firstID, 100, 1000)+statsLine(missingID, 100, 1000)) + if _, err := CollectResourceStats(); err != nil { + t.Fatalf("first collection failed: %v", err) + } + + failPath := statsOutput + ".fail" + if err := os.WriteFile(failPath, nil, 0o600); err != nil { + t.Fatalf("create failure marker: %v", err) + } + if _, err := CollectResourceStats(); err == nil { + t.Fatal("expected Podman failure") + } + if err := os.Remove(failPath); err != nil { + t.Fatalf("remove failure marker: %v", err) + } + + writeStatsOutput(t, statsOutput, statsLine(firstID, 300, 2000)) + stats, err := CollectResourceStats() + if err != nil { + t.Fatalf("collection after failure failed: %v", err) + } + if !stats[0].CPUUsageValid || stats[0].CPUUsagePercent != 20 { + t.Fatalf("expected preserved baseline to produce 20%% CPU, got %+v", stats[0]) + } + + previousResourceSamples.Lock() + _, retained := previousResourceSamples.byContainer[firstID] + _, missingRetained := previousResourceSamples.byContainer[missingID] + previousResourceSamples.Unlock() + if !retained || !missingRetained { + t.Fatalf("expected baselines for running containers to survive partial output: retained=%v missingRetained=%v", retained, missingRetained) + } + + writeContainersOutput(t, filepath.Join(filepath.Dir(statsOutput), "containers-output"), []string{firstID}) + writeStatsOutput(t, statsOutput, statsLine(firstID, 400, 3000)) + if _, err := CollectResourceStats(); err != nil { + t.Fatalf("collection after container removal failed: %v", err) + } + previousResourceSamples.Lock() + _, stoppedRetained := previousResourceSamples.byContainer[missingID] + previousResourceSamples.Unlock() + if stoppedRetained { + t.Fatal("expected stopped container baseline to be pruned") + } +} + +func installFakeStatsPodman(t *testing.T, containerIDs []string) string { + t.Helper() + dir := t.TempDir() + statsOutput := filepath.Join(dir, "stats-output") + containersOutput := filepath.Join(dir, "containers-output") + script := `#!/bin/sh +if [ "$1" = "ps" ]; then + cat "$PODMAN_CONTAINERS_OUTPUT" +elif [ -f "$PODMAN_STATS_OUTPUT.fail" ]; then + exit 1 +else + cat "$PODMAN_STATS_OUTPUT" +fi +` + if err := os.WriteFile(filepath.Join(dir, "podman"), []byte(script), 0o700); err != nil { + t.Fatalf("write fake podman: %v", err) + } + writeContainersOutput(t, containersOutput, containerIDs) + t.Setenv("PODMAN_STATS_OUTPUT", statsOutput) + t.Setenv("PODMAN_CONTAINERS_OUTPUT", containersOutput) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + return statsOutput +} + +func writeContainersOutput(t *testing.T, path string, containerIDs []string) { + t.Helper() + containersJSON := "[" + for i, containerID := range containerIDs { + if i > 0 { + containersJSON += "," + } + containersJSON += "{\"Id\":\"" + containerID + "\",\"Names\":[\"test\"],\"State\":\"running\",\"Labels\":{\"techulus.service.id\":\"service-1\",\"techulus.deployment.id\":\"deployment-1\"}}" + } + containersJSON += "]" + if err := os.WriteFile(path, []byte(containersJSON), 0o600); err != nil { + t.Fatalf("write fake containers: %v", err) + } +} + +func resetPreviousResourceSamples(t *testing.T) { + t.Helper() + previousResourceSamples.Lock() + previousResourceSamples.byContainer = make(map[string]podmanStatsSample) + previousResourceSamples.Unlock() + t.Cleanup(func() { + previousResourceSamples.Lock() + previousResourceSamples.byContainer = make(map[string]podmanStatsSample) + previousResourceSamples.Unlock() + }) +} + +func writeStatsOutput(t *testing.T, path, output string) { + t.Helper() + if err := os.WriteFile(path, []byte(output), 0o600); err != nil { + t.Fatalf("write stats output: %v", err) + } +} + +func statsLine(containerID string, cpuNano, systemNano uint64) string { + return containerID + "\t" + strconv.FormatUint(cpuNano, 10) + "\t" + strconv.FormatUint(systemNano, 10) + "\t100 MB / 1 GB\t10%\t1 MB / 2 MB\n" +} diff --git a/agent/internal/metrics/victoria.go b/agent/internal/metrics/victoria.go index 5145dbb2..0151b77b 100644 --- a/agent/internal/metrics/victoria.go +++ b/agent/internal/metrics/victoria.go @@ -24,8 +24,11 @@ type VictoriaMetricsSender struct { type serviceResourceStats struct { ServiceID string CPUUsagePercent float64 + CPUUsageValid bool MemoryUsagePercent float64 + MemoryUsageValid bool MemoryUsedBytes float64 + MemoryUsedValid bool NetworkReceiveBytes float64 NetworkTransmitBytes float64 } @@ -100,14 +103,33 @@ func (v *VictoriaMetricsSender) SendContainerStats(stats []container.ResourceSta serverID := escapeLabelValue(v.serverID) var buf bytes.Buffer + for _, stat := range stats { + labels := map[string]string{ + "deployment_id": escapeLabelValue(stat.DeploymentID), + "server_id": serverID, + "service_id": escapeLabelValue(stat.ServiceID), + } + if stat.CPUUsageValid { + writeGaugeWithLabels(&buf, "techulus_deployment_cpu_usage_cores", labels, stat.CPUUsagePercent/100, timestampMs) + } + if stat.MemoryUsedValid { + writeGaugeWithLabels(&buf, "techulus_deployment_memory_used_bytes", labels, stat.MemoryUsedBytes, timestampMs) + } + } for _, stat := range aggregates { labels := map[string]string{ "server_id": serverID, "service_id": escapeLabelValue(stat.ServiceID), } - writeGaugeWithLabels(&buf, "techulus_service_cpu_usage_percent", labels, stat.CPUUsagePercent, timestampMs) - writeGaugeWithLabels(&buf, "techulus_service_memory_usage_percent", labels, stat.MemoryUsagePercent, timestampMs) - writeGaugeWithLabels(&buf, "techulus_service_memory_used_bytes", labels, stat.MemoryUsedBytes, timestampMs) + if stat.CPUUsageValid { + writeGaugeWithLabels(&buf, "techulus_service_cpu_usage_percent", labels, stat.CPUUsagePercent, timestampMs) + } + if stat.MemoryUsageValid { + writeGaugeWithLabels(&buf, "techulus_service_memory_usage_percent", labels, stat.MemoryUsagePercent, timestampMs) + } + if stat.MemoryUsedValid { + writeGaugeWithLabels(&buf, "techulus_service_memory_used_bytes", labels, stat.MemoryUsedBytes, timestampMs) + } writeGaugeWithLabels(&buf, "techulus_service_network_receive_bytes_total", labels, stat.NetworkReceiveBytes, timestampMs) writeGaugeWithLabels(&buf, "techulus_service_network_transmit_bytes_total", labels, stat.NetworkTransmitBytes, timestampMs) } @@ -123,12 +145,22 @@ func aggregateContainerStats(stats []container.ResourceStats) []serviceResourceS } aggregate := byService[stat.ServiceID] if aggregate == nil { - aggregate = &serviceResourceStats{ServiceID: stat.ServiceID} + // Resource values are sums, so partial coverage must omit the aggregate + // rather than report a lower, misleading service value. + aggregate = &serviceResourceStats{ + ServiceID: stat.ServiceID, + CPUUsageValid: true, + MemoryUsageValid: true, + MemoryUsedValid: true, + } byService[stat.ServiceID] = aggregate } aggregate.CPUUsagePercent += stat.CPUUsagePercent + aggregate.CPUUsageValid = aggregate.CPUUsageValid && stat.CPUUsageValid aggregate.MemoryUsagePercent += stat.MemoryUsagePercent + aggregate.MemoryUsageValid = aggregate.MemoryUsageValid && stat.MemoryUsageValid aggregate.MemoryUsedBytes += stat.MemoryUsedBytes + aggregate.MemoryUsedValid = aggregate.MemoryUsedValid && stat.MemoryUsedValid aggregate.NetworkReceiveBytes += stat.NetworkReceiveBytes aggregate.NetworkTransmitBytes += stat.NetworkTransmitBytes } diff --git a/agent/internal/metrics/victoria_test.go b/agent/internal/metrics/victoria_test.go index 5647e368..07668ddc 100644 --- a/agent/internal/metrics/victoria_test.go +++ b/agent/internal/metrics/victoria_test.go @@ -139,24 +139,38 @@ func TestSendContainerStatsAggregatesStableServiceLabels(t *testing.T) { ServiceID: "svc-a", DeploymentID: "dep-a", CPUUsagePercent: 10, + CPUUsageValid: true, MemoryUsagePercent: 1.5, + MemoryUsageValid: true, MemoryUsedBytes: 1024, + MemoryUsedValid: true, }, { ContainerID: "container-b", ServiceID: "svc-a", DeploymentID: "dep-b", CPUUsagePercent: 20, + CPUUsageValid: true, MemoryUsagePercent: 2.5, + MemoryUsageValid: true, MemoryUsedBytes: 2048, + MemoryUsedValid: true, }, }, time.UnixMilli(1_700_000_000_000)) if err != nil { t.Fatalf("send container stats: %v", err) } - if strings.Contains(gotBody, "container_id") || strings.Contains(gotBody, "deployment_id") { - t.Fatalf("unexpected churn labels in body:\n%s", gotBody) + for _, line := range strings.Split(strings.TrimSpace(gotBody), "\n") { + if strings.HasPrefix(line, "techulus_service_") && strings.Contains(line, "deployment_id") { + t.Fatalf("unexpected deployment label in aggregate metric: %s", line) + } + } + if !strings.Contains(gotBody, `techulus_deployment_cpu_usage_cores{deployment_id="dep-a",server_id="server-1",service_id="svc-a"} 0.100000 1700000000000`) { + t.Fatalf("missing deployment CPU metric:\n%s", gotBody) + } + if !strings.Contains(gotBody, `techulus_deployment_memory_used_bytes{deployment_id="dep-b",server_id="server-1",service_id="svc-a"} 2048.000000 1700000000000`) { + t.Fatalf("missing deployment memory metric:\n%s", gotBody) } if !strings.Contains(gotBody, `techulus_service_cpu_usage_percent{server_id="server-1",service_id="svc-a"} 30.000000 1700000000000`) { t.Fatalf("missing aggregated CPU metric:\n%s", gotBody) @@ -168,3 +182,37 @@ func TestSendContainerStatsAggregatesStableServiceLabels(t *testing.T) { t.Fatalf("missing aggregated memory bytes metric:\n%s", gotBody) } } + +func TestSendContainerStatsOmitsInvalidDeploymentMetrics(t *testing.T) { + var gotBody string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + gotBody = string(body) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + sender := NewVictoriaMetricsSender(server.URL, "server-1") + err := sender.SendContainerStats([]container.ResourceStats{{ + ServiceID: "svc-a", + DeploymentID: "dep-a", + CPUUsagePercent: 0, + CPUUsageValid: true, + MemoryUsedBytes: 0, + MemoryUsedValid: false, + MemoryUsagePercent: 0, + }}, time.UnixMilli(1_700_000_000_000)) + if err != nil { + t.Fatalf("send container stats: %v", err) + } + + if !strings.Contains(gotBody, `techulus_deployment_cpu_usage_cores{deployment_id="dep-a",server_id="server-1",service_id="svc-a"} 0.000000 1700000000000`) { + t.Fatalf("missing genuine zero CPU metric:\n%s", gotBody) + } + if strings.Contains(gotBody, "techulus_deployment_memory_used_bytes") { + t.Fatalf("invalid memory metric was emitted:\n%s", gotBody) + } + if strings.Contains(gotBody, "techulus_service_memory_used_bytes") || strings.Contains(gotBody, "techulus_service_memory_usage_percent") { + t.Fatalf("invalid aggregate memory metric was emitted:\n%s", gotBody) + } +} diff --git a/docs/api/public-api.mdx b/docs/api/public-api.mdx index 745d70cd..35a0cd3d 100644 --- a/docs/api/public-api.mdx +++ b/docs/api/public-api.mdx @@ -206,6 +206,23 @@ Use automatic placement for stateless services: { "mode": "automatic", "replicas": 3 } ``` +To autoscale an eligible service, replace fixed `replicas` with a range: + +```json +{ + "mode": "automatic", + "autoscaling": { "minReplicas": 2, "maxReplicas": 8 } +} +``` + +Autoscaling requires automatic placement, a stateless non-serverless service +without volumes, and both CPU and memory limits. Minimum and maximum values must +be from 1 through 32, with minimum no greater than maximum. CPU and memory use +fixed 60% targets. Configuration responses expose the mutable policy in +`current.placement` and the deployed policy plus concrete active target in +`active.placement`. Policy changes participate in plan/apply fingerprints. +To disable autoscaling, send the fixed automatic-placement form with `replicas`. + Use manual placement to choose exact servers: ```json @@ -219,6 +236,10 @@ Use manual placement to choose exact servers: Manual placement requires online servers with WireGuard configured. Serverless services require proxy servers, including when automatic placement is used. A replacement that removes the final public HTTP domain can move the service to worker placement because the same update disables serverless. Automatic placement is not available for stateful or volume-backed services. Submit replica changes through `placement`; the API rejects a top-level `replicas` field. +CLI manifests remain fixed-replica only in V1. Configure autoscaling through the +dashboard or public configuration API. `tc apply` continues to send the fixed +replica manifest contract. + The API only manages stateless services with HTTP ports. Existing volumes, stateful mode, TCP or UDP ports, TLS passthrough, or invalid resource limits return a conflict with an actionable code. ## Deployments and builds diff --git a/docs/services/scaling.mdx b/docs/services/scaling.mdx index 416898e1..e43f486e 100644 --- a/docs/services/scaling.mdx +++ b/docs/services/scaling.mdx @@ -9,6 +9,33 @@ Each service can run multiple replicas across your cluster. Configure how many r Replica count ranges from 1 to 32 per service. +Automatic placement supports two replica-management modes: + +- **Fixed** keeps the configured replica count running. +- **Autoscaled** keeps the active count within a configured minimum and maximum + from 1 through 32. CPU and memory each use a fixed 60% target; thresholds are + not configurable. + +Autoscaling is available only for automatic-placement, stateless, +non-serverless services without volumes. Both CPU and memory limits must be +configured because utilization is measured against those limits. Saving stages +the policy; deploying snapshots it into the immutable revision and activates it. + +The controller evaluates trailing three-minute CPU and memory averages every +minute. Scale-down additionally requires five minutes of complete low-usage +observations. Missing or stale metrics hold the current count. A scaling attempt +starts a ten-minute cooldown, including failed attempts. + +Autoscaling V1 scales up directly to the CPU- or memory-based recommendation. +After stabilization, it scales down one replica at a time. Each change uses the +normal rolling full-fleet replacement. Existing containers remain active until +the complete replacement fleet is healthy and routing has converged, but this +can temporarily require both the old and replacement capacity and can reset +long-lived connections. + +Changing the configured range clamps an active count outside the new bounds +directly to the nearest bound in one rollout. + ## Serverless scaling Public HTTP services can be configured to sleep when idle on proxy nodes. A @@ -42,9 +69,9 @@ Serverless services require at least one configured replica. ## Placement Stateless services support automatic and manual placement. Automatic placement -stores the desired replica count and distributes replicas across online, -configured nodes during rollout. Manual placement selects exact target servers -and replica counts. +stores either a fixed count or an autoscaling range and distributes the concrete +replica target across online, configured nodes during rollout. Manual placement +selects exact target servers and replica counts. Serverless services support automatic placement across online, configured proxy nodes. Public ingress must use health checks to avoid proxy nodes that do not diff --git a/web/actions/projects.ts b/web/actions/projects.ts index b034026b..3d1c983a 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -1010,6 +1010,12 @@ export async function updateServiceResourceLimits( if (!service) { throw new Error("Service not found"); } + if ( + service.autoscalingEnabled && + (validated.cpuCores === null || validated.memoryMb === null) + ) { + throw new Error("Disable autoscaling before removing resource limits"); + } await db.transaction(async (tx) => { await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext(${serviceId}))`); @@ -1086,6 +1092,9 @@ export async function updateServiceServerlessSettings( } if (validated.enabled) { + if (service.autoscalingEnabled) { + throw new Error("Disable autoscaling before enabling serverless"); + } const publicHttpPorts = await tx .select({ id: servicePorts.id }) .from(servicePorts) @@ -1171,17 +1180,35 @@ export type ServiceConfigUpdate = { ports?: { add?: PortConfig[]; remove?: string[] }; placement?: | { mode: "automatic"; replicas: number } + | { + mode: "automatic"; + autoscaling: { minReplicas: number; maxReplicas: number }; + } | { mode: "manual"; placements: { serverId: string; count: number }[]; }; }; -const placementInputSchema = z.discriminatedUnion("mode", [ +const autoscalingRangeSchema = z + .strictObject({ + minReplicas: z.number().int().min(1).max(32), + maxReplicas: z.number().int().min(1).max(32), + }) + .refine((value) => value.minReplicas <= value.maxReplicas, { + message: "Minimum replicas cannot exceed maximum replicas", + path: ["minReplicas"], + }); + +const placementInputSchema = z.union([ z.strictObject({ mode: z.literal("automatic"), replicas: z.number().int().min(1).max(32), }), + z.strictObject({ + mode: z.literal("automatic"), + autoscaling: autoscalingRangeSchema, + }), z .strictObject({ mode: z.literal("manual"), @@ -1407,6 +1434,8 @@ export async function updateServiceConfig( serverlessEnabled: services.serverlessEnabled, stateful: services.stateful, placementMode: services.placementMode, + resourceCpuLimit: services.resourceCpuLimit, + resourceMemoryLimitMb: services.resourceMemoryLimitMb, }) .from(services) .where(eq(services.id, serviceId)) @@ -1423,9 +1452,39 @@ export async function updateServiceConfig( throw new Error( "Automatic placement is not supported for stateful services or services with volumes", ); + if ("autoscaling" in placement) { + if (currentService.serverlessEnabled) + throw new Error("Disable serverless before enabling autoscaling"); + if ( + currentService.resourceCpuLimit === null || + currentService.resourceMemoryLimitMb === null + ) + throw new Error( + "Set both CPU and memory limits before enabling autoscaling", + ); + } + const replicas = + "autoscaling" in placement + ? Math.min( + placement.autoscaling.maxReplicas, + Math.max(placement.autoscaling.minReplicas, service.replicas), + ) + : placement.replicas; await tx .update(services) - .set({ placementMode: "automatic", replicas: placement.replicas }) + .set({ + placementMode: "automatic", + replicas, + autoscalingEnabled: "autoscaling" in placement, + autoscalingMinReplicas: + "autoscaling" in placement + ? placement.autoscaling.minReplicas + : replicas, + autoscalingMaxReplicas: + "autoscaling" in placement + ? placement.autoscaling.maxReplicas + : replicas, + }) .where(eq(services.id, serviceId)); await tx .delete(serviceReplicas) @@ -1470,6 +1529,7 @@ export async function updateServiceConfig( .set({ placementMode: "manual", replicas: replicas.reduce((sum, replica) => sum + replica.count, 0), + autoscalingEnabled: false, }) .where(eq(services.id, serviceId)); await tx diff --git a/web/app/api/inngest/route.ts b/web/app/api/inngest/route.ts index 26237d24..c197d81f 100644 --- a/web/app/api/inngest/route.ts +++ b/web/app/api/inngest/route.ts @@ -2,6 +2,7 @@ import { serve } from "inngest/next"; import { inngest } from "@/lib/inngest/client"; import { agentUpgradeTimeoutCheck, + autoscalingCheck, backupWorkflow, buildTriggerWorkflow, buildWorkflow, @@ -29,6 +30,7 @@ import { export const { GET, POST, PUT } = serve({ client: inngest, functions: [ + autoscalingCheck, rolloutWorkflow, onDeploymentFailed, staleServerCheck, diff --git a/web/components/service/details/replicas-section.tsx b/web/components/service/details/replicas-section.tsx index be1632c8..3f79928a 100644 --- a/web/components/service/details/replicas-section.tsx +++ b/web/components/service/details/replicas-section.tsx @@ -15,6 +15,7 @@ import { import { Input } from "@/components/ui/input"; import { Slider } from "@/components/ui/slider"; import { Spinner } from "@/components/ui/spinner"; +import { Switch } from "@/components/ui/switch"; import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs"; import type { Server as ServerType, @@ -27,6 +28,7 @@ type ServerInfo = Pick< "id" | "name" | "isProxy" | "status" | "wireguardIp" >; type PlacementMode = "manual" | "automatic"; +type ReplicaManagement = "fixed" | "autoscaled"; const fetcher = async (url: string): Promise => { const servers = await fetchJson(url); @@ -59,6 +61,13 @@ export const ReplicasSection = memo(function ReplicasSection({ service.placementMode, ); const [desiredReplicas, setDesiredReplicas] = useState(service.replicas); + const [replicaManagement, setReplicaManagement] = useState( + service.autoscalingEnabled ? "autoscaled" : "fixed", + ); + const [autoscalingRange, setAutoscalingRange] = useState([ + service.autoscalingMinReplicas, + service.autoscalingMaxReplicas, + ]); const [isEditing, setIsEditing] = useState(false); const [isSaving, setIsSaving] = useState(false); @@ -91,6 +100,11 @@ export const ReplicasSection = memo(function ReplicasSection({ setLocalReplicas(replicaMap); setPlacementMode(service.placementMode); setDesiredReplicas(service.replicas); + setReplicaManagement(service.autoscalingEnabled ? "autoscaled" : "fixed"); + setAutoscalingRange([ + service.autoscalingMinReplicas, + service.autoscalingMaxReplicas, + ]); } }, [ servers, @@ -99,6 +113,9 @@ export const ReplicasSection = memo(function ReplicasSection({ service.lockedServerId, service.placementMode, service.replicas, + service.autoscalingEnabled, + service.autoscalingMinReplicas, + service.autoscalingMaxReplicas, isEditing, ]); @@ -118,8 +135,15 @@ export const ReplicasSection = memo(function ReplicasSection({ placementMode !== service.placementMode ) return; - if (placementMode === "automatic" && desiredReplicas === service.replicas) { - setIsEditing(false); + if (placementMode === "automatic") { + const autoscalingMatches = + (replicaManagement === "autoscaled") === service.autoscalingEnabled && + (replicaManagement !== "autoscaled" || + (autoscalingRange[0] === service.autoscalingMinReplicas && + autoscalingRange[1] === service.autoscalingMaxReplicas)); + if (desiredReplicas === service.replicas && autoscalingMatches) { + setIsEditing(false); + } return; } if (placementMode === "manual") { @@ -141,11 +165,16 @@ export const ReplicasSection = memo(function ReplicasSection({ }, [ configuredReplicas, desiredReplicas, + replicaManagement, + autoscalingRange, isEditing, localReplicas, placementMode, service.placementMode, service.replicas, + service.autoscalingEnabled, + service.autoscalingMinReplicas, + service.autoscalingMaxReplicas, service.stateful, ]); @@ -157,7 +186,13 @@ export const ReplicasSection = memo(function ReplicasSection({ } if (placementMode !== service.placementMode) return true; if (placementMode === "automatic") { - return desiredReplicas !== service.replicas; + return ( + desiredReplicas !== service.replicas || + (replicaManagement === "autoscaled") !== service.autoscalingEnabled || + (replicaManagement === "autoscaled" && + (autoscalingRange[0] !== service.autoscalingMinReplicas || + autoscalingRange[1] !== service.autoscalingMaxReplicas)) + ); } const configuredMap = new Map( @@ -171,10 +206,15 @@ export const ReplicasSection = memo(function ReplicasSection({ }, [ configuredReplicas, desiredReplicas, + replicaManagement, + autoscalingRange, localReplicas, placementMode, service.placementMode, service.replicas, + service.autoscalingEnabled, + service.autoscalingMinReplicas, + service.autoscalingMaxReplicas, service.stateful, selectedServerId, ]); @@ -236,7 +276,16 @@ export const ReplicasSection = memo(function ReplicasSection({ return; } else { await updateServiceConfig(service.id, { - placement: { mode: "automatic", replicas: desiredReplicas }, + placement: + replicaManagement === "autoscaled" + ? { + mode: "automatic", + autoscaling: { + minReplicas: autoscalingRange[0], + maxReplicas: autoscalingRange[1], + }, + } + : { mode: "automatic", replicas: desiredReplicas }, }); onUpdate(); return; @@ -270,6 +319,14 @@ export const ReplicasSection = memo(function ReplicasSection({ }; const manualTotalIsValid = totalReplicas >= 1 && totalReplicas <= 32; + const autoscalingIneligibility = service.serverlessEnabled + ? "Disable serverless before enabling autoscaling." + : service.volumes && service.volumes.length > 0 + ? "Remove volumes before enabling autoscaling." + : service.resourceCpuLimit == null || + service.resourceMemoryLimitMb == null + ? "Set both CPU and memory limits before enabling autoscaling." + : null; if (service.stateful) { return ( @@ -385,7 +442,7 @@ export const ReplicasSection = memo(function ReplicasSection({ return (
@@ -402,39 +459,99 @@ export const ReplicasSection = memo(function ReplicasSection({ {placementMode === "automatic" ? (
-
+
-

Desired replicas

+

Autoscaling

- The control plane distributes replicas evenly across healthy - {service.serverlessEnabled ? " proxy nodes" : " nodes"} and - moves them after failures. + Adjusts replicas within your selected range using average CPU + and memory usage. Uses fixed 60% targets.

- - {desiredReplicas} - -
-
- { + { setIsEditing(true); - setDesiredReplicas(value); + setReplicaManagement(checked ? "autoscaled" : "fixed"); }} + aria-label="Enable autoscaling" /> -
- 1 - 32 -
+ {replicaManagement === "autoscaled" && autoscalingIneligibility ? ( +

+ {autoscalingIneligibility} +

+ ) : null} + {replicaManagement === "autoscaled" ? ( +
+
+ Replica range + + {autoscalingRange[0]}-{autoscalingRange[1]} + +
+ + index === 0 ? "Minimum replicas" : "Maximum replicas" + } + onValueChange={(value) => { + setIsEditing(true); + setAutoscalingRange(value); + }} + /> +

+ Scaling changes perform a rolling full-fleet replacement. +

+
+ ) : ( +
+
+
+

Desired replicas

+

+ The control plane distributes replicas evenly across + healthy + {service.serverlessEnabled ? " proxy nodes" : " nodes"}{" "} + and moves them after failures. +

+
+ + {desiredReplicas} + +
+
+ { + setIsEditing(true); + setDesiredReplicas(value); + }} + /> +
+ 1 + 32 +
+
+
+ )} {hasChanges ? (
-
diff --git a/web/components/service/details/service-details-overview.tsx b/web/components/service/details/service-details-overview.tsx index 44d24917..7e2f9875 100644 --- a/web/components/service/details/service-details-overview.tsx +++ b/web/components/service/details/service-details-overview.tsx @@ -496,6 +496,13 @@ function ServiceConfigPanel({ {hasResourceLimits ? ( {formatResources(service)} ) : null} + {service.activeConfig?.placement?.autoscaling?.enabled ? ( + + {service.activeConfig.placement.autoscaling.minReplicas}- + {service.activeConfig.placement.autoscaling.maxReplicas} ·{" "} + {overview.runningDeployments} active + + ) : null}
diff --git a/web/db/schema.ts b/web/db/schema.ts index 2e2edbdc..1c48e43c 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -466,6 +466,13 @@ export const services = pgTable( githubBranch: text("github_branch").default("main"), githubRootDir: text("github_root_dir"), replicas: integer("replicas").notNull().default(1), + autoscalingEnabled: boolean("autoscaling_enabled").notNull().default(false), + autoscalingMinReplicas: integer("autoscaling_min_replicas") + .notNull() + .default(1), + autoscalingMaxReplicas: integer("autoscaling_max_replicas") + .notNull() + .default(1), placementMode: text("placement_mode", { enum: ["manual", "automatic"] }) .notNull() .default("manual"), @@ -476,6 +483,9 @@ export const services = pgTable( "last_automatic_recovery_attempt_at", { withTimezone: true }, ), + lastAutoscaleAttemptAt: timestamp("last_autoscale_attempt_at", { + withTimezone: true, + }), stateful: boolean("stateful").notNull().default(false), lockedServerId: text("locked_server_id").references(() => servers.id, { onDelete: "set null", @@ -539,6 +549,9 @@ export const services = pgTable( table.environmentId, ), index("services_environment_id_idx").on(table.environmentId), + index("services_last_autoscale_attempt_idx").on( + table.lastAutoscaleAttemptAt, + ), ], ); @@ -753,6 +766,11 @@ export const deployments = pgTable( index("deployments_container_id_idx").on(table.containerId), index("deployments_rollout_id_idx").on(table.rolloutId), index("deployments_service_id_idx").on(table.serviceId), + index("deployments_service_traffic_runtime_idx").on( + table.serviceId, + table.trafficState, + table.runtimeDesiredState, + ), index("deployments_service_revision_id_idx").on(table.serviceRevisionId), index("deployments_server_id_idx").on(table.serverId), uniqueIndex("deployments_server_id_ip_address_unique_idx") @@ -794,6 +812,7 @@ export const rollouts = pgTable( table.serviceId, table.createdAt, ), + index("rollouts_service_status_idx").on(table.serviceId, table.status), uniqueIndex("rollouts_service_revision_id_unique_idx").on( table.serviceRevisionId, ), diff --git a/web/lib/autoscaling.ts b/web/lib/autoscaling.ts new file mode 100644 index 00000000..b043a36e --- /dev/null +++ b/web/lib/autoscaling.ts @@ -0,0 +1,236 @@ +import { + escapePromQL, + getQueryEndpoint, + isMetricsEnabled, + queryRangePromQL, +} from "@/lib/victoria-metrics"; + +const TARGET_UTILIZATION = 60; +const WINDOW_SECONDS = 180; +const STABILIZATION_MINUTES = 5; + +export type AutoscalingHoldReason = + | "metrics-disabled" + | "metrics-not-queried" + | "no-active-deployments" + | "provider-error" + | "unexpected-series" + | "invalid-sample" + | "incomplete-coverage" + | "stale-sample"; + +export type AutoscalingEvaluationPoint = { + timestamp: string; + cpuUtilizationPercent: number; + memoryUtilizationPercent: number; + coverage: Array<{ + deploymentId: string; + cpuSourceTimestamp: string; + memorySourceTimestamp: string; + }>; +}; + +export type AutoscalingMetricResult = + | { status: "ready"; points: AutoscalingEvaluationPoint[] } + | { status: "hold"; reason: AutoscalingHoldReason }; + +export type AutoscalingRecommendation = + | { + status: "scale"; + direction: "up" | "down"; + targetReplicas: number; + reason: "above-maximum" | "below-minimum" | "utilization"; + } + | { + status: "hold"; + reason: AutoscalingHoldReason | "stable" | "downscale-stabilizing"; + }; + +type MatrixResult = Awaited>[number]; + +/** + * Reads six one-minute evaluation points (now and the previous five minutes). + * CPU is normalized against configured cores. The runtime rounds `--cpus` to + * two decimals, so this introduces a small normalization error for finer limits. + */ +export async function queryAutoscalingMetrics(options: { + serviceId: string; + deploymentIds: string[]; + cpuLimitCores: number; + memoryLimitMb: number; + now?: Date; +}): Promise { + if (!isMetricsEnabled()) + return { status: "hold", reason: "metrics-disabled" }; + const deploymentIds = [...new Set(options.deploymentIds)].sort(); + if (deploymentIds.length === 0) + return { status: "hold", reason: "no-active-deployments" }; + if (!(options.cpuLimitCores > 0) || !(options.memoryLimitMb > 0)) + return { status: "hold", reason: "invalid-sample" }; + + const endpoint = getQueryEndpoint(); + if (!endpoint) return { status: "hold", reason: "metrics-disabled" }; + const end = new Date( + Math.floor((options.now ?? new Date()).getTime() / 60_000) * 60_000, + ); + const start = new Date(end.getTime() - STABILIZATION_MINUTES * 60_000); + const ids = deploymentIds.map(escapeRegex).join("|"); + const selector = `service_id="${escapePromQL(options.serviceId)}",deployment_id=~"^(?:${ids})$"`; + const metricQueries = [ + `avg_over_time(techulus_deployment_cpu_usage_cores{${selector}}[3m])`, + `tlast_over_time(techulus_deployment_cpu_usage_cores{${selector}}[3m])`, + `avg_over_time(techulus_deployment_memory_used_bytes{${selector}}[3m])`, + `tlast_over_time(techulus_deployment_memory_used_bytes{${selector}}[3m])`, + ]; + + let results: MatrixResult[][]; + try { + results = await Promise.all( + metricQueries.map((query) => + queryRangePromQL(endpoint, { query, start, end, stepSeconds: 60 }), + ), + ); + } catch { + return { status: "hold", reason: "provider-error" }; + } + + const allowed = new Set(deploymentIds); + if ( + results.some((set) => + set.some((series) => !allowed.has(series.metric.deployment_id ?? "")), + ) + ) + return { status: "hold", reason: "unexpected-series" }; + const maps = results.map(toSeriesMap); + const points: AutoscalingEvaluationPoint[] = []; + for (let time = start.getTime(); time <= end.getTime(); time += 60_000) { + let cpuTotal = 0; + let memoryTotal = 0; + const coverage: AutoscalingEvaluationPoint["coverage"] = []; + for (const deploymentId of deploymentIds) { + const values = maps.map((map) => map.get(deploymentId)?.get(time / 1000)); + if (values.some((value) => value === undefined)) + return { status: "hold", reason: "incomplete-coverage" }; + const [cpu, cpuTimestamp, memory, memoryTimestamp] = values as number[]; + if ( + ![cpu, cpuTimestamp, memory, memoryTimestamp].every(Number.isFinite) || + cpu < 0 || + memory < 0 + ) + return { status: "hold", reason: "invalid-sample" }; + if ( + time / 1000 - cpuTimestamp > WINDOW_SECONDS || + time / 1000 - memoryTimestamp > WINDOW_SECONDS + ) + return { status: "hold", reason: "stale-sample" }; + cpuTotal += (cpu / options.cpuLimitCores) * 100; + memoryTotal += (memory / (options.memoryLimitMb * 1024 * 1024)) * 100; + coverage.push({ + deploymentId, + cpuSourceTimestamp: new Date(cpuTimestamp * 1000).toISOString(), + memorySourceTimestamp: new Date(memoryTimestamp * 1000).toISOString(), + }); + } + points.push({ + timestamp: new Date(time).toISOString(), + cpuUtilizationPercent: cpuTotal / deploymentIds.length, + memoryUtilizationPercent: memoryTotal / deploymentIds.length, + coverage, + }); + } + return { status: "ready", points }; +} + +export function calculateAutoscalingRecommendation(options: { + currentReplicas: number; + minReplicas: number; + maxReplicas: number; + metrics?: AutoscalingMetricResult; +}): AutoscalingRecommendation { + const { currentReplicas, minReplicas, maxReplicas } = options; + if (currentReplicas < minReplicas) + return { + status: "scale", + direction: "up", + targetReplicas: minReplicas, + reason: "below-minimum", + }; + if (currentReplicas > maxReplicas) + return { + status: "scale", + direction: "down", + targetReplicas: maxReplicas, + reason: "above-maximum", + }; + if (!options.metrics) + return { status: "hold", reason: "metrics-not-queried" }; + if (options.metrics.status === "hold") return options.metrics; + const recommendations = options.metrics.points.map((point) => ({ + cpu: resourceRecommendation(currentReplicas, point.cpuUtilizationPercent), + memory: resourceRecommendation( + currentReplicas, + point.memoryUtilizationPercent, + ), + })); + const latest = recommendations.at(-1); + if (!latest) return { status: "hold", reason: "incomplete-coverage" }; + const desiredUp = Math.max(latest.cpu, latest.memory); + if (desiredUp > currentReplicas && currentReplicas < maxReplicas) + return { + status: "scale", + direction: "up", + targetReplicas: Math.min(maxReplicas, desiredUp), + reason: "utilization", + }; + if (latest.cpu >= currentReplicas || latest.memory >= currentReplicas) + return { status: "hold", reason: "stable" }; + if ( + recommendations.length < STABILIZATION_MINUTES + 1 || + recommendations.some( + (value) => Math.max(value.cpu, value.memory) >= currentReplicas, + ) + ) + return { status: "hold", reason: "downscale-stabilizing" }; + const targetReplicas = Math.max(minReplicas, currentReplicas - 1); + if (targetReplicas === currentReplicas) + return { status: "hold", reason: "stable" }; + return { + status: "scale", + direction: "down", + targetReplicas, + reason: "utilization", + }; +} + +function resourceRecommendation(current: number, utilization: number): number { + if (utilization >= 54 && utilization <= 66) return current; + return Math.ceil((current * utilization) / TARGET_UTILIZATION); +} + +function toSeriesMap( + results: MatrixResult[], +): Map> { + const output = new Map>(); + const duplicates = new Set(); + for (const result of results) { + const deploymentId = result.metric.deployment_id; + if (!deploymentId) continue; + if (duplicates.has(deploymentId)) continue; + if (output.has(deploymentId)) { + output.delete(deploymentId); + duplicates.add(deploymentId); + continue; + } + output.set( + deploymentId, + new Map( + result.values.map(([timestamp, value]) => [timestamp, Number(value)]), + ), + ); + } + return output; +} + +function escapeRegex(value: string): string { + return escapePromQL(value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")); +} diff --git a/web/lib/inngest/functions/crons.ts b/web/lib/inngest/functions/crons.ts index c137e4b3..2b1052de 100644 --- a/web/lib/inngest/functions/crons.ts +++ b/web/lib/inngest/functions/crons.ts @@ -14,6 +14,7 @@ import { MAX_AUTOMATIC_RECOVERIES_PER_RUN, rebalanceAutomaticServices, recoverInvalidAutomaticPlacements, + runAutoscalingController, } from "@/lib/scheduler"; import { inngest } from "../client"; @@ -47,6 +48,17 @@ export const staleServerCheck = inngest.createFunction( }, ); +export const autoscalingCheck = inngest.createFunction( + { + id: "cron-autoscaling-check", + triggers: [cron("* * * * *")], + singleton: { mode: "skip" }, + }, + async ({ step }) => { + await step.run("evaluate-autoscaling-services", runAutoscalingController); + }, +); + export const scheduledDeploymentsCheck = inngest.createFunction( { id: "cron-scheduled-deployments", diff --git a/web/lib/inngest/functions/index.ts b/web/lib/inngest/functions/index.ts index 7e94bf59..fac237a5 100644 --- a/web/lib/inngest/functions/index.ts +++ b/web/lib/inngest/functions/index.ts @@ -3,6 +3,7 @@ export { buildTriggerWorkflow } from "./build-trigger-workflow"; export { buildWorkflow } from "./build-workflow"; export { agentUpgradeTimeoutCheck, + autoscalingCheck, certificateRenewal, challengeCleanup, controlPlaneUpdateCheck, diff --git a/web/lib/inngest/functions/rollout-helpers.ts b/web/lib/inngest/functions/rollout-helpers.ts index 60322d35..e056f122 100644 --- a/web/lib/inngest/functions/rollout-helpers.ts +++ b/web/lib/inngest/functions/rollout-helpers.ts @@ -540,7 +540,10 @@ export async function completeRollout( const serviceUpdate = specification.placement.mode === "automatic" - ? { lastAutomaticPlacementAt: new Date() } + ? { + lastAutomaticPlacementAt: new Date(), + replicas: specification.placement.replicas, + } : lockedServerId ? { lockedServerId } : null; diff --git a/web/lib/public-api.ts b/web/lib/public-api.ts index cc0f51e7..a6159dce 100644 --- a/web/lib/public-api.ts +++ b/web/lib/public-api.ts @@ -280,7 +280,7 @@ function sanitizeSpec(specification: unknown) { const replicas = getServiceRevisionTotalReplicas(spec); const placement = spec.placement.mode === "automatic" - ? { mode: "automatic" as const, replicas } + ? { mode: "automatic" as const, replicas, autoscaling: spec.autoscaling } : { mode: "manual" as const, placements: spec.placements, replicas }; return { source: @@ -380,7 +380,17 @@ export async function safeConfiguration(service: NestedService) { }); const placement = service.placementMode === "automatic" - ? { mode: "automatic" as const, replicas: replicaCount } + ? { + mode: "automatic" as const, + replicas: replicaCount, + autoscaling: service.autoscalingEnabled + ? { + enabled: true as const, + minReplicas: service.autoscalingMinReplicas, + maxReplicas: service.autoscalingMaxReplicas, + } + : undefined, + } : { mode: "manual" as const, placements: sortedPlacements, @@ -551,11 +561,25 @@ const hostnameSchema = z /^[a-z0-9]+(?:-[a-z0-9]+)*$/, "hostname must contain only lowercase letters, numbers, and hyphens", ); -export const placementSchema = z.discriminatedUnion("mode", [ +const autoscalingRangeSchema = z + .strictObject({ + enabled: z.literal(true).optional(), + minReplicas: z.number().int().min(1).max(32), + maxReplicas: z.number().int().min(1).max(32), + }) + .refine((value) => value.minReplicas <= value.maxReplicas, { + message: "Minimum replicas cannot exceed maximum replicas", + }); +export const placementSchema = z.union([ z.strictObject({ mode: z.literal("automatic"), replicas: z.number().int().min(1).max(32), }), + z.strictObject({ + mode: z.literal("automatic"), + replicas: z.number().int().min(1).max(32).optional(), + autoscaling: autoscalingRangeSchema, + }), z .strictObject({ mode: z.literal("manual"), @@ -684,7 +708,15 @@ function canonicalReplacementState( ), placement: service.placementMode === "automatic" - ? { mode: "automatic" as const, replicas: service.replicas } + ? service.autoscalingEnabled + ? { + mode: "automatic" as const, + autoscaling: { + minReplicas: service.autoscalingMinReplicas, + maxReplicas: service.autoscalingMaxReplicas, + }, + } + : { mode: "automatic" as const, replicas: service.replicas } : { mode: "manual" as const, placements: placements @@ -721,7 +753,15 @@ export function canonicalDesired(input: ReplacementInput) { a.serverId.localeCompare(b.serverId, "en"), ), } - : input.placement, + : "autoscaling" in input.placement + ? { + mode: "automatic" as const, + autoscaling: { + minReplicas: input.placement.autoscaling.minReplicas, + maxReplicas: input.placement.autoscaling.maxReplicas, + }, + } + : input.placement, }; } @@ -897,6 +937,23 @@ async function replaceConfigurationInternal( 400, ); } + if ( + input.placement.mode === "automatic" && + "autoscaling" in input.placement + ) { + if (effectiveServerlessEnabled) + domainError( + "Autoscaling is not supported for serverless services", + "AUTOSCALING_UNSUPPORTED", + 400, + ); + if (input.resources?.cpuCores == null || input.resources.memoryMb == null) + domainError( + "Autoscaling requires both CPU and memory limits", + "AUTOSCALING_RESOURCE_LIMITS_REQUIRED", + 400, + ); + } const blockers = getManagementBlockers({ service: persisted, source, @@ -1087,9 +1144,27 @@ async function replaceConfigurationInternal( set.image = input.source.image; } if (input.placement) { + const requestedPlacement = + input.placement.mode === "automatic" && "autoscaling" in input.placement + ? { + mode: "automatic" as const, + autoscaling: { + minReplicas: input.placement.autoscaling.minReplicas, + maxReplicas: input.placement.autoscaling.maxReplicas, + }, + } + : input.placement; const desiredReplicas = input.placement.mode === "automatic" - ? input.placement.replicas + ? "autoscaling" in input.placement + ? Math.min( + input.placement.autoscaling.maxReplicas, + Math.max( + input.placement.autoscaling.minReplicas, + persisted.replicas, + ), + ) + : input.placement.replicas : input.placement.placements.reduce( (sum, item) => sum + item.count, 0, @@ -1098,25 +1173,46 @@ async function replaceConfigurationInternal( changed( "placement", persisted.placementMode === "automatic" - ? { mode: "automatic", replicas: persisted.replicas } + ? persisted.autoscalingEnabled + ? { + mode: "automatic", + autoscaling: { + minReplicas: persisted.autoscalingMinReplicas, + maxReplicas: persisted.autoscalingMaxReplicas, + }, + } + : { mode: "automatic", replicas: persisted.replicas } : { mode: "manual", placements: placements .map(({ serverId, count }) => ({ serverId, count })) .toSorted((a, b) => a.serverId.localeCompare(b.serverId)), }, - input.placement.mode === "manual" + requestedPlacement.mode === "manual" ? { - ...input.placement, - placements: input.placement.placements.toSorted((a, b) => + ...requestedPlacement, + placements: requestedPlacement.placements.toSorted((a, b) => a.serverId.localeCompare(b.serverId), ), } - : input.placement, + : requestedPlacement, ) ) { set.placementMode = input.placement.mode; set.replicas = desiredReplicas; + set.autoscalingEnabled = + input.placement.mode === "automatic" && + "autoscaling" in input.placement; + set.autoscalingMinReplicas = + input.placement.mode === "automatic" && + "autoscaling" in input.placement + ? input.placement.autoscaling.minReplicas + : desiredReplicas; + set.autoscalingMaxReplicas = + input.placement.mode === "automatic" && + "autoscaling" in input.placement + ? input.placement.autoscaling.maxReplicas + : desiredReplicas; await tx .delete(serviceReplicas) .where(eq(serviceReplicas.serviceId, service.id)); diff --git a/web/lib/scheduler.ts b/web/lib/scheduler.ts index 13f6ea67..6593e6b9 100644 --- a/web/lib/scheduler.ts +++ b/web/lib/scheduler.ts @@ -1,5 +1,15 @@ import { CronExpressionParser } from "cron-parser"; -import { and, eq, inArray, isNotNull, isNull, lt, ne, sql } from "drizzle-orm"; +import { + and, + eq, + inArray, + isNotNull, + isNull, + lt, + ne, + or, + sql, +} from "drizzle-orm"; import { db } from "@/db"; import { deployments, @@ -9,6 +19,10 @@ import { services, workQueue, } from "@/db/schema"; +import { + calculateAutoscalingRecommendation, + queryAutoscalingMetrics, +} from "@/lib/autoscaling"; import { DAY_IN_MILLISECONDS, isDateAfter, @@ -24,7 +38,11 @@ import { import { notify } from "@/lib/notifications"; import { sendRolloutCreated } from "@/lib/rollout-enqueue"; import { parseServiceRevisionSpec } from "@/lib/service-revision-changes"; -import { cloneActiveRevisionAndQueueSystemRollout } from "@/lib/service-revisions"; +import { + AUTOSCALE_ATTEMPT_COOLDOWN_MS, + cloneActiveRevisionAndQueueSystemRollout, + cloneActiveRevisionForAutoscaling, +} from "@/lib/service-revisions"; import { WORK_QUEUE_LEASE_DURATION_MS, WORK_QUEUE_MAX_ATTEMPTS, @@ -34,6 +52,186 @@ const STALE_THRESHOLD_MS = 75 * SECOND_IN_MILLISECONDS; export const AUTOMATIC_PLACEMENT_COOLDOWN_MS = 30 * MINUTE_IN_MILLISECONDS; export const MAX_REBALANCES_PER_RUN = 5; export const MAX_AUTOMATIC_RECOVERIES_PER_RUN = 5; +export const MAX_AUTOSCALING_CANDIDATES_PER_RUN = 25; +export const MAX_AUTOSCALING_ROLLOUTS_PER_RUN = 5; + +export async function runAutoscalingController( + maxCreated = MAX_AUTOSCALING_ROLLOUTS_PER_RUN, +): Promise { + if (maxCreated <= 0) return 0; + const now = new Date(); + const cooldownCutoff = new Date( + now.getTime() - AUTOSCALE_ATTEMPT_COOLDOWN_MS, + ); + const candidates = await db + .select({ + id: services.id, + name: services.name, + lastAutoscaleAttemptAt: services.lastAutoscaleAttemptAt, + }) + .from(services) + .innerJoin(deployments, eq(deployments.serviceId, services.id)) + .innerJoin( + serviceRevisions, + eq(serviceRevisions.id, deployments.serviceRevisionId), + ) + .where( + and( + isNull(services.deletedAt), + or( + isNull(services.lastAutoscaleAttemptAt), + lt(services.lastAutoscaleAttemptAt, cooldownCutoff), + ), + eq(deployments.trafficState, "active"), + inArray(deployments.runtimeDesiredState, ["running", "stopped"]), + eq( + sql`${serviceRevisions.specification} -> 'autoscaling' ->> 'enabled'`, + "true", + ), + ), + ) + .groupBy(services.id, services.name, services.lastAutoscaleAttemptAt) + .orderBy(sql`random()`) + .limit(MAX_AUTOSCALING_CANDIDATES_PER_RUN); + let created = 0; + for (const service of candidates) { + if (created >= maxCreated) break; + const skip = (reason: string) => + console.log(`[autoscaling] skipping ${service.name}: ${reason}`); + try { + const [state, pending] = await Promise.all([ + db + .select({ + deploymentId: deployments.id, + revisionId: deployments.serviceRevisionId, + runtimeDesiredState: deployments.runtimeDesiredState, + observedPhase: deployments.observedPhase, + specification: serviceRevisions.specification, + migrationStatus: services.migrationStatus, + lastAutoscaleAttemptAt: services.lastAutoscaleAttemptAt, + }) + .from(deployments) + .innerJoin(services, eq(services.id, deployments.serviceId)) + .innerJoin( + serviceRevisions, + eq(serviceRevisions.id, deployments.serviceRevisionId), + ) + .where( + and( + eq(deployments.serviceId, service.id), + eq(deployments.trafficState, "active"), + inArray(deployments.runtimeDesiredState, ["running", "stopped"]), + ), + ) + .orderBy(deployments.id), + db + .select({ id: rollouts.id }) + .from(rollouts) + .where( + and( + eq(rollouts.serviceId, service.id), + inArray(rollouts.status, ["queued", "in_progress"]), + ), + ) + .limit(1) + .then((rows) => rows[0]), + ]); + const first = state[0]; + if (!first) { + skip("no active topology"); + continue; + } + if (first.migrationStatus) { + skip("migration in progress"); + continue; + } + if (pending) { + skip("rollout in progress"); + continue; + } + if (state.some((item) => item.runtimeDesiredState === "stopped")) { + skip("active runtime is stopped"); + continue; + } + if ( + new Set(state.map((item) => item.revisionId)).size !== 1 || + state.some( + (item) => !["healthy", "running"].includes(item.observedPhase), + ) + ) { + skip("active topology is not exactly one ready revision"); + continue; + } + if ( + first.lastAutoscaleAttemptAt && + now.getTime() - first.lastAutoscaleAttemptAt.getTime() < + AUTOSCALE_ATTEMPT_COOLDOWN_MS + ) { + skip("attempt cooldown"); + continue; + } + const spec = parseServiceRevisionSpec(first.specification); + if ( + spec.placement.mode !== "automatic" || + !spec.autoscaling?.enabled || + spec.stateful || + spec.serverless.enabled || + spec.volumes.length > 0 || + spec.resourceLimits.cpuCores === null || + spec.resourceLimits.memoryMb === null + ) { + skip("active revision is unsupported"); + continue; + } + const current = state.length; + if (spec.placement.replicas !== current) { + skip("active topology does not match revision target"); + continue; + } + const outsideBounds = + current < spec.autoscaling.minReplicas || + current > spec.autoscaling.maxReplicas; + const metrics = outsideBounds + ? undefined + : await queryAutoscalingMetrics({ + serviceId: service.id, + deploymentIds: state.map((item) => item.deploymentId), + cpuLimitCores: spec.resourceLimits.cpuCores, + memoryLimitMb: spec.resourceLimits.memoryMb, + now, + }); + const recommendation = calculateAutoscalingRecommendation({ + currentReplicas: current, + minReplicas: spec.autoscaling.minReplicas, + maxReplicas: spec.autoscaling.maxReplicas, + metrics, + }); + if (recommendation.status === "hold") { + skip(`recommendation held (${recommendation.reason})`); + continue; + } + const result = await cloneActiveRevisionForAutoscaling({ + serviceId: service.id, + expectedRevisionId: first.revisionId, + expectedDeploymentIds: state.map((item) => item.deploymentId), + expectedDeploymentCount: state.length, + expectedMinReplicas: spec.autoscaling.minReplicas, + expectedMaxReplicas: spec.autoscaling.maxReplicas, + targetReplicas: recommendation.targetReplicas, + now, + }); + if (!result.created) { + skip(`recommendation discarded (${result.reason})`); + continue; + } + created++; + await sendRolloutCreated(result.rolloutId, service.id); + } catch (error) { + console.error(`[autoscaling] failed to evaluate ${service.name}`, error); + } + } + return created; +} export async function rebalanceAutomaticServices( maxCreated = MAX_REBALANCES_PER_RUN, diff --git a/web/lib/service-config.ts b/web/lib/service-config.ts index f7f30dd8..16438957 100644 --- a/web/lib/service-config.ts +++ b/web/lib/service-config.ts @@ -1,5 +1,6 @@ import { getServiceRevisionTotalReplicas, + type ServiceAutoscalingPolicy, type ServiceRevisionSpec, } from "@/lib/service-revision-spec"; @@ -55,6 +56,7 @@ export type ResourceLimitsConfig = { export type PlacementConfig = { mode?: "manual" | "automatic"; replicas: number; + autoscaling?: ServiceAutoscalingPolicy; }; export type ServerlessConfig = { @@ -128,6 +130,9 @@ export function buildCurrentConfig( resourceMemoryLimitMb: number | null; replicas: number; placementMode?: "manual" | "automatic" | null; + autoscalingEnabled?: boolean | null; + autoscalingMinReplicas?: number | null; + autoscalingMaxReplicas?: number | null; stateful?: boolean | null; serverlessEnabled?: boolean | null; serverlessSleepAfterSeconds?: number | null; @@ -161,6 +166,13 @@ export function buildCurrentConfig( placement: { mode: placementMode, replicas: replicaCount, + autoscaling: service.autoscalingEnabled + ? { + enabled: true, + minReplicas: service.autoscalingMinReplicas ?? 1, + maxReplicas: service.autoscalingMaxReplicas ?? 1, + } + : undefined, }, replicas: replicas.map((r) => ({ serverId: r.serverId, @@ -411,6 +423,21 @@ export function diffConfigs( deployedPlacementMode === "automatic" && currentPlacementMode === "automatic" ) { + const deployedAutoscaling = deployed.placement?.autoscaling; + const currentAutoscaling = current.placement?.autoscaling; + if ( + JSON.stringify(deployedAutoscaling) !== JSON.stringify(currentAutoscaling) + ) { + const describe = (policy: ServiceAutoscalingPolicy | undefined) => + policy?.enabled + ? `${policy.minReplicas}-${policy.maxReplicas}` + : "Disabled"; + changes.push({ + field: "Autoscaling", + from: describe(deployedAutoscaling), + to: describe(currentAutoscaling), + }); + } if (deployed.placement?.replicas !== current.placement?.replicas) { changes.push({ field: "Desired replicas", @@ -706,6 +733,7 @@ export function revisionSpecToDeployedConfig( placement: { mode: specification.placement.mode, replicas: replicaCount, + autoscaling: specification.autoscaling, }, replicas: specification.placements.map((placement) => ({ serverId: placement.serverId, diff --git a/web/lib/service-revision-changes.ts b/web/lib/service-revision-changes.ts index cdfa0b31..52a6fb0a 100644 --- a/web/lib/service-revision-changes.ts +++ b/web/lib/service-revision-changes.ts @@ -49,6 +49,16 @@ const serviceRevisionSpecFields = { cpuCores: z.number().nullable(), memoryMb: z.number().nullable(), }), + autoscaling: z + .discriminatedUnion("enabled", [ + z.strictObject({ enabled: z.literal(false) }), + z.strictObject({ + enabled: z.literal(true), + minReplicas: z.number().int().min(1).max(32), + maxReplicas: z.number().int().min(1).max(32), + }), + ]) + .optional(), placements: z.array( z.strictObject({ serverId: z.string(), count: z.number() }), ), @@ -106,6 +116,34 @@ const serviceRevisionSpecSchema = z code: "custom", message: "Stateful services cannot use automatic placement", }); + if (spec.autoscaling?.enabled) { + if (spec.autoscaling.minReplicas > spec.autoscaling.maxReplicas) + context.addIssue({ + code: "custom", + message: "Autoscaling minimum cannot exceed maximum", + }); + if ( + spec.placement.mode !== "automatic" || + spec.stateful || + spec.serverless.enabled || + spec.volumes.length > 0 || + spec.resourceLimits.cpuCores === null || + spec.resourceLimits.memoryMb === null + ) + context.addIssue({ + code: "custom", + message: "Revision is not eligible for autoscaling", + }); + if ( + spec.placement.mode === "automatic" && + (spec.placement.replicas < spec.autoscaling.minReplicas || + spec.placement.replicas > spec.autoscaling.maxReplicas) + ) + context.addIssue({ + code: "custom", + message: "Concrete replica target must be within autoscaling bounds", + }); + } }); export type ServiceRevisionChange = { @@ -303,6 +341,17 @@ export function diffServiceRevisionSpecs( previous.placement.mode === "automatic" ? "Automatic" : "Manual", current.placement.mode === "automatic" ? "Automatic" : "Manual", ); + const autoscalingDescription = ( + policy: ServiceRevisionSpec["autoscaling"], + ) => + policy?.enabled + ? `Enabled (${policy.minReplicas}-${policy.maxReplicas} replicas)` + : "Disabled"; + add( + "Autoscaling", + autoscalingDescription(previous.autoscaling), + autoscalingDescription(current.autoscaling), + ); if ( previous.placement.mode === "automatic" && current.placement.mode === "automatic" diff --git a/web/lib/service-revision-spec.ts b/web/lib/service-revision-spec.ts index 315784b1..343dd508 100644 --- a/web/lib/service-revision-spec.ts +++ b/web/lib/service-revision-spec.ts @@ -35,6 +35,10 @@ export type ServiceRevisionPlacementIntent = | { mode: "manual" } | { mode: "automatic"; replicas: number }; +export type ServiceAutoscalingPolicy = + | { enabled: false } + | { enabled: true; minReplicas: number; maxReplicas: number }; + export type ServiceRevisionPort = { containerPort: number; isPublic: boolean; @@ -167,6 +171,7 @@ export type ServiceRevisionSpec = { memoryMb: number | null; }; placement: ServiceRevisionPlacementIntent; + autoscaling?: ServiceAutoscalingPolicy; placements: ServiceRevisionPlacement[]; ports: ServiceRevisionPort[]; secrets: ServiceRevisionSecret[]; @@ -193,6 +198,9 @@ export type ServiceRevisionDraft = { resourceMemoryLimitMb: number | null; placementMode?: "manual" | "automatic" | null; replicas?: number; + autoscalingEnabled?: boolean | null; + autoscalingMinReplicas?: number | null; + autoscalingMaxReplicas?: number | null; }; placements: Array<{ serverId: string; count: number }>; ports: Array<{ @@ -241,6 +249,33 @@ function validateServiceRevisionSpec( if (totalReplicas > 32) { throw new Error("Maximum 32 replicas allowed"); } + if (specification.autoscaling?.enabled) { + const { minReplicas, maxReplicas } = specification.autoscaling; + if ( + !Number.isInteger(minReplicas) || + !Number.isInteger(maxReplicas) || + minReplicas < 1 || + maxReplicas > 32 || + minReplicas > maxReplicas + ) { + throw new Error( + "Autoscaling range must satisfy 1 <= minimum <= maximum <= 32", + ); + } + if (specification.placement.mode !== "automatic") + throw new Error("Autoscaling requires automatic placement"); + if (specification.stateful) + throw new Error("Autoscaling is not supported for stateful services"); + if (specification.serverless.enabled) + throw new Error("Autoscaling is not supported for serverless services"); + if (specification.volumes.length > 0) + throw new Error("Autoscaling is not supported for services with volumes"); + if ( + specification.resourceLimits.cpuCores === null || + specification.resourceLimits.memoryMb === null + ) + throw new Error("Autoscaling requires both CPU and memory limits"); + } if ( specification.placement.mode === "automatic" && specification.placements.length @@ -285,6 +320,19 @@ export function buildServiceRevisionSpec( ): ServiceRevisionSpec { const { service } = draft; const image = overrides.image?.trim() || service.image.trim(); + const autoscaling = service.autoscalingEnabled + ? { + enabled: true as const, + minReplicas: service.autoscalingMinReplicas ?? 1, + maxReplicas: service.autoscalingMaxReplicas ?? 1, + } + : undefined; + const replicas = autoscaling + ? Math.min( + autoscaling.maxReplicas, + Math.max(autoscaling.minReplicas, service.replicas ?? 1), + ) + : (service.replicas ?? 1); const specification: ServiceRevisionSpec = { schemaVersion: SERVICE_REVISION_SCHEMA_VERSION, @@ -318,8 +366,9 @@ export function buildServiceRevisionSpec( }, placement: service.placementMode === "automatic" - ? { mode: "automatic", replicas: service.replicas ?? 1 } + ? { mode: "automatic", replicas } : { mode: "manual" }, + autoscaling, placements: (service.placementMode === "automatic" ? [] : draft.placements) .filter((placement) => placement.count > 0) .map((placement) => ({ diff --git a/web/lib/service-revisions.ts b/web/lib/service-revisions.ts index bf670261..4af8aba7 100644 --- a/web/lib/service-revisions.ts +++ b/web/lib/service-revisions.ts @@ -12,6 +12,7 @@ import { services, serviceVolumes, } from "@/db/schema"; +import { isObservedReady } from "@/lib/deployment-status"; import { resolvePersistedSourceFromRows } from "@/lib/public-api"; import type { ServiceRevisionActor } from "@/lib/service-revision-actor"; import { parseServiceRevisionSpec } from "@/lib/service-revision-changes"; @@ -433,6 +434,150 @@ export async function cloneActiveRevisionAndQueueSystemRollout( }); } +export const AUTOSCALE_ATTEMPT_COOLDOWN_MS = 10 * 60 * 1000; + +/** Compare the sampled active fleet under the service lock before scaling it. */ +export async function cloneActiveRevisionForAutoscaling(input: { + serviceId: string; + expectedRevisionId: string; + expectedDeploymentIds: string[]; + expectedDeploymentCount: number; + expectedMinReplicas: number; + expectedMaxReplicas: number; + targetReplicas: number; + now?: Date; +}) { + return db.transaction(async (tx) => { + await tx.execute( + sql`select pg_advisory_xact_lock(hashtext(${input.serviceId}))`, + ); + const service = await tx + .select({ + id: services.id, + deletedAt: services.deletedAt, + migrationStatus: services.migrationStatus, + lastAutoscaleAttemptAt: services.lastAutoscaleAttemptAt, + }) + .from(services) + .where(eq(services.id, input.serviceId)) + .then((rows) => rows[0]); + if (!service || service.deletedAt || service.migrationStatus) + return { created: false, reason: "service-unavailable" } as const; + const now = input.now ?? new Date(); + if ( + service.lastAutoscaleAttemptAt && + now.getTime() - service.lastAutoscaleAttemptAt.getTime() < + AUTOSCALE_ATTEMPT_COOLDOWN_MS + ) + return { created: false, reason: "cooldown" } as const; + const pending = await tx + .select({ id: rollouts.id }) + .from(rollouts) + .where( + and( + eq(rollouts.serviceId, input.serviceId), + inArray(rollouts.status, ["queued", "in_progress"]), + ), + ) + .limit(1) + .then((rows) => rows[0]); + if (pending) return { created: false, reason: "pending-rollout" } as const; + const active = await tx + .select({ + id: deployments.id, + revisionId: deployments.serviceRevisionId, + runtimeDesiredState: deployments.runtimeDesiredState, + observedPhase: deployments.observedPhase, + specification: serviceRevisions.specification, + }) + .from(deployments) + .innerJoin( + serviceRevisions, + eq(serviceRevisions.id, deployments.serviceRevisionId), + ) + .where( + and( + eq(deployments.serviceId, input.serviceId), + eq(deployments.trafficState, "active"), + inArray(deployments.runtimeDesiredState, ["running", "stopped"]), + ), + ) + .orderBy(deployments.id); + const expectedIds = new Set(input.expectedDeploymentIds); + if ( + expectedIds.size !== input.expectedDeploymentCount || + active.length !== expectedIds.size || + active.some( + (deployment) => + !expectedIds.has(deployment.id) || + deployment.revisionId !== input.expectedRevisionId || + deployment.runtimeDesiredState !== "running" || + !isObservedReady(deployment.observedPhase), + ) + ) + return { created: false, reason: "stale-topology" } as const; + const source = active[0]; + if (!source) return { created: false, reason: "stale-topology" } as const; + const specification = parseServiceRevisionSpec(source.specification); + const autoscaling = specification.autoscaling; + const targetDelta = input.targetReplicas - active.length; + const scalesUpWithinPolicy = + autoscaling?.enabled === true && + targetDelta > 0 && + input.targetReplicas <= autoscaling.maxReplicas && + (active.length >= autoscaling.minReplicas || + input.targetReplicas >= autoscaling.minReplicas); + const scalesDownOne = + autoscaling?.enabled === true && + targetDelta === -1 && + active.length <= autoscaling.maxReplicas && + input.targetReplicas >= autoscaling.minReplicas; + const clampsToMaximum = + autoscaling?.enabled === true && + active.length > autoscaling.maxReplicas && + input.targetReplicas === autoscaling.maxReplicas; + if ( + specification.placement.mode !== "automatic" || + !autoscaling?.enabled || + autoscaling.minReplicas !== input.expectedMinReplicas || + autoscaling.maxReplicas !== input.expectedMaxReplicas || + specification.placement.replicas !== active.length || + input.targetReplicas < 1 || + input.targetReplicas > 32 || + (!scalesUpWithinPolicy && !scalesDownOne && !clampsToMaximum) + ) + return { created: false, reason: "stale-policy" } as const; + + // Cool down capacity/preflight failures as well as successful creations. + await tx + .update(services) + .set({ lastAutoscaleAttemptAt: now }) + .where(eq(services.id, input.serviceId)); + const revisionId = randomUUID(); + await tx.insert(serviceRevisions).values({ + id: revisionId, + serviceId: input.serviceId, + specification: { + ...specification, + placement: { + ...specification.placement, + replicas: input.targetReplicas, + }, + }, + actor: { type: "system" }, + }); + const rolloutId = randomUUID(); + await tx.insert(rollouts).values({ + id: rolloutId, + serviceId: input.serviceId, + serviceRevisionId: revisionId, + status: "queued", + currentStage: "queued", + }); + return { created: true, rolloutId, revisionId } as const; + }); +} + export async function createRolloutForServiceRevision( serviceId: string, serviceRevisionId: string, diff --git a/web/lib/victoria-metrics.ts b/web/lib/victoria-metrics.ts index 1b25242b..cbe44b92 100644 --- a/web/lib/victoria-metrics.ts +++ b/web/lib/victoria-metrics.ts @@ -100,7 +100,7 @@ const METRIC_NAMES = { diskUsedBytes: "techulus_node_disk_used_bytes", } as const; -function getQueryEndpoint(): EndpointConfig | undefined { +export function getQueryEndpoint(): EndpointConfig | undefined { const endpoint = process.env.VICTORIA_METRICS_PRIVATE_URL || process.env.VICTORIA_METRICS_URL; @@ -577,7 +577,7 @@ async function queryRangeMetric( })) .filter((point) => Number.isFinite(point.value)); } -async function queryRangePromQL( +export async function queryRangePromQL( endpoint: EndpointConfig, options: { query: string; @@ -744,7 +744,7 @@ function normalizeHTTPStatusFamily(code: string | undefined): string { return `${code.charAt(0)}xx`; } -function escapePromQL(value: string) { +export function escapePromQL(value: string) { return value .replace(/\\/g, "\\\\") .replace(/"/g, '\\"') diff --git a/web/tests/autoscaling.test.ts b/web/tests/autoscaling.test.ts new file mode 100644 index 00000000..a171ec8f --- /dev/null +++ b/web/tests/autoscaling.test.ts @@ -0,0 +1,315 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + type AutoscalingMetricResult, + calculateAutoscalingRecommendation, + queryAutoscalingMetrics, +} from "@/lib/autoscaling"; + +const point = (cpu: number, memory: number) => ({ + timestamp: "2026-08-02T12:00:00.000Z", + cpuUtilizationPercent: cpu, + memoryUtilizationPercent: memory, + coverage: [], +}); +const ready = ( + cpu: number, + memory: number, + count = 6, +): AutoscalingMetricResult => ({ + status: "ready", + points: Array.from({ length: count }, () => point(cpu, memory)), +}); + +describe("calculateAutoscalingRecommendation", () => { + it("lets either resource drive scale-up and requires both for scale-down", () => { + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 4, + minReplicas: 1, + maxReplicas: 10, + metrics: ready(30, 90), + }), + ).toMatchObject({ status: "scale", direction: "up", targetReplicas: 6 }); + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 4, + minReplicas: 1, + maxReplicas: 10, + metrics: ready(30, 60), + }), + ).toEqual({ status: "hold", reason: "stable" }); + }); + + it.each([ + [53.9, "down"], + [54, "hold"], + [66, "hold"], + [66.1, "up"], + ] as const)("applies the inclusive deadband at %s", (utilization, expected) => { + const result = calculateAutoscalingRecommendation({ + currentReplicas: 20, + minReplicas: 1, + maxReplicas: 32, + metrics: ready(utilization, utilization), + }); + expect(result.status === "scale" ? result.direction : result.status).toBe( + expected, + ); + }); + + it("clamps directly to policy bounds without metrics", () => { + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 1, + minReplicas: 4, + maxReplicas: 8, + }), + ).toMatchObject({ targetReplicas: 4, reason: "below-minimum" }); + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 10, + minReplicas: 2, + maxReplicas: 8, + }), + ).toMatchObject({ targetReplicas: 8, reason: "above-maximum" }); + }); + + it("reports when metrics were not queried within policy bounds", () => { + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 4, + minReplicas: 2, + maxReplicas: 8, + }), + ).toEqual({ status: "hold", reason: "metrics-not-queried" }); + }); + + it("clamps direct scale-up recommendations to the maximum", () => { + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 4, + minReplicas: 1, + maxReplicas: 8, + metrics: ready(200, 30), + }), + ).toMatchObject({ status: "scale", direction: "up", targetReplicas: 8 }); + }); + + it("continues scaling down only one replica after stabilization", () => { + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 8, + minReplicas: 1, + maxReplicas: 10, + metrics: ready(30, 30), + }), + ).toMatchObject({ status: "scale", direction: "down", targetReplicas: 7 }); + }); + + it("clamps utilization actions and propagates incomplete coverage", () => { + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 32, + minReplicas: 1, + maxReplicas: 32, + metrics: ready(100, 100), + }), + ).toEqual({ status: "hold", reason: "stable" }); + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 1, + minReplicas: 1, + maxReplicas: 32, + metrics: ready(1, 1), + }), + ).toEqual({ status: "hold", reason: "stable" }); + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 4, + minReplicas: 1, + maxReplicas: 8, + metrics: { status: "hold", reason: "incomplete-coverage" }, + }), + ).toEqual({ status: "hold", reason: "incomplete-coverage" }); + }); + + it("holds low utilization at the minimum replica count", () => { + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 2, + minReplicas: 2, + maxReplicas: 8, + metrics: ready(20, 20), + }), + ).toEqual({ status: "hold", reason: "stable" }); + }); + + it("requires all six points to remain below current for downscale", () => { + const metrics = ready(30, 30); + if (metrics.status === "ready") metrics.points[0] = point(90, 30); + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 4, + minReplicas: 1, + maxReplicas: 8, + metrics, + }), + ).toEqual({ status: "hold", reason: "downscale-stabilizing" }); + expect( + calculateAutoscalingRecommendation({ + currentReplicas: 4, + minReplicas: 1, + maxReplicas: 8, + metrics: ready(30, 30, 5), + }), + ).toEqual({ status: "hold", reason: "downscale-stabilizing" }); + }); +}); + +describe("queryAutoscalingMetrics", () => { + afterEach(() => { + delete process.env.VICTORIA_METRICS_URL; + vi.unstubAllGlobals(); + }); + + it("checks metrics configuration explicitly", async () => { + await expect( + queryAutoscalingMetrics({ + serviceId: "svc", + deploymentIds: ["dep"], + cpuLimitCores: 1, + memoryLimitMb: 100, + }), + ).resolves.toEqual({ status: "hold", reason: "metrics-disabled" }); + }); + + it("returns six normalized points with exact per-deployment coverage", async () => { + process.env.VICTORIA_METRICS_URL = "http://metrics.test"; + const times = Array.from( + { length: 6 }, + (_, index) => 1_754_136_900 + index * 60, + ); + let call = 0; + vi.stubGlobal( + "fetch", + vi.fn(async (input: string | URL) => { + const query = new URL(String(input)).searchParams.get("query") ?? ""; + expect(query).toContain('service_id="svc"'); + expect(query).toContain('deployment_id=~"^(?:dep-a|dep-b)$"'); + const timestampQuery = query.startsWith("tlast_over_time"); + const memoryQuery = query.includes("memory_used_bytes"); + call++; + return new Response( + JSON.stringify({ + status: "success", + data: { + result: ["dep-a", "dep-b"].map((deploymentId) => ({ + metric: { deployment_id: deploymentId }, + values: times.map((time) => [ + time, + String( + timestampQuery + ? time + : memoryQuery + ? 50 * 1024 * 1024 + : 0.5, + ), + ]), + })), + }, + }), + { status: 200 }, + ); + }), + ); + const result = await queryAutoscalingMetrics({ + serviceId: "svc", + deploymentIds: ["dep-b", "dep-a"], + cpuLimitCores: 1, + memoryLimitMb: 100, + now: new Date(times[5] * 1000), + }); + expect(call).toBe(4); + expect(result.status).toBe("ready"); + if (result.status === "ready") { + expect(result.points).toHaveLength(6); + expect(result.points[0]).toMatchObject({ + cpuUtilizationPercent: 50, + memoryUtilizationPercent: 50, + coverage: [{ deploymentId: "dep-a" }, { deploymentId: "dep-b" }], + }); + } + }); + + it.each([ + ["missing series", "incomplete-coverage", []], + ["unexpected series", "unexpected-series", ["other"]], + ] as const)("holds for %s", async (_name, reason, ids) => { + process.env.VICTORIA_METRICS_URL = "http://metrics.test"; + vi.stubGlobal( + "fetch", + vi.fn( + async () => + new Response( + JSON.stringify({ + status: "success", + data: { + result: ids.map((deployment_id) => ({ + metric: { deployment_id }, + values: [], + })), + }, + }), + ), + ), + ); + await expect( + queryAutoscalingMetrics({ + serviceId: "svc", + deploymentIds: ["dep"], + cpuLimitCores: 1, + memoryLimitMb: 100, + now: new Date("2026-08-02T12:00:00Z"), + }), + ).resolves.toEqual({ status: "hold", reason }); + }); + + it("holds coverage after any number of duplicate deployment series", async () => { + process.env.VICTORIA_METRICS_URL = "http://metrics.test"; + const end = Date.parse("2026-08-02T12:00:00Z") / 1000; + const times = Array.from( + { length: 6 }, + (_, index) => end - 300 + index * 60, + ); + vi.stubGlobal( + "fetch", + vi.fn(async (input: string | URL) => { + const query = new URL(String(input)).searchParams.get("query") ?? ""; + const timestampQuery = query.startsWith("tlast_over_time"); + return new Response( + JSON.stringify({ + status: "success", + data: { + result: Array.from({ length: 3 }, () => ({ + metric: { deployment_id: "dep" }, + values: times.map((time) => [ + time, + String(timestampQuery ? time : 1), + ]), + })), + }, + }), + ); + }), + ); + await expect( + queryAutoscalingMetrics({ + serviceId: "svc", + deploymentIds: ["dep"], + cpuLimitCores: 1, + memoryLimitMb: 100, + now: new Date("2026-08-02T12:00:00Z"), + }), + ).resolves.toEqual({ status: "hold", reason: "incomplete-coverage" }); + }); +}); diff --git a/web/tests/inngest-route.test.ts b/web/tests/inngest-route.test.ts index 14a802b1..21326bb5 100644 --- a/web/tests/inngest-route.test.ts +++ b/web/tests/inngest-route.test.ts @@ -7,6 +7,7 @@ type ServeOptions = { const mocks = vi.hoisted(() => { const functions = { agentUpgradeTimeoutCheck: { id: "agent-upgrade-timeout-check" }, + autoscalingCheck: { id: "autoscaling-check" }, backupWorkflow: { id: "backup-workflow" }, buildTriggerWorkflow: { id: "build-trigger-workflow" }, buildWorkflow: { id: "build-workflow" }, diff --git a/web/tests/public-api-configuration.test.ts b/web/tests/public-api-configuration.test.ts index 904c08f3..dfa27584 100644 --- a/web/tests/public-api-configuration.test.ts +++ b/web/tests/public-api-configuration.test.ts @@ -21,7 +21,7 @@ const mocks = vi.hoisted(() => { vi.mock("@/db", () => ({ db: { select: mocks.select } })); -import { safeConfiguration } from "@/lib/public-api"; +import { placementSchema, safeConfiguration } from "@/lib/public-api"; describe("public API configuration state", () => { beforeEach(() => { @@ -120,4 +120,40 @@ describe("public API configuration state", () => { expect(configuration.current.replicas).toBe(4); expect(configuration.current.placements).toEqual([]); }); + + it("serializes autoscaling placement that the replacement schema accepts", async () => { + mocks.rows.push([], [], [], [], []); + const configuration = await safeConfiguration({ + id: "service-1", + name: "Autoscaled", + sourceType: "image", + image: "nginx", + hostname: null, + stateful: false, + placementMode: "automatic", + replicas: 4, + autoscalingEnabled: true, + autoscalingMinReplicas: 2, + autoscalingMaxReplicas: 8, + healthCheckCmd: null, + startCommand: null, + resourceCpuLimit: 1, + resourceMemoryLimitMb: 512, + serverlessEnabled: false, + serverlessSleepAfterSeconds: 300, + serverlessWakeTimeoutSeconds: 300, + deploymentSchedule: null, + backupEnabled: false, + backupSchedule: null, + } as never); + + expect(configuration.current.placement).toEqual({ + mode: "automatic", + replicas: 4, + autoscaling: { enabled: true, minReplicas: 2, maxReplicas: 8 }, + }); + expect( + placementSchema.safeParse(configuration.current.placement).success, + ).toBe(true); + }); }); diff --git a/web/tests/public-api-plan.test.ts b/web/tests/public-api-plan.test.ts index 5a44f29d..c9813780 100644 --- a/web/tests/public-api-plan.test.ts +++ b/web/tests/public-api-plan.test.ts @@ -67,6 +67,45 @@ describe("configuration plan protocol", () => { expect(result.changes).toEqual([]); }); + it("treats an echoed autoscaling placement as a no-op", () => { + const current = { + name: "web", + source: { type: "image" as const, image: "nginx" }, + hostname: "web", + ports: [], + placement: { + mode: "automatic" as const, + autoscaling: { minReplicas: 2, maxReplicas: 8 }, + }, + healthCheck: null, + startCommand: null, + resources: { cpuCores: 1, memoryMb: 512 }, + serverless: { enabled: false }, + }; + const result = planCanonicalConfiguration(current, { + name: current.name, + source: current.source, + hostname: current.hostname, + ports: current.ports, + placement: { + mode: "automatic", + replicas: 4, + autoscaling: { + enabled: true, + minReplicas: 2, + maxReplicas: 8, + }, + }, + healthCheck: current.healthCheck, + startCommand: current.startCommand, + resources: current.resources, + }); + + expect(result.action).toBe("noop"); + expect(result.changes).toEqual([]); + expect(result.desiredVersion).toBe(result.currentVersion); + }); + it("sorts ports and manual placements deterministically", () => { const desired = canonicalDesired({ name: "web", diff --git a/web/tests/public-api-source.test.ts b/web/tests/public-api-source.test.ts index 22f97aea..a0520b8b 100644 --- a/web/tests/public-api-source.test.ts +++ b/web/tests/public-api-source.test.ts @@ -155,6 +155,12 @@ describe("public API placement schema", () => { it.each([ { mode: "automatic", replicas: 32 }, + { mode: "automatic", autoscaling: { minReplicas: 2, maxReplicas: 8 } }, + { + mode: "automatic", + replicas: 4, + autoscaling: { enabled: true, minReplicas: 2, maxReplicas: 8 }, + }, { mode: "automatic", replicas: 3 }, { mode: "manual", placements: [{ serverId: "server-1", count: 32 }] }, { mode: "manual", placements: [{ serverId: "server-1", count: 2 }] }, @@ -168,6 +174,8 @@ describe("public API placement schema", () => { it.each([ { mode: "automatic", replicas: 0 }, { mode: "automatic", replicas: 33 }, + { mode: "automatic", autoscaling: { minReplicas: 9, maxReplicas: 8 } }, + { mode: "automatic", autoscaling: { minReplicas: 0, maxReplicas: 8 } }, { mode: "manual", placements: [] }, { mode: "manual", diff --git a/web/tests/service-config.test.ts b/web/tests/service-config.test.ts index db76a47b..ef0c9d33 100644 --- a/web/tests/service-config.test.ts +++ b/web/tests/service-config.test.ts @@ -301,6 +301,25 @@ describe("service config", () => { }); }); + it("reports autoscaling policy changes as pending config", () => { + const fixed = deployedConfig({ + placement: { mode: "automatic", replicas: 2 }, + }); + const autoscaled = deployedConfig({ + placement: { + mode: "automatic", + replicas: 2, + autoscaling: { enabled: true, minReplicas: 2, maxReplicas: 8 }, + }, + }); + + expect(diffConfigs(fixed, autoscaled)).toContainEqual({ + field: "Autoscaling", + from: "Disabled", + to: "2-8", + }); + }); + it("requires a build when the Dockerfile path is added, updated, or removed", () => { const dockerfileSecret = { key: TECHULUS_DOCKERFILE_PATH, diff --git a/web/tests/service-revision-changes.test.ts b/web/tests/service-revision-changes.test.ts index 4839a784..d28c7776 100644 --- a/web/tests/service-revision-changes.test.ts +++ b/web/tests/service-revision-changes.test.ts @@ -147,6 +147,25 @@ describe("diffServiceRevisionSpecs", () => { ]); }); + it("reports autoscaling policy changes", () => { + const previous = spec(); + previous.placement = { mode: "automatic", replicas: 2 }; + previous.placements = []; + previous.volumes = []; + previous.resourceLimits = { cpuCores: 1, memoryMb: 512 }; + const current = structuredClone(previous); + current.autoscaling = { enabled: true, minReplicas: 2, maxReplicas: 8 }; + + expect(diffServiceRevisionSpecs(previous, current)).toContainEqual({ + field: "Autoscaling", + from: "Disabled", + to: "Enabled (2-8 replicas)", + }); + expect(parseServiceRevisionSpec(current).autoscaling).toEqual( + current.autoscaling, + ); + }); + it("accepts persisted automatic serverless revisions", () => { const automatic = spec(); automatic.placement = { mode: "automatic", replicas: 1 }; diff --git a/web/tests/service-revision-spec.test.ts b/web/tests/service-revision-spec.test.ts index b8e1755c..1bae1205 100644 --- a/web/tests/service-revision-spec.test.ts +++ b/web/tests/service-revision-spec.test.ts @@ -258,6 +258,38 @@ describe("service revision specification", () => { }); }); + it("snapshots an eligible autoscaling policy and clamps its concrete target", () => { + const input = draft({ volumes: [] }); + Object.assign(input.service, { + placementMode: "automatic", + replicas: 20, + autoscalingEnabled: true, + autoscalingMinReplicas: 2, + autoscalingMaxReplicas: 8, + resourceCpuLimit: 1, + resourceMemoryLimitMb: 512, + }); + + expect(buildServiceRevisionSpec(input)).toMatchObject({ + placement: { mode: "automatic", replicas: 8 }, + autoscaling: { enabled: true, minReplicas: 2, maxReplicas: 8 }, + }); + }); + + it("rejects ineligible autoscaling policies", () => { + const input = draft({ volumes: [] }); + Object.assign(input.service, { + placementMode: "automatic", + replicas: 2, + autoscalingEnabled: true, + autoscalingMinReplicas: 2, + autoscalingMaxReplicas: 8, + }); + expect(() => buildServiceRevisionSpec(input)).toThrow( + "Autoscaling requires both CPU and memory limits", + ); + }); + it("rejects more than 32 automatic replicas", () => { const input = draft({ volumes: [] }); input.service.placementMode = "automatic";