From 11b8e3f5cdd92d6055bcfdb10b8bc89c06b9cc8e Mon Sep 17 00:00:00 2001 From: liangxu Date: Sat, 27 Jun 2026 16:38:25 +0800 Subject: [PATCH] fix monitoring daily dispatch cap --- server/cmd/scheduler/main.go | 12 +--- server/cmd/scheduler/main_test.go | 7 +- .../app/monitoring_daily_task_worker.go | 22 ++++-- .../app/monitoring_daily_task_worker_test.go | 11 +++ .../app/monitoring_phase2_desktop_tasks.go | 70 ++++++++++++++++--- .../monitoring_phase2_desktop_tasks_test.go | 28 ++++++++ ...itoring_daily_global_dispatch_cap.down.sql | 7 ++ ...onitoring_daily_global_dispatch_cap.up.sql | 7 ++ 8 files changed, 135 insertions(+), 29 deletions(-) create mode 100644 server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.down.sql create mode 100644 server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.up.sql diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index 151e646..79dcab6 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -344,18 +344,8 @@ func shutdownSchedulerMetricsServer(server *http.Server, logger interface { } func monitoringDailyRunOptions(jobCtx internalscheduler.JobRunContext) tenantapp.MonitoringDailyTaskRunOptions { - maxDesktopDispatch := 0 - if jobCtx.Job != nil && jobCtx.Job.BatchSize != nil { - maxDesktopDispatch = *jobCtx.Job.BatchSize - } - if maxDesktopDispatch <= 0 { - maxDesktopDispatch = internalscheduler.AsInt(jobCtx.Config, "max_desktop_dispatch_per_run", 0) - } - if maxDesktopDispatch <= 0 { - maxDesktopDispatch = internalscheduler.AsInt(jobCtx.Config, "batch_size", 0) - } return tenantapp.MonitoringDailyTaskRunOptions{ - MaxDesktopDispatchPerRun: maxDesktopDispatch, + MaxDesktopDispatchPerRun: internalscheduler.AsInt(jobCtx.Config, "max_desktop_dispatch_per_run", 0), MaxDesktopClientBacklog: internalscheduler.AsInt(jobCtx.Config, "max_desktop_client_backlog", 0), } } diff --git a/server/cmd/scheduler/main_test.go b/server/cmd/scheduler/main_test.go index bf132b9..0ffe385 100644 --- a/server/cmd/scheduler/main_test.go +++ b/server/cmd/scheduler/main_test.go @@ -59,7 +59,7 @@ func TestSchedulerMetricsAuthMiddleware(t *testing.T) { } } -func TestMonitoringDailyRunOptionsUsesJobBatchSizeForDesktopDispatchOnly(t *testing.T) { +func TestMonitoringDailyRunOptionsIgnoresJobBatchSizeForDailyCollection(t *testing.T) { batchSize := 88 got := monitoringDailyRunOptions(internalscheduler.JobRunContext{ Job: &opsdomain.SchedulerJob{BatchSize: &batchSize}, @@ -68,11 +68,12 @@ func TestMonitoringDailyRunOptionsUsesJobBatchSizeForDesktopDispatchOnly(t *test "max_materialize_per_plan": 12, "max_materialize_per_brand": 4, "max_desktop_client_backlog": 6, + "batch_size": 77, }, }) - if got.MaxDesktopDispatchPerRun != 88 { - t.Fatalf("MaxDesktopDispatchPerRun = %d, want 88", got.MaxDesktopDispatchPerRun) + if got.MaxDesktopDispatchPerRun != 0 { + t.Fatalf("MaxDesktopDispatchPerRun = %d, want unlimited", got.MaxDesktopDispatchPerRun) } if got.MaxMaterializePerPlan != 0 || got.MaxMaterializePerBrand != 0 { t.Fatalf("legacy plan/brand throttles should be ignored: %#v", got) diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 284737d..06afe5f 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -29,7 +29,7 @@ const ( defaultMonitoringDailyMaxMaterializePerRun = 0 defaultMonitoringDailyMaxMaterializePerPlan = 0 defaultMonitoringDailyMaxMaterializePerBrand = 0 - defaultMonitoringDailyMaxDesktopDispatch = 12 + defaultMonitoringDailyMaxDesktopDispatch = 0 defaultMonitoringDailyMaxDesktopClientBacklog = 12 monitoringDailyPlatformActiveWindow = 24 * time.Hour monitoringDailyMaterializeStartOffset = 5 * time.Minute @@ -283,7 +283,6 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( } options := normalizeMonitoringDailyRunOptions(runOptions...) globalMaterializeRemaining := options.MaxMaterializePerRun - desktopDispatchRemaining := options.MaxDesktopDispatchPerRun for _, plan := range plans { if monitoringDailyLimitReached(globalMaterializeRemaining, options.MaxMaterializePerRun) { @@ -328,6 +327,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( if options.MaxMaterializePerRun > 0 && (planMaterializeRemaining <= 0 || globalMaterializeRemaining < planMaterializeRemaining) { planMaterializeRemaining = globalMaterializeRemaining } + planDesktopDispatchRemaining := monitoringDailyPlanDesktopDispatchBudget(options) for _, brand := range brands { if monitoringDailyLimitReached(planMaterializeRemaining, options.MaxMaterializePerPlan) || @@ -360,7 +360,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( planMaterializeRemaining, globalMaterializeRemaining, ) - dispatchLimit := monitoringDailyDispatchLimit(len(candidates), desktopDispatchRemaining, options.MaxDesktopDispatchPerRun) + dispatchLimit := monitoringDailyDispatchLimit(len(candidates), planDesktopDispatchRemaining, planDesktopDispatchRemaining) createdCount, eligibleCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand( ctx, projectionService, @@ -390,8 +390,8 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( if options.MaxMaterializePerRun > 0 { globalMaterializeRemaining -= int(dueCount) } - if options.MaxDesktopDispatchPerRun > 0 { - desktopDispatchRemaining -= int(desktopCount + deferredCount) + if planDesktopDispatchRemaining > 0 { + planDesktopDispatchRemaining -= int(desktopCount + deferredCount) } summary.BrandCount++ summary.PlannedTaskCount += int64(len(candidates)) @@ -479,7 +479,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( if dispatchLimit <= 0 { return createdCount, eligibleCount, dueCount, 0, 0, nil } - dispatchablePlatformIDs, err := s.loadDailyMonitorDispatchablePlatformIDs(ctx, plan.TenantID, plan.WorkspaceID, candidates) + dispatchablePlatformIDs, err := s.loadDailyMonitorDispatchablePlatformIDs(ctx, plan.TenantID, plan.WorkspaceID, candidates, maxDesktopClientBacklog) if err != nil { return createdCount, eligibleCount, dueCount, 0, 0, err } @@ -984,6 +984,7 @@ func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs( ctx context.Context, tenantID, workspaceID int64, candidates []monitoringDailyTaskCandidate, + maxDesktopClientBacklog int, ) ([]string, error) { if len(candidates) == 0 { return nil, nil @@ -1006,7 +1007,7 @@ func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs( return nil, nil } - targets, err := s.loadMonitorDesktopTaskTargets(ctx, tenantID, workspaceID, 0, specs) + targets, err := s.loadIdleMonitorDesktopTaskTargets(ctx, tenantID, workspaceID, 0, specs, maxDesktopClientBacklog) if err != nil { return nil, err } @@ -1443,6 +1444,13 @@ func monitoringDailyDispatchLimit(candidateCount, remaining, configuredLimit int return monitoringDailyBoundedLimit(candidateCount, remaining) } +func monitoringDailyPlanDesktopDispatchBudget(options MonitoringDailyTaskRunOptions) int { + if options.MaxDesktopDispatchPerRun > 0 { + return options.MaxDesktopDispatchPerRun + } + return options.MaxDesktopClientBacklog +} + func selectDueMonitoringDailyCandidates( candidates []monitoringDailyTaskCandidate, existing map[monitoringDailyTaskKey]struct{}, diff --git a/server/internal/tenant/app/monitoring_daily_task_worker_test.go b/server/internal/tenant/app/monitoring_daily_task_worker_test.go index cd5a7f9..e8ef704 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -47,6 +47,7 @@ func TestNormalizeMonitoringDailyRunOptionsDefaultsDoNotThrottleByPlanOrBrand(t assert.Equal(t, 0, got.MaxMaterializePerRun) assert.Equal(t, 0, got.MaxMaterializePerPlan) assert.Equal(t, 0, got.MaxMaterializePerBrand) + assert.Equal(t, 0, got.MaxDesktopDispatchPerRun) } func TestMonitoringDailyBoundedLimitTreatsZeroCapsAsUnlimited(t *testing.T) { @@ -61,6 +62,16 @@ func TestMonitoringDailyDispatchLimitStopsWhenRunCapacityIsExhausted(t *testing. assert.Equal(t, 0, monitoringDailyDispatchLimit(102, 0, 12)) } +func TestMonitoringDailyPlanDesktopDispatchBudgetUsesBacklogWhenUnconfigured(t *testing.T) { + assert.Equal(t, 12, monitoringDailyPlanDesktopDispatchBudget(MonitoringDailyTaskRunOptions{ + MaxDesktopClientBacklog: 12, + })) + assert.Equal(t, 6, monitoringDailyPlanDesktopDispatchBudget(MonitoringDailyTaskRunOptions{ + MaxDesktopDispatchPerRun: 6, + MaxDesktopClientBacklog: 12, + })) +} + func TestAutomaticDailySelectionIncludesAllUserQuestionsAndPlatforms(t *testing.T) { businessDay := time.Date(2026, 6, 27, 0, 0, 0, 0, monitoringBusinessLocation()) questions := make([]monitoringConfiguredQuestion, 0, 17) diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 2877a46..b19ec6d 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -66,6 +66,11 @@ type monitorDesktopTaskTargetCandidate struct { ClientOrder int } +type monitorDesktopTaskTargetOptions struct { + MaxClientBacklog int + RequireIdleClient bool +} + func monitorDesktopTaskSpecIDs(specs []monitorDesktopTaskSpec) []int64 { if len(specs) == 0 { return nil @@ -332,14 +337,37 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( specs []monitorDesktopTaskSpec, maxClientBacklog ...int, ) (map[string]monitorDesktopTaskTarget, error) { - platformIDs := uniqueMonitorDesktopTaskPlatformIDs(specs) - if len(platformIDs) == 0 { - return nil, nil - } backlogLimit := 0 if len(maxClientBacklog) > 0 && maxClientBacklog[0] > 0 { backlogLimit = maxClientBacklog[0] } + return s.loadMonitorDesktopTaskTargetsWithOptions(ctx, tenantID, workspaceID, userID, specs, monitorDesktopTaskTargetOptions{ + MaxClientBacklog: backlogLimit, + }) +} + +func (s *MonitoringService) loadIdleMonitorDesktopTaskTargets( + ctx context.Context, + tenantID, workspaceID, userID int64, + specs []monitorDesktopTaskSpec, + maxClientBacklog int, +) (map[string]monitorDesktopTaskTarget, error) { + return s.loadMonitorDesktopTaskTargetsWithOptions(ctx, tenantID, workspaceID, userID, specs, monitorDesktopTaskTargetOptions{ + MaxClientBacklog: maxClientBacklog, + RequireIdleClient: true, + }) +} + +func (s *MonitoringService) loadMonitorDesktopTaskTargetsWithOptions( + ctx context.Context, + tenantID, workspaceID, userID int64, + specs []monitorDesktopTaskSpec, + options monitorDesktopTaskTargetOptions, +) (map[string]monitorDesktopTaskTarget, error) { + platformIDs := uniqueMonitorDesktopTaskPlatformIDs(specs) + if len(platformIDs) == 0 { + return nil, nil + } rows, err := s.businessPool.Query(ctx, ` SELECT @@ -429,11 +457,15 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( if err != nil { return nil, err } - activeBacklog, err := s.loadMonitorDesktopClientBacklog(ctx, tenantID, workspaceID, clientIDs, backlogLimit) + backlogLookupLimit := options.MaxClientBacklog + if options.RequireIdleClient && backlogLookupLimit <= 0 { + backlogLookupLimit = 1 + } + activeBacklog, err := s.loadMonitorDesktopClientBacklog(ctx, tenantID, workspaceID, clientIDs, backlogLookupLimit) if err != nil { return nil, err } - return selectMonitorDesktopTaskTargets(candidates, accountPresence, onlineClients, activeBacklog, backlogLimit), nil + return selectMonitorDesktopTaskTargetsWithOptions(candidates, accountPresence, onlineClients, activeBacklog, options), nil } func uniqueMonitorDesktopTaskPlatformIDs(specs []monitorDesktopTaskSpec) []string { @@ -572,6 +604,18 @@ func selectMonitorDesktopTaskTargets( onlineClients map[uuid.UUID]bool, clientBacklog map[uuid.UUID]int, maxClientBacklog int, +) map[string]monitorDesktopTaskTarget { + return selectMonitorDesktopTaskTargetsWithOptions(candidates, accountPresence, onlineClients, clientBacklog, monitorDesktopTaskTargetOptions{ + MaxClientBacklog: maxClientBacklog, + }) +} + +func selectMonitorDesktopTaskTargetsWithOptions( + candidates []monitorDesktopAccountCandidate, + accountPresence map[uuid.UUID][]desktopAccountPresenceClient, + onlineClients map[uuid.UUID]bool, + clientBacklog map[uuid.UUID]int, + options monitorDesktopTaskTargetOptions, ) map[string]monitorDesktopTaskTarget { result := make(map[string]monitorDesktopTaskTarget) candidatesByPlatform := make(map[string][]monitorDesktopTaskTargetCandidate) @@ -634,11 +678,11 @@ func selectMonitorDesktopTaskTargets( return platformCandidates[i].betterThan(platformCandidates[j]) }) for _, candidate := range platformCandidates { - if monitorDesktopClientBacklogFull(candidate.Target.ClientID, clientBacklog, maxClientBacklog) { + if monitorDesktopClientBacklogUnavailable(candidate.Target.ClientID, clientBacklog, options) { continue } result[platformID] = candidate.Target - if maxClientBacklog > 0 { + if options.MaxClientBacklog > 0 { clientBacklog[candidate.Target.ClientID]++ } break @@ -647,6 +691,16 @@ func selectMonitorDesktopTaskTargets( return result } +func monitorDesktopClientBacklogUnavailable(clientID uuid.UUID, clientBacklog map[uuid.UUID]int, options monitorDesktopTaskTargetOptions) bool { + if clientID == uuid.Nil { + return false + } + if options.RequireIdleClient && clientBacklog[clientID] > 0 { + return true + } + return monitorDesktopClientBacklogFull(clientID, clientBacklog, options.MaxClientBacklog) +} + func monitorDesktopClientBacklogFull(clientID uuid.UUID, clientBacklog map[uuid.UUID]int, maxClientBacklog int) bool { return maxClientBacklog > 0 && clientID != uuid.Nil && clientBacklog[clientID] >= maxClientBacklog } diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go index a29398d..007428b 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -239,6 +239,34 @@ func TestSelectMonitorDesktopTaskTargetsConsumesBackpressureWithinRun(t *testing assert.NotContains(t, targets, "qwen") } +func TestSelectMonitorDesktopTaskTargetsRequireIdleSkipsAnyActiveBacklog(t *testing.T) { + clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000708") + accountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000808") + + targets := selectMonitorDesktopTaskTargetsWithOptions( + []monitorDesktopAccountCandidate{ + { + PlatformID: "qwen", + AccountID: accountID, + DBClientID: &clientID, + }, + }, + nil, + map[uuid.UUID]bool{ + clientID: true, + }, + map[uuid.UUID]int{ + clientID: 1, + }, + monitorDesktopTaskTargetOptions{ + MaxClientBacklog: 12, + RequireIdleClient: true, + }, + ) + + assert.Empty(t, targets) +} + func TestReserveMonitorDesktopClientBacklogAllowsExistingSameClientWithoutConsumingSlot(t *testing.T) { clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000704") backlog := map[uuid.UUID]int{ diff --git a/server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.down.sql b/server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.down.sql new file mode 100644 index 0000000..ff69178 --- /dev/null +++ b/server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.down.sql @@ -0,0 +1,7 @@ +UPDATE ops.scheduler_jobs +SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。', + batch_size = 12, + config = (config - 'max_desktop_dispatch_per_run' - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand') + || '{"max_desktop_client_backlog":12}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_daily_task'; diff --git a/server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.up.sql b/server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.up.sql new file mode 100644 index 0000000..d1c1d97 --- /dev/null +++ b/server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.up.sql @@ -0,0 +1,7 @@ +UPDATE ops.scheduler_jobs +SET description = '客户端在线时按业务日补齐当天监控采集任务，离线不生成待采积压；不设置全局派发上限，客户端侧由 backlog 控制。', + batch_size = NULL, + config = (config - 'batch_size' - 'max_desktop_dispatch_per_run' - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand') + || '{"max_desktop_client_backlog":12}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_daily_task';