admin: count plugin-runtime workers in worker metrics (#10884)

* admin: count plugin-runtime workers in worker metrics

The admin server keeps two worker registries: the legacy maintenance-worker
map, filled by workers registering over the worker gRPC stream, and the plugin
worker registry, filled by workers started as `weed worker`. Both the
SeaweedFS_admin_workers_connected / SeaweedFS_admin_worker_slots gauges and the
dashboard's Workers card read only the legacy map, so a cluster that runs the
admin and its workers as separate components reported 0 workers even while its
workers showed up on the plugin pages and ran scheduled jobs.

Aggregate both registries instead. The two are merged by worker ID: `weed mini`
starts both runtimes out of one working directory, so they share the persisted
worker ID and must not be counted twice. For such a worker the slot numbers
still come from the legacy registry, which keeps mini's existing readings.
Plugin workers report their slots in the heartbeat, so detection and execution
slots are summed from there; a worker that has connected but not yet sent a
heartbeat counts as connected with zero slots.

Fixes #10525

* admin: clamp negative worker-reported slot values in metrics merge

A plugin worker's self-reported heartbeat slot counts are untrusted
input; clamp them to 0 before summing so a stale or misbehaving
worker can't drive the aggregate gauge negative, matching the same
defensiveness already used in registry.go's own slot arithmetic.

weed/admin/dash/admin_server.go

@@ -432,7 +432,72 @@ func (s *AdminServer) publishMaintenanceMetrics(ctx context.Context) {
 	}
 }
 
+// workerFleetTotals aggregates connected workers and their task slots across
+// BOTH worker registries the admin server keeps: the legacy maintenance-worker
+// registry (workers that register over the worker gRPC stream) and the plugin
+// worker registry (workers started as `weed worker`). Reading only the legacy
+// one reported zero workers on clusters that run the admin and the workers as
+// separate components, where no legacy worker ever registers.
+//
+// A worker can appear in both registries: `weed mini` starts both runtimes from
+// one working directory, so they share the persisted worker ID. Merging by ID
+// keeps such a worker counted once, and its slots are taken from the legacy
+// registry, which is where they were accounted for before.
+func (s *AdminServer) workerFleetTotals() (workers, usedSlots, maxSlots int) {
+	var legacySlots map[string]maintenance.WorkerSlots
+	if s.maintenanceManager != nil {
+		legacySlots = s.maintenanceManager.GetWorkerSlots()
+	}
+	return mergeWorkerFleetTotals(legacySlots, s.GetPluginWorkers())
+}
+
+// mergeWorkerFleetTotals unions the legacy and plugin worker registries by
+// worker ID. Plugin workers report their slots in the heartbeat, so one that
+// has connected but not yet sent a heartbeat adds to the worker count with zero
+// slots until its first heartbeat lands.
+func mergeWorkerFleetTotals(legacySlots map[string]maintenance.WorkerSlots, pluginWorkers []*adminplugin.WorkerSession) (workers, usedSlots, maxSlots int) {
+	for _, slots := range legacySlots {
+		workers++
+		usedSlots += slots.Used
+		maxSlots += slots.Max
+	}
+
+	for _, session := range pluginWorkers {
+		if session == nil {
+			continue
+		}
+		if _, counted := legacySlots[session.WorkerID]; counted {
+			continue
+		}
+		workers++
+		if heartbeat := session.Heartbeat; heartbeat != nil {
+			used := int(heartbeat.DetectionSlotsUsed) + int(heartbeat.ExecutionSlotsUsed)
+			max := int(heartbeat.DetectionSlotsTotal) + int(heartbeat.ExecutionSlotsTotal)
+			// A worker's self-reported slots are untrusted input; a stale or
+			// misbehaving one should not be able to drive the aggregate gauge
+			// negative, matching the same defensiveness as registry.go's own
+			// slot arithmetic.
+			if used < 0 {
+				used = 0
+			}
+			if max < 0 {
+				max = 0
+			}
+			usedSlots += used
+			maxSlots += max
+		}
+	}
+	return
+}
+
 func (s *AdminServer) collectMaintenanceMetrics() {
+	// Published before the maintenanceManager guard below: plugin workers are
+	// tracked independently of the maintenance manager.
+	workers, usedSlots, maxSlots := s.workerFleetTotals()
+	stats_collect.AdminWorkersConnected.Set(float64(workers))
+	stats_collect.AdminWorkerSlots.WithLabelValues("used").Set(float64(usedSlots))
+	stats_collect.AdminWorkerSlots.WithLabelValues("max").Set(float64(maxSlots))
+
 	if s.maintenanceManager == nil {
 		return
 	}
@@ -456,11 +521,6 @@ func (s *AdminServer) collectMaintenanceMetrics() {
 	} else {
 		stats_collect.AdminMaintenanceNextScanTimestampSeconds.Set(0)
 	}
-
-	workers, usedSlots, maxSlots := s.maintenanceManager.GetWorkerSlotTotals()
-	stats_collect.AdminWorkersConnected.Set(float64(workers))
-	stats_collect.AdminWorkerSlots.WithLabelValues("used").Set(float64(usedSlots))
-	stats_collect.AdminWorkerSlots.WithLabelValues("max").Set(float64(maxSlots))
 }
 
 // loadTaskConfigurationsFromPersistence loads saved task configurations from protobuf files

