root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix monitoring daily dispatch cap

Browse Source
This commit is contained in:
liangxu
2026-06-27 16:38:25 +08:00
parent beb290a5a7
commit 11b8e3f5cd
8 changed files with 135 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/cmd/scheduler/main.go
+1 -11
View File
@@ -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),
}
}
server/cmd/scheduler/main_test.go
+4 -3
View File
@@ -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)
server/internal/tenant/app/monitoring_daily_task_worker.go
+15 -7
View File
@@ -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{},
server/internal/tenant/app/monitoring_daily_task_worker_test.go
+11
View File
@@ -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)
server/internal/tenant/app/monitoring_phase2_desktop_tasks.go
+62 -8
View File
@@ -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
}
server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go
+28
View File
@@ -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{
server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.down.sql
+7
View File
@@ -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';
server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.up.sql
+7
View File
@@ -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';