diff --git a/CHANGELOG.md b/CHANGELOG.md index f21f3ce..5c428fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 including history kept for instances that exited on their own. A reload cleans up only once it commits: a reload that fails and rolls back brings the process back with its counters intact. +- **Lowering `scale` through a reload or an API edit drops the old + instances' series.** Both paths replace the process's supervisor, and the + cleanup only ran for a live scale-down, so after editing `scale: 10` to + `scale: 2` instances 2-9 stayed at `cbox_init_process_up` 0 (with their + last CPU and memory readings) forever. The manager now keeps only the + series of the instances the new supervisor runs. +- **Turning a process into a scheduled task drops its series.** A scheduled + task exports no process metrics, but the supervisor it replaced left + `process_up` at 0, `desired_scale` at its old value and the last + health check status behind, so `process_up == 0`, scale-drift and health + alerts fired for good. ### Documentation diff --git a/docs/observability/metrics.md b/docs/observability/metrics.md index 77fdeae..cbbcbb7 100644 --- a/docs/observability/metrics.md +++ b/docs/observability/metrics.md @@ -155,7 +155,13 @@ scrape target's `instance` is dropped for these series. `_last_exit_code`, and every resource metric), so `cbox_init_process_up == 0` does not keep firing for an instance that was scaled away on purpose. The process-level series (`restarts_total`, - `desired_scale`, `current_scale`, health checks) stay. + `desired_scale`, `current_scale`, health checks) stay. This applies to a + live scale-down and to a config reload or API edit that lowers `scale`. +- A process **removed from the config** (through the API or a reload) has + every one of its series deleted. So does a process that an edit or reload + turns into a **scheduled** task: scheduled tasks export no process + metrics (see [Scheduled Tasks](../features/scheduled-tasks)). A reload + that fails and rolls back keeps the series, because the process comes back. ### Process Lifecycle Metrics diff --git a/internal/metrics/cleanup_test.go b/internal/metrics/cleanup_test.go index 52876ae..80da4ba 100644 --- a/internal/metrics/cleanup_test.go +++ b/internal/metrics/cleanup_test.go @@ -222,3 +222,55 @@ func TestRemoveProcessMetrics_LeavesNoSeries(t *testing.T) { } } } + +// TestRetainInstances covers the cleanup after a supervisor is replaced by one +// running fewer instances: every instance series of the process outside keep +// goes, under either label key, and nothing of any other process is touched, +// including one whose name starts with this one's. +func TestRetainInstances(t *testing.T) { + const proc = "retain-fpm" + const other = "retain-fpm-worker" + + for _, id := range []string{proc + "-0", proc + "-1", proc + "-2"} { + populateInstance(proc, id) + } + populateProcess(proc) + populateInstance(other, other+"-0") + + otherBefore := len(seriesWith(t, prometheus.Labels{labelName: other})) + + len(seriesWith(t, prometheus.Labels{labelProcess: other})) + + RetainInstances(proc, []string{proc + "-0"}) + + for _, id := range []string{proc + "-1", proc + "-2"} { + for _, sel := range instanceSelectors(proc, id) { + if left := seriesWith(t, sel); len(left) > 0 { + t.Errorf("instance %s is not in keep but survived: %v", id, left) + } + } + } + kept := 0 + for _, sel := range instanceSelectors(proc, proc+"-0") { + kept += len(seriesWith(t, sel)) + } + if want := len(instanceSeries) + 1; kept != want { + t.Errorf("kept instance has %d series, want %d", kept, want) + } + if left := seriesWith(t, prometheus.Labels{labelName: proc, "reason": "crash"}); len(left) != 1 { + t.Errorf("per-process series must survive RetainInstances, got %v", left) + } + otherAfter := len(seriesWith(t, prometheus.Labels{labelName: other})) + + len(seriesWith(t, prometheus.Labels{labelProcess: other})) + if otherAfter != otherBefore { + t.Errorf("%q: %d series after retaining %q's instances, want %d (over-match)", other, otherAfter, proc, otherBefore) + } + + // An empty keep drops every instance series but leaves the per-process ones. + RetainInstances(proc, nil) + if left := seriesWith(t, prometheus.Labels{labelInstance: proc + "-0"}); len(left) > 0 { + t.Errorf("empty keep left instance series %v", left) + } + if left := seriesWith(t, prometheus.Labels{labelName: proc, "reason": "crash"}); len(left) != 1 { + t.Errorf("empty keep removed per-process series, got %v", left) + } +} diff --git a/internal/metrics/collector.go b/internal/metrics/collector.go index 56ad489..856b2d2 100644 --- a/internal/metrics/collector.go +++ b/internal/metrics/collector.go @@ -1,9 +1,11 @@ package metrics import ( + "runtime" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" - "runtime" + dto "github.com/prometheus/client_model/go" ) // Adding a metric labelled per process or per instance? Add it to @@ -358,6 +360,70 @@ func RemoveInstanceMetrics(processName, instanceID string) { } } +// RetainInstances drops the per-instance series of every instance of +// processName that is not in keep. +// +// RemoveInstanceMetrics needs to know which instances went away. When a +// supervisor is replaced (a reload or an API edit restarts the process with a +// new definition), the manager only knows the instances the new supervisor +// runs; the old supervisor's list is gone. Lowering the scale that way left +// the higher instances at process_up 0 forever. This reads the instance IDs +// from the series themselves, so it drops whatever the new supervisor does not +// run, however it came to exist. +func RetainInstances(processName string, keep []string) { + kept := make(map[string]bool, len(keep)) + for _, id := range keep { + kept[id] = true + } + for _, f := range instanceSeries { + for _, id := range instanceIDs(f, processName) { + if !kept[id] { + f.vec.DeletePartialMatch(prometheus.Labels{ + f.processLabel: processName, + labelInstance: id, + }) + } + } + } +} + +// instanceIDs returns the distinct instance label values that processName +// currently has in the family f. +func instanceIDs(f seriesFamily, processName string) []string { + c, ok := f.vec.(prometheus.Collector) + if !ok { + return nil + } + ch := make(chan prometheus.Metric) + go func() { + c.Collect(ch) + close(ch) + }() + + seen := make(map[string]bool) + var ids []string + for m := range ch { + var pb dto.Metric + if err := m.Write(&pb); err != nil { + continue + } + var proc, inst string + for _, lp := range pb.GetLabel() { + switch lp.GetName() { + case f.processLabel: + proc = lp.GetValue() + case labelInstance: + inst = lp.GetValue() + } + } + if proc == processName && inst != "" && !seen[inst] { + seen[inst] = true + ids = append(ids, inst) + } + } + return ids +} + // RemoveProcessMetrics drops every series belonging to a process that has been // removed from the config entirely, including the process-level ones that carry // no instance label. diff --git a/internal/metrics/resource.go b/internal/metrics/resource.go index cfa05a2..24790ef 100644 --- a/internal/metrics/resource.go +++ b/internal/metrics/resource.go @@ -229,6 +229,24 @@ func (rc *ResourceCollector) RemoveProcess(processName string) { } } +// RetainInstances drops the buffers and handles of every instance of +// processName that is not in keep: the counterpart of metrics.RetainInstances +// for a process whose supervisor was replaced by one running fewer instances. +func (rc *ResourceCollector) RetainInstances(processName string, keep []string) { + rc.mu.Lock() + defer rc.mu.Unlock() + + kept := make(map[string]bool, len(keep)) + for _, id := range keep { + kept[id] = true + } + for instanceID := range rc.instances[processName] { + if !kept[instanceID] { + rc.removeLocked(processName, instanceID) + } + } +} + // track records that instanceID of processName holds a buffer or handle. // Callers hold rc.mu. func (rc *ResourceCollector) track(processName, instanceID string) { diff --git a/internal/metrics/resource_test.go b/internal/metrics/resource_test.go index 4a8de44..5961787 100644 --- a/internal/metrics/resource_test.go +++ b/internal/metrics/resource_test.go @@ -805,3 +805,32 @@ func TestResourceCollector_RemoveProcess(t *testing.T) { t.Errorf("%d buffers left after no-op removals, want 1", got) } } + +// TestResourceCollector_RetainInstances drops the buffers of instances outside +// keep and leaves both the kept instance and other processes alone. +func TestResourceCollector_RetainInstances(t *testing.T) { + rc := NewResourceCollector(time.Second, 10, slog.New(slog.NewTextHandler(os.Stderr, nil))) + + sample := ResourceSample{Timestamp: time.Now(), CPUPercent: 1} + for _, id := range []string{"php-fpm-0", "php-fpm-1", "php-fpm-2"} { + rc.AddSample("php-fpm", id, sample) + } + rc.AddSample("php-fpm-worker", "php-fpm-worker-1", sample) + + rc.RetainInstances("php-fpm", []string{"php-fpm-0"}) + + if _, ok := rc.GetLatest("php-fpm", "php-fpm-0"); !ok { + t.Error("kept instance lost its buffer") + } + for _, id := range []string{"php-fpm-1", "php-fpm-2"} { + if _, ok := rc.GetLatest("php-fpm", id); ok { + t.Errorf("buffer for %s survived though it is not in keep", id) + } + } + if _, ok := rc.GetLatest("php-fpm-worker", "php-fpm-worker-1"); !ok { + t.Error("another process's buffer was removed") + } + if got := len(rc.GetBufferSizes()); got != 2 { + t.Errorf("%d buffers left, want 2", got) + } +} diff --git a/internal/process/manager_config.go b/internal/process/manager_config.go index 7632bf3..fd7d4e4 100644 --- a/internal/process/manager_config.go +++ b/internal/process/manager_config.go @@ -186,6 +186,10 @@ func (m *Manager) updateProcessLocked(ctx context.Context, name string, procCfg if stats := m.scheduler.Stats(); stats.TotalJobs > 0 && !stats.Started { m.scheduler.Start() } + // A scheduled process exports no process metrics, so whatever its old + // supervisor left (process_up at 0, desired_scale, health checks) is + // stale from here on. + m.forgetProcessTelemetry(name) m.logger.Info("Scheduled process updated", "name", name, "schedule", procCfg.Schedule) m.auditLogger.LogProcessUpdated(name, procCfg.Command, procCfg.Scale) @@ -242,6 +246,7 @@ func (m *Manager) updateProcessLocked(ctx context.Context, name string, procCfg } m.processes[name] = newSupervisor + m.retainSupervisorTelemetry(name, newSupervisor) m.logger.Info("Process updated and restarted", "name", name) } else { // New config is disabled, just remove from running processes @@ -421,6 +426,16 @@ func (m *Manager) ReloadConfig(ctx context.Context) error { for _, name := range toStop { m.forgetProcessTelemetry(name) } + // Changed processes run on a new supervisor now. One that became scheduled + // exports no process metrics at all; one that runs fewer instances than + // before must not keep the old higher instances' series. + for _, name := range toUpdate { + if procCfg := newCfg.Processes[name]; procCfg != nil && procCfg.Enabled && procCfg.Schedule != "" { + m.forgetProcessTelemetry(name) + } else if sup, ok := m.processes[name]; ok { + m.retainSupervisorTelemetry(name, sup) + } + } m.logger.Info("Configuration reloaded successfully") @@ -443,6 +458,27 @@ func (m *Manager) forgetProcessTelemetry(name string) { } } +// retainSupervisorTelemetry drops the per-instance series and resource buffers +// of every instance of name that sup does not run. Call it after sup replaced +// an earlier supervisor: the earlier one's instance list is gone, and lowering +// the scale that way would otherwise leave its higher instances at process_up +// 0 forever. A supervisor that runs no instances (defined but not started) +// changes nothing, because a stopped process keeps its series. +func (m *Manager) retainSupervisorTelemetry(name string, sup *Supervisor) { + infos := sup.GetInstances() + if len(infos) == 0 { + return + } + ids := make([]string, 0, len(infos)) + for _, info := range infos { + ids = append(ids, info.ID) + } + metrics.RetainInstances(name, ids) + if m.resourceCollector != nil { + m.resourceCollector.RetainInstances(name, ids) + } +} + // rollbackReload restores the previous configuration after a reload failed // partway through. It stops whatever the failed reload left running for the // touched processes, then restores the old config and restarts the processes diff --git a/internal/process/update_metrics_test.go b/internal/process/update_metrics_test.go new file mode 100644 index 0000000..d59b6b4 --- /dev/null +++ b/internal/process/update_metrics_test.go @@ -0,0 +1,184 @@ +package process + +import ( + "context" + "os" + "strconv" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + + "github.com/cboxdk/init/internal/config" +) + +// seriesByInstance returns every exposed cbox_init_ series that carries the +// given instance label value. +func seriesByInstance(t *testing.T, instanceID string) []string { + t.Helper() + families, err := prometheus.DefaultGatherer.Gather() + if err != nil { + t.Fatalf("gather: %v", err) + } + var found []string + for _, mf := range families { + if !strings.HasPrefix(mf.GetName(), "cbox_init_") { + continue + } + for _, m := range mf.GetMetric() { + for _, lp := range m.GetLabel() { + if lp.GetName() == "instance" && lp.GetValue() == instanceID { + found = append(found, mf.GetName()) + } + } + } + } + return found +} + +// telemetryConfigScaled is telemetryConfig with an explicit scale and command, +// so a test can change either and have the process count as updated. +func telemetryConfigScaled(name string, scale int, sleep string) string { + cfg := telemetryConfig(name) + cfg = strings.Replace(cfg, "scale: 2", "scale: "+strconv.Itoa(scale), 1) + return strings.Replace(cfg, `["sleep", "60"]`, `["sleep", "`+sleep+`"]`, 1) +} + +// assertInstancesGone checks that the instances scaled away by an update left +// no series, while the ones still running keep theirs. +func assertInstancesGone(t *testing.T, gone, kept []string) { + t.Helper() + for _, id := range gone { + if left := seriesByInstance(t, id); len(left) > 0 { + t.Errorf("instance %s no longer exists but still exposes %v", id, left) + } + } + for _, id := range kept { + if len(seriesByInstance(t, id)) == 0 { + t.Errorf("running instance %s has no series", id) + } + } +} + +// waitForInstanceSeries waits until the instance has published series. +func waitForInstanceSeries(t *testing.T, ids ...string) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + ready := true + for _, id := range ids { + ready = ready && len(seriesByInstance(t, id)) > 0 + } + if ready { + return + } + if time.Now().After(deadline) { + t.Fatalf("instances %v never published series", ids) + } + time.Sleep(50 * time.Millisecond) + } +} + +// TestManager_ReloadLowerScaleDropsInstanceSeries: a reload that changes a +// process and lowers its scale replaces its supervisor. The old supervisor's +// higher instances are never started again, so their series must go, the +// same as after a live scale-down. +func TestManager_ReloadLowerScaleDropsInstanceSeries(t *testing.T) { + const name = "updreload" + m, path := startTelemetryManager(t, name) + + if err := os.WriteFile(path, []byte(telemetryConfigScaled(name, 1, "61")), 0600); err != nil { + t.Fatalf("write config: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := m.ReloadConfig(ctx); err != nil { + t.Fatalf("ReloadConfig: %v", err) + } + waitForInstanceSeries(t, name+"-0") + + assertInstancesGone(t, []string{name + "-1"}, []string{name + "-0"}) +} + +// TestManager_UpdateLowerScaleDropsInstanceSeries: the same through +// UpdateProcess (API/TUI edit). +func TestManager_UpdateLowerScaleDropsInstanceSeries(t *testing.T) { + const name = "updapi" + m, path := startTelemetryManager(t, name) + + cfg, err := config.LoadWithEnvExpansion(writeTemp(t, path, telemetryConfigScaled(name, 1, "61"))) + if err != nil { + t.Fatalf("load updated config: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := m.UpdateProcess(ctx, name, cfg.Processes[name]); err != nil { + t.Fatalf("UpdateProcess: %v", err) + } + waitForInstanceSeries(t, name+"-0") + + assertInstancesGone(t, []string{name + "-1"}, []string{name + "-0"}) +} + +// writeTemp writes content next to path and returns the new file's path. +func writeTemp(t *testing.T, path, content string) string { + t.Helper() + p := path + ".updated.yaml" + if err := os.WriteFile(p, []byte(content), 0600); err != nil { + t.Fatalf("write %s: %v", p, err) + } + return p +} + +// scheduledConfig turns the process into a cron job. +func scheduledConfig(name string) string { + return "version: \"1.0\"\n" + + "global:\n shutdown_timeout: 5\n log_level: error\n" + + " resource_metrics_enabled: true\n resource_metrics_interval: 1\n" + + "processes:\n" + + " " + name + ":\n enabled: true\n type: oneshot\n" + + " command: [\"true\"]\n restart: never\n schedule: \"0 3 * * *\"\n" +} + +// TestManager_ReloadToScheduledDropsSeries: a reload that turns a longrun +// process into a scheduled one drops its supervisor. A scheduled process +// exports no process metrics, so every series the supervisor left (process_up +// at 0, desired_scale, health checks) is stale and must go. +func TestManager_ReloadToScheduledDropsSeries(t *testing.T) { + const name = "updsched" + m, path := startTelemetryManager(t, name) + + if err := os.WriteFile(path, []byte(scheduledConfig(name)), 0600); err != nil { + t.Fatalf("write config: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := m.ReloadConfig(ctx); err != nil { + t.Fatalf("ReloadConfig: %v", err) + } + + if left := processSeriesLeft(t, name); len(left) > 0 { + t.Errorf("process %q is now scheduled but still exposes %v", name, left) + } +} + +// TestManager_UpdateToScheduledDropsSeries: the same through UpdateProcess. +func TestManager_UpdateToScheduledDropsSeries(t *testing.T) { + const name = "updapisched" + m, path := startTelemetryManager(t, name) + + cfg, err := config.LoadWithEnvExpansion(writeTemp(t, path, scheduledConfig(name))) + if err != nil { + t.Fatalf("load updated config: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := m.UpdateProcess(ctx, name, cfg.Processes[name]); err != nil { + t.Fatalf("UpdateProcess: %v", err) + } + + if left := processSeriesLeft(t, name); len(left) > 0 { + t.Errorf("process %q is now scheduled but still exposes %v", name, left) + } +}