weed/admin/dash/dashboard_metrics.go

@@ -73,9 +73,12 @@ func (s *AdminServer) recordDashboardSample() {
 				}
 			}
 			sample.tasks = float64(active)
-			sample.workers = float64(stats.ActiveWorkers)
 		}
 	}
+	// Counted across both worker registries, same as the Prometheus gauge, so
+	// the card isn't stuck at 0 on clusters that only run plugin workers.
+	workers, _, _ := s.workerFleetTotals()
+	sample.workers = float64(workers)
 
 	s.dashSamplesMu.Lock()
 	s.dashSamples = append(s.dashSamples, sample)

weed/admin/dash/worker_fleet_metrics_test.go

@@ -0,0 +1,130 @@
+package dash
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/admin/maintenance"
+	adminplugin "github.com/seaweedfs/seaweedfs/weed/admin/plugin"
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+// TestMergeWorkerFleetTotals covers the accounting behind
+// SeaweedFS_admin_workers_connected / SeaweedFS_admin_worker_slots and the
+// dashboard's Workers card. Both used to read the legacy maintenance-worker
+// registry only, so a cluster whose admin and workers run as separate
+// components reported 0 workers (issue #10525).
+func TestMergeWorkerFleetTotals(t *testing.T) {
+	pluginWorker := func(id string, detectUsed, detectTotal, executeUsed, executeTotal int32) *adminplugin.WorkerSession {
+		return &adminplugin.WorkerSession{
+			WorkerID: id,
+			Heartbeat: &plugin_pb.WorkerHeartbeat{
+				DetectionSlotsUsed:  detectUsed,
+				DetectionSlotsTotal: detectTotal,
+				ExecutionSlotsUsed:  executeUsed,
+				ExecutionSlotsTotal: executeTotal,
+			},
+		}
+	}
+
+	testCases := []struct {
+		name          string
+		legacySlots   map[string]maintenance.WorkerSlots
+		pluginWorkers []*adminplugin.WorkerSession
+		wantWorkers   int
+		wantUsedSlots int
+		wantMaxSlots  int
+	}{
+		{
+			name: "no workers at all",
+		},
+		{
+			name: "legacy workers only, unchanged accounting",
+			legacySlots: map[string]maintenance.WorkerSlots{
+				"w-legacy-a": {Used: 1, Max: 2},
+				"w-legacy-b": {Used: 0, Max: 4},
+			},
+			wantWorkers:   2,
+			wantUsedSlots: 1,
+			wantMaxSlots:  6,
+		},
+		{
+			// The reported bug: admin and workers deployed separately, so
+			// nothing ever lands in the legacy registry.
+			name: "plugin workers only are counted",
+			pluginWorkers: []*adminplugin.WorkerSession{
+				pluginWorker("w-plugin-a", 0, 1, 2, 4),
+				pluginWorker("w-plugin-b", 1, 1, 0, 4),
+			},
+			wantWorkers:   2,
+			wantUsedSlots: 3,
+			wantMaxSlots:  10,
+		},
+		{
+			name: "distinct workers in both registries are summed",
+			legacySlots: map[string]maintenance.WorkerSlots{
+				"w-legacy-a": {Used: 1, Max: 2},
+			},
+			pluginWorkers: []*adminplugin.WorkerSession{
+				pluginWorker("w-plugin-a", 0, 1, 1, 4),
+			},
+			wantWorkers:   2,
+			wantUsedSlots: 2,
+			wantMaxSlots:  7,
+		},
+		{
+			// `weed mini` runs both worker runtimes out of one working
+			// directory, so they share the persisted worker ID and must not be
+			// counted twice. Legacy slots win for such a worker.
+			name: "same worker ID in both registries counts once",
+			legacySlots: map[string]maintenance.WorkerSlots{
+				"w-host-abcd": {Used: 1, Max: 2},
+			},
+			pluginWorkers: []*adminplugin.WorkerSession{
+				pluginWorker("w-host-abcd", 1, 1, 3, 4),
+			},
+			wantWorkers:   1,
+			wantUsedSlots: 1,
+			wantMaxSlots:  2,
+		},
+		{
+			name: "plugin worker without a heartbeat still counts as connected",
+			pluginWorkers: []*adminplugin.WorkerSession{
+				{WorkerID: "w-plugin-a"},
+			},
+			wantWorkers: 1,
+		},
+		{
+			name:          "nil session is skipped",
+			pluginWorkers: []*adminplugin.WorkerSession{nil, pluginWorker("w-plugin-a", 0, 1, 0, 2)},
+			wantWorkers:   1,
+			wantMaxSlots:  3,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			workers, usedSlots, maxSlots := mergeWorkerFleetTotals(tc.legacySlots, tc.pluginWorkers)
+			if workers != tc.wantWorkers {
+				t.Errorf("workers = %d, want %d", workers, tc.wantWorkers)
+			}
+			if usedSlots != tc.wantUsedSlots {
+				t.Errorf("usedSlots = %d, want %d", usedSlots, tc.wantUsedSlots)
+			}
+			if maxSlots != tc.wantMaxSlots {
+				t.Errorf("maxSlots = %d, want %d", maxSlots, tc.wantMaxSlots)
+			}
+		})
+	}
+}
+
+// TestWorkerFleetTotalsWithoutMaintenanceManager guards the nil-manager path:
+// an admin server with no maintenance manager must still report its plugin
+// workers rather than returning early.
+func TestWorkerFleetTotalsWithoutMaintenanceManager(t *testing.T) {
+	server := &AdminServer{}
+
+	workers, usedSlots, maxSlots := server.workerFleetTotals()
+	if workers != 0 || usedSlots != 0 || maxSlots != 0 {
+		t.Fatalf("empty admin server reported workers=%d used=%d max=%d, want all 0", workers, usedSlots, maxSlots)
+	}
+}

