fix(metrics): drop stale series when an edit or reload replaces a supervisor

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
 

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
 

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)
+	}
+}

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.

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) {

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)
+	}
+}

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