diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index 37f89de83..623dfeba8 100644 --- a/weed/admin/dash/admin_server.go +++ b/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 diff --git a/weed/admin/dash/dashboard_metrics.go b/weed/admin/dash/dashboard_metrics.go index be208354d..2efd00890 100644 --- a/weed/admin/dash/dashboard_metrics.go +++ b/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) diff --git a/weed/admin/dash/worker_fleet_metrics_test.go b/weed/admin/dash/worker_fleet_metrics_test.go new file mode 100644 index 000000000..412ab338d --- /dev/null +++ b/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) + } +} diff --git a/weed/admin/maintenance/maintenance_manager.go b/weed/admin/maintenance/maintenance_manager.go index 509cdc899..898478107 100644 --- a/weed/admin/maintenance/maintenance_manager.go +++ b/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 diff --git a/weed/admin/maintenance/maintenance_queue.go b/weed/admin/maintenance/maintenance_queue.go index 8af9c8f90..b71063106 100644 --- a/weed/admin/maintenance/maintenance_queue.go +++ b/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 diff --git a/weed/admin/maintenance/maintenance_types.go b/weed/admin/maintenance/maintenance_types.go index 811893953..949440486 100644 --- a/weed/admin/maintenance/maintenance_types.go +++ b/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