weed/admin/maintenance/maintenance_manager.go

@@ -520,9 +520,9 @@ func (mm *MaintenanceManager) GetWorkers() []*MaintenanceWorker {
 	return mm.queue.GetWorkers()
 }
 
-// GetWorkerSlotTotals returns worker count and aggregate used/max task slots.
-func (mm *MaintenanceManager) GetWorkerSlotTotals() (workers, used, max int) {
-	return mm.queue.GetWorkerSlotTotals()
+// GetWorkerSlots returns used/max task slots per worker ID.
+func (mm *MaintenanceManager) GetWorkerSlots() map[string]WorkerSlots {
+	return mm.queue.GetWorkerSlots()
 }
 
 // TriggerScan manually triggers a maintenance scan

weed/admin/maintenance/maintenance_queue.go

@@ -884,18 +884,19 @@ func (mq *MaintenanceQueue) GetWorkers() []*MaintenanceWorker {
 	return workers
 }
 
-// GetWorkerSlotTotals aggregates worker count and used/max task slots under the
-// lock, so callers don't read live worker fields that task updates mutate.
-func (mq *MaintenanceQueue) GetWorkerSlotTotals() (workers, used, max int) {
+// GetWorkerSlots returns used/max task slots per worker ID, snapshotted under
+// the lock so callers don't read live worker fields that task updates mutate.
+// Keyed by ID so callers can merge this registry with the plugin worker
+// registry without counting a worker that appears in both twice.
+func (mq *MaintenanceQueue) GetWorkerSlots() map[string]WorkerSlots {
 	mq.mutex.RLock()
 	defer mq.mutex.RUnlock()
 
-	for _, worker := range mq.workers {
-		workers++
-		used += worker.CurrentLoad
-		max += worker.MaxConcurrent
+	slots := make(map[string]WorkerSlots, len(mq.workers))
+	for id, worker := range mq.workers {
+		slots[id] = WorkerSlots{Used: worker.CurrentLoad, Max: worker.MaxConcurrent}
 	}
-	return
+	return slots
 }
 
 // generateTaskID generates a unique ID for tasks

weed/admin/maintenance/maintenance_types.go

@@ -194,6 +194,13 @@ type MaintenanceWorker struct {
 	CurrentLoad   int                   `json:"current_load"`
 }
 
+// WorkerSlots is a point-in-time snapshot of one worker's task slot usage,
+// taken under the queue lock so callers never read live worker fields.
+type WorkerSlots struct {
+	Used int `json:"used"`
+	Max  int `json:"max"`
+}
+
 // MaintenanceQueue manages the task queue and worker coordination
 type MaintenanceQueue struct {
 	tasks        map[string]*MaintenanceTask