From beb290a5a725d39636dbbddf791c23a7a541e82b Mon Sep 17 00:00:00 2001 From: liangxu Date: Sat, 27 Jun 2026 16:17:02 +0800 Subject: [PATCH] fix monitoring online dispatch retries --- .../tenant/app/monitoring_callback_service.go | 170 ++++++-- .../app/monitoring_callback_service_test.go | 67 +++ .../app/monitoring_daily_task_metrics.go | 4 + .../app/monitoring_daily_task_worker.go | 384 +++++++++++++----- .../app/monitoring_daily_task_worker_test.go | 63 ++- .../app/monitoring_metrics_prometheus.go | 1 + ...000_create_scheduler_control_tables.up.sql | 2 +- ...43000_harden_scheduler_jobs_for_k3s.up.sql | 2 +- ...ework_monitoring_daily_backpressure.up.sql | 2 +- ...ment_online_only_monitoring_daily.down.sql | 6 + ...cument_online_only_monitoring_daily.up.sql | 7 + 11 files changed, 555 insertions(+), 153 deletions(-) create mode 100644 server/migrations_ops/20260627184500_document_online_only_monitoring_daily.down.sql create mode 100644 server/migrations_ops/20260627184500_document_online_only_monitoring_daily.up.sql diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index f37c038..d6637a6 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -40,7 +40,9 @@ const ( // reports skip writes and only refresh the row at this interval so dashboards // do not depend on per-heartbeat database churn. monitoringHeartbeatAccessSnapshotRefreshAfter = 10 * time.Minute - monitoringRuntimeUnknownMaxDispatchAttempts = 3 + monitoringRuntimeUnknownMaxDispatchAttempts = 2 + monitoringAutomaticFailureMaxAttempts = 2 + monitoringAutomaticFailureRetryAfter = 2 * time.Minute ) var monitoringCitationMarkerPattern = regexp.MustCompile(`(?i)\[\s*citation\s*:\s*\d+\s*\]`) @@ -240,27 +242,28 @@ type monitoringHeartbeatSnapshotDecision struct { } type monitoringCollectTask struct { - ID int64 - TenantID int64 - BrandID int64 - QuestionID int64 - QuestionHash []byte - AIPlatformID string - CollectorType string - RunMode string - BusinessDate time.Time - Status string - ExecutionOwner string - LeaseTokenHash sql.NullString - LeasedToExecutor sql.NullString - LeaseExpiresAt sql.NullTime - CallbackReceivedAt sql.NullTime - CompletedAt sql.NullTime - SkipReason sql.NullString - TriggerSource string - QuestionText string - TargetClientID sql.NullString - DispatchAttempts int + ID int64 + TenantID int64 + BrandID int64 + QuestionID int64 + QuestionHash []byte + AIPlatformID string + CollectorType string + RunMode string + BusinessDate time.Time + Status string + ExecutionOwner string + LeaseTokenHash sql.NullString + LeasedToExecutor sql.NullString + LeaseExpiresAt sql.NullTime + CallbackReceivedAt sql.NullTime + CompletedAt sql.NullTime + SkipReason sql.NullString + TriggerSource string + QuestionText string + TargetClientID sql.NullString + DispatchAttempts int + AutomaticFailureCount int } type monitoringDesktopExecutionLease struct { @@ -1062,6 +1065,32 @@ type monitoringSkipOutcome struct { RetryAfter time.Duration } +type monitoringFailedResultOutcome struct { + TaskStatus string + Requeue bool + RetryAfter time.Duration +} + +func monitoringFailedResultOutcomeForTask(task *monitoringCollectTask, req MonitoringTaskResultRequest) monitoringFailedResultOutcome { + if task == nil { + return monitoringFailedResultOutcome{TaskStatus: "failed"} + } + if !strings.EqualFold(strings.TrimSpace(task.TriggerSource), "automatic") { + return monitoringFailedResultOutcome{TaskStatus: "failed"} + } + if monitoringTaskFailureDispositionFromRequest(req).NonRetryable { + return monitoringFailedResultOutcome{TaskStatus: "failed"} + } + if task.AutomaticFailureCount < monitoringAutomaticFailureMaxAttempts-1 { + return monitoringFailedResultOutcome{ + TaskStatus: "pending", + Requeue: true, + RetryAfter: monitoringAutomaticFailureRetryAfter, + } + } + return monitoringFailedResultOutcome{TaskStatus: "failed"} +} + func monitoringSkipOutcomeForReason(skipReason string, dispatchAttempts int, triggerSource string) monitoringSkipOutcome { switch strings.TrimSpace(skipReason) { case "runtime_unknown": @@ -1562,6 +1591,11 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI trigger_source, COALESCE(dispatch_attempts, 0) AS dispatch_attempts, COALESCE(target_client_id::text, '') AS target_client_id, + CASE + WHEN COALESCE(t.request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$' + THEN (t.request_payload_json ->> 'automatic_failure_count')::int + ELSE 0 + END AS automatic_failure_count, COALESCE(NULLIF(t.question_text_snapshot, ''), ( SELECT question_text_snapshot FROM monitoring_question_config_snapshots @@ -1600,6 +1634,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI &item.TriggerSource, &item.DispatchAttempts, &item.TargetClientID, + &item.AutomaticFailureCount, &item.QuestionText, ); err != nil { if err == pgx.ErrNoRows { @@ -1635,6 +1670,11 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas trigger_source, COALESCE(dispatch_attempts, 0) AS dispatch_attempts, COALESCE(target_client_id::text, '') AS target_client_id, + CASE + WHEN COALESCE(t.request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$' + THEN (t.request_payload_json ->> 'automatic_failure_count')::int + ELSE 0 + END AS automatic_failure_count, COALESCE(NULLIF(t.question_text_snapshot, ''), ( SELECT question_text_snapshot FROM monitoring_question_config_snapshots @@ -1672,6 +1712,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas &item.TriggerSource, &item.DispatchAttempts, &item.TargetClientID, + &item.AutomaticFailureCount, &item.QuestionText, ); err != nil { if err == pgx.ErrNoRows { @@ -2312,12 +2353,20 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task } defer tx.Rollback(ctx) - runID, err := s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload) - if err != nil { - return nil, err + var failedOutcome monitoringFailedResultOutcome + if runStatus == "failed" { + failedOutcome = monitoringFailedResultOutcomeForTask(task, req) } - if runStatus == "succeeded" { + var runID int64 + if !failedOutcome.Requeue { + runID, err = s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload) + if err != nil { + return nil, err + } + } + + if runStatus == "succeeded" && !failedOutcome.Requeue { if err := s.upsertParseResult(ctx, tx, task, runID, parseReq, parseVersion); err != nil { return nil, err } @@ -2325,10 +2374,12 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task citationSourceCount := 0 contentCitationCount := 0 - if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil { - return nil, err + if !failedOutcome.Requeue { + if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil { + return nil, err + } } - if runStatus == "succeeded" { + if runStatus == "succeeded" && !failedOutcome.Requeue { facts, buildErr := s.buildCitationFacts(ctx, tx, task.TenantID, task.AIPlatformID, sourceInputs) if buildErr != nil { return nil, buildErr @@ -2346,12 +2397,16 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task taskStatus := "completed" if runStatus == "failed" { - taskStatus = "failed" + taskStatus = failedOutcome.TaskStatus } - if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil { + if failedOutcome.Requeue { + if err := s.requeueCollectTaskAfterFailedResult(ctx, tx, task.ID, failedOutcome.RetryAfter, req.ErrorMessage); err != nil { + return nil, err + } + } else if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil { return nil, err } - if runStatus == "failed" { + if runStatus == "failed" && !failedOutcome.Requeue { if _, err := s.failRemainingMonitoringTasksForAuthorizationFailure(ctx, tx, task, req, completedAt); err != nil { return nil, err } @@ -2365,9 +2420,13 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring result ingest") } + var runIDPtr *int64 + if !failedOutcome.Requeue { + runIDPtr = &runID + } return &MonitoringTaskResultResponse{ TaskID: task.ID, - RunID: &runID, + RunID: runIDPtr, TaskStatus: taskStatus, CitationSourceCount: citationSourceCount, ContentCitationCount: contentCitationCount, @@ -3113,6 +3172,51 @@ func (s *MonitoringCallbackService) updateCollectTaskStatus(ctx context.Context, return nil } +func (s *MonitoringCallbackService) requeueCollectTaskAfterFailedResult(ctx context.Context, tx pgx.Tx, taskID int64, retryAfter time.Duration, errorMessage *string) error { + if retryAfter <= 0 { + retryAfter = monitoringAutomaticFailureRetryAfter + } + if _, err := tx.Exec(ctx, ` + UPDATE monitoring_collect_tasks + SET status = 'pending', + lease_token_hash = NULL, + leased_to_executor = NULL, + leased_at = NULL, + lease_expires_at = NULL, + callback_received_at = NULL, + completed_at = NULL, + skip_reason = NULL, + last_dispatched_at = NULL, + dispatch_after = NOW() + $2::interval, + error_message = $3, + request_payload_json = ( + CASE + WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object' + THEN '{}'::jsonb + ELSE request_payload_json + END + ) || jsonb_build_object( + 'automatic_failure_count', + ( + CASE + WHEN COALESCE(request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$' + THEN (request_payload_json ->> 'automatic_failure_count')::int + ELSE 0 + END + ) + 1, + 'automatic_failure_retry_after_seconds', + $4::int, + 'automatic_failure_last_requeued_at', + to_jsonb(NOW()) + ), + updated_at = NOW() + WHERE id = $1 + `, taskID, formatPgInterval(retryAfter), nullableString(errorMessage), int(retryAfter.Seconds())); err != nil { + return response.ErrInternal(50041, "task_update_failed", "failed to requeue monitoring task after failed result") + } + return nil +} + func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorizationFailure( ctx context.Context, tx pgx.Tx, diff --git a/server/internal/tenant/app/monitoring_callback_service_test.go b/server/internal/tenant/app/monitoring_callback_service_test.go index f5df44f..b436c2b 100644 --- a/server/internal/tenant/app/monitoring_callback_service_test.go +++ b/server/internal/tenant/app/monitoring_callback_service_test.go @@ -499,6 +499,73 @@ func TestMonitoringTaskFailureDispositionDetectsAuthorizationFailureFromMessage( } } +func TestMonitoringFailedResultOutcomeRequeuesAutomaticFirstFailure(t *testing.T) { + message := "模型调用超时" + got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{ + TriggerSource: "automatic", + AutomaticFailureCount: 0, + }, MonitoringTaskResultRequest{ + Status: "failed", + ErrorMessage: &message, + }) + + if got.TaskStatus != "pending" { + t.Fatalf("TaskStatus = %q, want pending", got.TaskStatus) + } + if !got.Requeue { + t.Fatal("Requeue = false, want true") + } + if got.RetryAfter != monitoringAutomaticFailureRetryAfter { + t.Fatalf("RetryAfter = %v, want %v", got.RetryAfter, monitoringAutomaticFailureRetryAfter) + } +} + +func TestMonitoringFailedResultOutcomeFailsAutomaticSecondFailure(t *testing.T) { + got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{ + TriggerSource: "automatic", + AutomaticFailureCount: monitoringAutomaticFailureMaxAttempts - 1, + }, MonitoringTaskResultRequest{Status: "failed"}) + + if got.TaskStatus != "failed" { + t.Fatalf("TaskStatus = %q, want failed", got.TaskStatus) + } + if got.Requeue { + t.Fatal("Requeue = true, want false") + } +} + +func TestMonitoringFailedResultOutcomeDoesNotRetryAuthorizationFailure(t *testing.T) { + message := "登录态已失效，请重新授权后再试。" + got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{ + TriggerSource: "automatic", + AutomaticFailureCount: 0, + }, MonitoringTaskResultRequest{ + Status: "failed", + ErrorMessage: &message, + }) + + if got.TaskStatus != "failed" { + t.Fatalf("TaskStatus = %q, want failed", got.TaskStatus) + } + if got.Requeue { + t.Fatal("Requeue = true, want false") + } +} + +func TestMonitoringFailedResultOutcomeDoesNotRetryManualTask(t *testing.T) { + got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{ + TriggerSource: "manual", + AutomaticFailureCount: 0, + }, MonitoringTaskResultRequest{Status: "failed"}) + + if got.TaskStatus != "failed" { + t.Fatalf("TaskStatus = %q, want failed", got.TaskStatus) + } + if got.Requeue { + t.Fatal("Requeue = true, want false") + } +} + func TestMonitoringAuthorizationFailureTargetClientPrefersExplicitTargetClient(t *testing.T) { targetClientID := uuid.New().String() task := &monitoringCollectTask{ diff --git a/server/internal/tenant/app/monitoring_daily_task_metrics.go b/server/internal/tenant/app/monitoring_daily_task_metrics.go index fdd20a2..a57e9a7 100644 --- a/server/internal/tenant/app/monitoring_daily_task_metrics.go +++ b/server/internal/tenant/app/monitoring_daily_task_metrics.go @@ -20,6 +20,7 @@ type MonitoringDailyTaskMetricsSnapshot struct { BrandFailureTotal int64 `json:"brand_failure_total"` SkippedPlanTotal int64 `json:"skipped_plan_total"` PlannedTaskTotal int64 `json:"planned_task_total"` + EligibleTaskTotal int64 `json:"eligible_task_total"` DueTaskTotal int64 `json:"due_task_total"` CreatedTaskTotal int64 `json:"created_task_total"` DesktopTaskTotal int64 `json:"desktop_task_total"` @@ -44,6 +45,7 @@ type monitoringDailyTaskMetricsRecorder struct { brandFailureTotal atomic.Int64 skippedPlanTotal atomic.Int64 plannedTaskTotal atomic.Int64 + eligibleTaskTotal atomic.Int64 dueTaskTotal atomic.Int64 createdTaskTotal atomic.Int64 desktopTaskTotal atomic.Int64 @@ -98,6 +100,7 @@ func (r *monitoringDailyTaskMetricsRecorder) record(summary *MonitoringDailyTask r.brandFailureTotal.Add(int64(summary.FailedBrandCount)) r.skippedPlanTotal.Add(int64(summary.SkippedPlanCount)) r.plannedTaskTotal.Add(summary.PlannedTaskCount) + r.eligibleTaskTotal.Add(summary.EligibleTaskCount) r.dueTaskTotal.Add(summary.DueTaskCount) r.createdTaskTotal.Add(summary.CreatedTaskCount) r.desktopTaskTotal.Add(summary.DesktopTaskCount) @@ -133,6 +136,7 @@ func (r *monitoringDailyTaskMetricsRecorder) snapshot() MonitoringDailyTaskMetri BrandFailureTotal: r.brandFailureTotal.Load(), SkippedPlanTotal: r.skippedPlanTotal.Load(), PlannedTaskTotal: r.plannedTaskTotal.Load(), + EligibleTaskTotal: r.eligibleTaskTotal.Load(), DueTaskTotal: r.dueTaskTotal.Load(), CreatedTaskTotal: r.createdTaskTotal.Load(), DesktopTaskTotal: r.desktopTaskTotal.Load(), diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 8d32ef0..284737d 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -50,6 +50,7 @@ type MonitoringDailyTaskGenerationSummary struct { FailedPlanCount int FailedBrandCount int PlannedTaskCount int64 + EligibleTaskCount int64 DueTaskCount int64 CreatedTaskCount int64 DesktopTaskCount int64 @@ -238,6 +239,7 @@ func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context, options ...M stats["failed_plan_count"] = summary.FailedPlanCount stats["failed_brand_count"] = summary.FailedBrandCount stats["planned_task_count"] = summary.PlannedTaskCount + stats["eligible_task_count"] = summary.EligibleTaskCount stats["due_task_count"] = summary.DueTaskCount stats["created_task_count"] = summary.CreatedTaskCount stats["desktop_task_count"] = summary.DesktopTaskCount @@ -359,7 +361,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( globalMaterializeRemaining, ) dispatchLimit := monitoringDailyDispatchLimit(len(candidates), desktopDispatchRemaining, options.MaxDesktopDispatchPerRun) - createdCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand( + createdCount, eligibleCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand( ctx, projectionService, plan, @@ -393,6 +395,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( } summary.BrandCount++ summary.PlannedTaskCount += int64(len(candidates)) + summary.EligibleTaskCount += eligibleCount summary.DueTaskCount += dueCount summary.CreatedTaskCount += createdCount summary.DesktopTaskCount += desktopCount @@ -460,86 +463,101 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( dispatchLimit int, executionOwner string, maxDesktopClientBacklog int, -) (int64, int64, int64, int64, error) { +) (int64, int64, int64, int64, int64, error) { if materializeLimit <= 0 && dispatchLimit <= 0 { - return 0, 0, 0, 0, nil + return 0, 0, 0, 0, 0, nil } if executionOwner != "desktop_tasks" { - return 0, 0, 0, 0, nil + return 0, 0, 0, 0, 0, nil } var createdCount int64 + var eligibleCount int64 var dueCount int64 - if materializeLimit > 0 { - tx, err := s.monitoringPool.Begin(ctx) - if err != nil { - return 0, 0, 0, 0, response.ErrInternal(50041, "begin_failed", "failed to start monitoring daily task generation") - } - defer tx.Rollback(ctx) - - locked, err := acquireMonitoringDailyBrandLock(ctx, tx, plan.TenantID, brand.BrandID, monitoringBusinessDateText(businessDay)) - if err != nil { - return 0, 0, 0, 0, err - } - - if locked { - existingKeys, err := s.loadDailyMonitoringTaskKeys(ctx, tx, plan.TenantID, brand.BrandID, businessDay) - if err != nil { - return 0, 0, 0, 0, err - } - - dueCandidates := selectDueMonitoringDailyCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, materializeLimit) - dueCount = int64(len(dueCandidates)) - if len(dueCandidates) > 0 { - if err := s.syncMonitoringQuestionSnapshots(ctx, tx, plan.TenantID, brand.BrandID, questions, true); err != nil { - return 0, 0, 0, 0, err - } - - createdCount, err = s.ensureDailyMonitoringCollectTasks(ctx, tx, plan, brand, businessDay, dueCandidates, executionOwner) - if err != nil { - return 0, 0, 0, 0, err - } - - if projectionService != nil && createdCount > 0 { - if err := projectionService.rebuildBrandDailyProjection(ctx, tx, plan.TenantID, brand.BrandID, businessDay); err != nil { - return 0, 0, 0, 0, err - } - } - } - } - - if err := tx.Commit(ctx); err != nil { - return 0, 0, 0, 0, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring daily task generation") - } - } + specs := make([]monitorDesktopTaskSpec, 0) if dispatchLimit <= 0 { - return createdCount, dueCount, 0, 0, nil + return createdCount, eligibleCount, dueCount, 0, 0, nil } dispatchablePlatformIDs, err := s.loadDailyMonitorDispatchablePlatformIDs(ctx, plan.TenantID, plan.WorkspaceID, candidates) if err != nil { - return createdCount, dueCount, 0, 0, err + return createdCount, eligibleCount, dueCount, 0, 0, err } if len(dispatchablePlatformIDs) == 0 { - return createdCount, dueCount, 0, 0, nil + return createdCount, eligibleCount, dueCount, 0, 0, nil } - specs, err := s.loadDueDailyMonitorDesktopTaskSpecs( + dispatchablePlatformSet := monitoringDailyPlatformSet(dispatchablePlatformIDs) + dispatchableCandidates := filterMonitoringDailyCandidatesByPlatform(candidates, dispatchablePlatformSet) + eligibleCount = int64(len(dispatchableCandidates)) + + retrySpecs, err := s.loadDueDailyMonitorDesktopTaskSpecs( ctx, s.monitoringPool, plan.TenantID, plan.WorkspaceID, brand.BrandID, businessDay, - now.Add(defaultMonitoringDailyLookahead), + now, now.Add(-defaultMonitoringDailyDesktopRetryCooldown), dispatchablePlatformIDs, dispatchLimit, ) if err != nil { - return createdCount, dueCount, 0, 0, err + return createdCount, eligibleCount, dueCount, 0, 0, err + } + if len(retrySpecs) > 0 { + specs = append(specs, retrySpecs...) + createdCount += int64(len(retrySpecs)) + dueCount += int64(len(retrySpecs)) + } + + remainingCreateLimit := dispatchLimit - len(specs) + if remainingCreateLimit <= 0 { + return s.dispatchDailyMonitoringDesktopSpecs(ctx, specs, maxDesktopClientBacklog, createdCount, eligibleCount, dueCount) + } + + materializeLimit = monitoringDailyBoundedLimit(materializeLimit, dispatchLimit) + if materializeLimit > 0 && len(dispatchableCandidates) > 0 { + var materializeErr error + var newCreatedCount int64 + var newDueCount int64 + var newSpecs []monitorDesktopTaskSpec + newCreatedCount, newDueCount, newSpecs, materializeErr = s.createDispatchableDailyMonitoringTasksForBrand( + ctx, + projectionService, + plan, + brand, + businessDay, + now, + questions, + dispatchableCandidates, + monitoringDailyBoundedLimit(materializeLimit, remainingCreateLimit), + executionOwner, + ) + if materializeErr != nil { + return createdCount, eligibleCount, dueCount, 0, 0, materializeErr + } + createdCount += newCreatedCount + dueCount += newDueCount + specs = append(specs, newSpecs...) } if len(specs) == 0 { - return createdCount, dueCount, 0, 0, nil + return createdCount, eligibleCount, dueCount, 0, 0, nil + } + + return s.dispatchDailyMonitoringDesktopSpecs(ctx, specs, maxDesktopClientBacklog, createdCount, eligibleCount, dueCount) +} + +func (s *MonitoringService) dispatchDailyMonitoringDesktopSpecs( + ctx context.Context, + specs []monitorDesktopTaskSpec, + maxDesktopClientBacklog int, + createdCount int64, + eligibleCount int64, + dueCount int64, +) (int64, int64, int64, int64, int64, error) { + if len(specs) == 0 { + return createdCount, eligibleCount, dueCount, 0, 0, nil } publishedTasks, deferredTaskIDs, err := s.dispatchMonitorDesktopTasks(ctx, specs, maxDesktopClientBacklog) @@ -550,7 +568,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( zap.Error(err), ) } - return createdCount, dueCount, 0, 0, err + return createdCount, eligibleCount, dueCount, 0, 0, err } for _, task := range publishedTasks { @@ -576,7 +594,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( cancel() } - return createdCount, dueCount, int64(len(publishedTasks)), int64(len(deferredTaskIDs)), nil + return createdCount, eligibleCount, dueCount, int64(len(publishedTasks)), int64(len(deferredTaskIDs)), nil } func (s *MonitoringService) loadDailyMonitoringPlans(ctx context.Context) ([]monitoringDailyPlan, error) { @@ -901,6 +919,67 @@ func (s *MonitoringService) loadDailyMonitoringTaskKeys(ctx context.Context, q m return keys, nil } +func (s *MonitoringService) createDispatchableDailyMonitoringTasksForBrand( + ctx context.Context, + projectionService *MonitoringCallbackService, + plan monitoringDailyPlan, + brand monitoringDailyBrand, + businessDay time.Time, + now time.Time, + questions []monitoringConfiguredQuestion, + candidates []monitoringDailyTaskCandidate, + limit int, + executionOwner string, +) (int64, int64, []monitorDesktopTaskSpec, error) { + if limit <= 0 || len(candidates) == 0 { + return 0, 0, nil, nil + } + + tx, err := s.monitoringPool.Begin(ctx) + if err != nil { + return 0, 0, nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring daily task generation") + } + defer tx.Rollback(ctx) + + locked, err := acquireMonitoringDailyBrandLock(ctx, tx, plan.TenantID, brand.BrandID, monitoringBusinessDateText(businessDay)) + if err != nil { + return 0, 0, nil, err + } + if !locked { + return 0, 0, nil, nil + } + + existingKeys, err := s.loadDailyMonitoringTaskKeys(ctx, tx, plan.TenantID, brand.BrandID, businessDay) + if err != nil { + return 0, 0, nil, err + } + dueCandidates := selectDueMonitoringDailyCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, limit) + dueCount := int64(len(dueCandidates)) + if len(dueCandidates) == 0 { + if err := tx.Commit(ctx); err != nil { + return 0, dueCount, nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring daily task generation") + } + return 0, dueCount, nil, nil + } + + if err := s.syncMonitoringQuestionSnapshots(ctx, tx, plan.TenantID, brand.BrandID, questions, true); err != nil { + return 0, dueCount, nil, err + } + createdCount, specs, err := s.ensureDailyMonitoringCollectTasks(ctx, tx, plan, brand, businessDay, dueCandidates, executionOwner) + if err != nil { + return 0, dueCount, nil, err + } + if projectionService != nil && createdCount > 0 { + if err := projectionService.rebuildBrandDailyProjection(ctx, tx, plan.TenantID, brand.BrandID, businessDay); err != nil { + return 0, dueCount, nil, err + } + } + if err := tx.Commit(ctx); err != nil { + return 0, dueCount, nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring daily task generation") + } + return createdCount, dueCount, specs, nil +} + func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs( ctx context.Context, tenantID, workspaceID int64, @@ -941,6 +1020,30 @@ func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs( return reconcileEnabledMonitoringPlatforms(platformIDs), nil } +func monitoringDailyPlatformSet(platformIDs []string) map[string]struct{} { + result := make(map[string]struct{}, len(platformIDs)) + for _, platformID := range platformIDs { + platformID = normalizeMonitoringPlatformID(platformID) + if platformID != "" { + result[platformID] = struct{}{} + } + } + return result +} + +func filterMonitoringDailyCandidatesByPlatform(candidates []monitoringDailyTaskCandidate, platforms map[string]struct{}) []monitoringDailyTaskCandidate { + if len(candidates) == 0 || len(platforms) == 0 { + return nil + } + result := make([]monitoringDailyTaskCandidate, 0, len(candidates)) + for _, candidate := range candidates { + if _, ok := platforms[normalizeMonitoringPlatformID(candidate.Platform.ID)]; ok { + result = append(result, candidate) + } + } + return result +} + func (s *MonitoringService) ensureDailyMonitoringCollectTasks( ctx context.Context, tx pgx.Tx, @@ -949,12 +1052,12 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( businessDay time.Time, candidates []monitoringDailyTaskCandidate, executionOwner string, -) (int64, error) { +) (int64, []monitorDesktopTaskSpec, error) { if len(candidates) == 0 { - return 0, nil + return 0, nil, nil } if strings.TrimSpace(executionOwner) != "desktop_tasks" { - return 0, response.ErrInternal(50041, "invalid_execution_owner", "daily monitoring tasks require desktop task execution") + return 0, nil, response.ErrInternal(50041, "invalid_execution_owner", "daily monitoring tasks require desktop task execution") } businessDate := monitoringBusinessDateText(businessDay) @@ -969,7 +1072,7 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( for _, candidate := range candidates { questionText, textErr := monitoringQuestionTextSnapshot(candidate.Question) if textErr != nil { - return 0, textErr + return 0, nil, textErr } dispatchAfter := candidate.DispatchAfter if dispatchAfter.IsZero() { @@ -982,47 +1085,90 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( dispatchAfterValues = append(dispatchAfterValues, dispatchAfter) } - tag, err := tx.Exec(ctx, ` - INSERT INTO monitoring_collect_tasks ( - tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id, - collector_type, trigger_source, run_mode, business_date, planned_at, status, - dispatch_priority, dispatch_lane, target_client_id, dispatch_after, - ingest_shard_key, execution_owner + rows, err := tx.Query(ctx, ` + WITH input AS ( + SELECT * + FROM unnest( + $7::bigint[], + $8::bytea[], + $9::text[], + $10::text[], + $11::timestamptz[] + ) AS item(question_id, question_hash, question_text, ai_platform_id, dispatch_after) + ), + upserted AS ( + INSERT INTO monitoring_collect_tasks ( + tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id, + collector_type, trigger_source, run_mode, business_date, planned_at, status, + dispatch_priority, dispatch_lane, target_client_id, dispatch_after, + ingest_shard_key, execution_owner + ) + SELECT + $1, + $2, + input.question_id, + input.question_hash, + input.question_text, + input.ai_platform_id, + $3, + 'automatic', + 'desktop_standard', + $4::date, + input.dispatch_after, + 'pending', + 100, + 'normal', + NULL, + input.dispatch_after, + $5, + $6 + FROM input + ON CONFLICT ( + tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date + ) + DO UPDATE SET + question_hash = EXCLUDED.question_hash, + question_text_snapshot = EXCLUDED.question_text_snapshot, + dispatch_after = LEAST( + COALESCE(monitoring_collect_tasks.dispatch_after, EXCLUDED.dispatch_after), + EXCLUDED.dispatch_after + ), + updated_at = NOW() + WHERE monitoring_collect_tasks.status = 'pending' + AND monitoring_collect_tasks.callback_received_at IS NULL + RETURNING * + ), + claimed AS ( + UPDATE monitoring_collect_tasks t + SET last_dispatched_at = NOW(), + dispatch_attempts = COALESCE(t.dispatch_attempts, 0) + 1, + updated_at = NOW() + FROM upserted u + WHERE t.id = u.id + AND t.status = 'pending' + AND t.callback_received_at IS NULL + AND (t.dispatch_after IS NULL OR t.dispatch_after <= NOW()) + AND ( + t.last_dispatched_at IS NULL + OR t.last_dispatched_at <= NOW() - $12::interval + ) + RETURNING t.* ) SELECT - $1, - $2, - input.question_id, - input.question_hash, - input.question_text, - input.ai_platform_id, - $3, - 'automatic', - 'desktop_standard', - $4::date, - input.dispatch_after, - 'pending', - 100, - 'normal', - NULL, - input.dispatch_after, - $5, - $6 - FROM unnest( - $7::bigint[], - $8::bytea[], - $9::text[], - $10::text[], - $11::timestamptz[] - ) AS input(question_id, question_hash, question_text, ai_platform_id, dispatch_after) - ON CONFLICT ( - tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date - ) - DO UPDATE SET - question_hash = EXCLUDED.question_hash, - question_text_snapshot = EXCLUDED.question_text_snapshot, - updated_at = NOW() - WHERE btrim(COALESCE(monitoring_collect_tasks.question_text_snapshot, '')) = '' + t.id, + t.tenant_id, + t.ai_platform_id, + t.business_date, + t.trigger_source, + t.run_mode, + t.question_id, + t.question_hash, + t.dispatch_priority, + t.dispatch_lane, + t.interrupt_generation, + t.question_text_snapshot + FROM claimed t + ORDER BY t.dispatch_priority DESC, t.id ASC `, plan.TenantID, brand.BrandID, @@ -1035,12 +1181,50 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( questionTexts, platformIDs, dispatchAfterValues, + formatPgInterval(defaultMonitoringDailyDesktopRetryCooldown), ) if err != nil { - return 0, response.ErrInternal(50041, "insert_failed", "failed to create daily monitoring tasks") + return 0, nil, response.ErrInternal(50041, "insert_failed", "failed to create daily monitoring tasks") } + defer rows.Close() - return tag.RowsAffected(), nil + specs := make([]monitorDesktopTaskSpec, 0, len(candidates)) + for rows.Next() { + var ( + spec monitorDesktopTaskSpec + questionHash []byte + businessDay time.Time + ) + if scanErr := rows.Scan( + &spec.MonitorTaskID, + &spec.TenantID, + &spec.PlatformID, + &businessDay, + &spec.TriggerSource, + &spec.RunMode, + &spec.QuestionID, + &questionHash, + &spec.Priority, + &spec.Lane, + &spec.InterruptGeneration, + &spec.QuestionText, + ); scanErr != nil { + return 0, nil, response.ErrInternal(50041, "scan_failed", "failed to parse created daily monitoring tasks") + } + spec.WorkspaceID = plan.WorkspaceID + spec.BusinessDate = businessDay.Format("2006-01-02") + spec.QuestionHash = encodeQuestionHash(questionHash) + spec.QuestionText = strings.TrimSpace(spec.QuestionText) + if spec.QuestionText == "" { + continue + } + spec.SchedulerGroupKey = monitoringSchedulerGroupKey(spec.QuestionHash, spec.QuestionID, spec.QuestionText) + specs = append(specs, spec) + } + if err := rows.Err(); err != nil { + return 0, nil, response.ErrInternal(50041, "scan_failed", "failed to iterate created daily monitoring tasks") + } + return int64(len(specs)), specs, nil } func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs( 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 045438c..cd5a7f9 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -225,10 +225,10 @@ func TestEnsureDailyMonitoringCollectTasksPersistsQuestionTextSnapshot(t *testin tx := &captureDailyMonitorTaskTx{commandTag: pgconn.NewCommandTag("INSERT 0 2")} businessDay := time.Date(2026, 6, 22, 0, 0, 0, 0, time.UTC) - created, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks( + created, specs, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks( ctx, tx, - monitoringDailyPlan{TenantID: 4}, + monitoringDailyPlan{TenantID: 4, WorkspaceID: 8}, monitoringDailyBrand{BrandID: 8}, businessDay, []monitoringDailyTaskCandidate{ @@ -253,23 +253,27 @@ func TestEnsureDailyMonitoringCollectTasksPersistsQuestionTextSnapshot(t *testin ) require.NoError(t, err) - assert.Equal(t, int64(2), created) - assert.True(t, tx.execCalled) - normalizedSQL := normalizeSQLWhitespace(tx.execSQL) + assert.Equal(t, int64(0), created) + assert.Empty(t, specs) + assert.True(t, tx.queryCalled) + normalizedSQL := normalizeSQLWhitespace(tx.querySQL) assert.Contains(t, normalizedSQL, "question_text_snapshot, ai_platform_id") assert.Contains(t, normalizedSQL, "$10::text[]") - assert.Contains(t, normalizedSQL, "AS input(question_id, question_hash, question_text, ai_platform_id, dispatch_after)") + assert.Contains(t, normalizedSQL, "AS item(question_id, question_hash, question_text, ai_platform_id, dispatch_after)") assert.Contains(t, normalizedSQL, "question_text_snapshot = EXCLUDED.question_text_snapshot") - require.Len(t, tx.execArgs, 11) - assert.Equal(t, []string{"幽灵门轨道长度定制", "幽灵门轨道长度定制"}, tx.execArgs[8]) - assert.Equal(t, []string{"qwen", "doubao"}, tx.execArgs[9]) + assert.Contains(t, normalizedSQL, "WHERE monitoring_collect_tasks.status = 'pending'") + assert.Contains(t, normalizedSQL, "t.dispatch_after IS NULL OR t.dispatch_after <= NOW()") + assert.Contains(t, normalizedSQL, "SET last_dispatched_at = NOW()") + require.Len(t, tx.queryArgs, 12) + assert.Equal(t, []string{"幽灵门轨道长度定制", "幽灵门轨道长度定制"}, tx.queryArgs[8]) + assert.Equal(t, []string{"qwen", "doubao"}, tx.queryArgs[9]) } func TestEnsureDailyMonitoringCollectTasksRejectsEmptyQuestionText(t *testing.T) { ctx := context.Background() tx := &captureDailyMonitorTaskTx{} - created, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks( + created, specs, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks( ctx, tx, monitoringDailyPlan{TenantID: 4}, @@ -290,10 +294,26 @@ func TestEnsureDailyMonitoringCollectTasksRejectsEmptyQuestionText(t *testing.T) require.Error(t, err) assert.Equal(t, int64(0), created) + assert.Empty(t, specs) assert.False(t, tx.execCalled) + assert.False(t, tx.queryCalled) assert.Contains(t, err.Error(), "missing_question_text_snapshot") } +func TestFilterMonitoringDailyCandidatesByPlatformKeepsOnlyOnlineDispatchablePlatforms(t *testing.T) { + candidates := []monitoringDailyTaskCandidate{ + {Question: monitoringConfiguredQuestion{ID: 1}, Platform: monitoringPlatformMetadata{ID: "qwen"}}, + {Question: monitoringConfiguredQuestion{ID: 1}, Platform: monitoringPlatformMetadata{ID: "doubao"}}, + {Question: monitoringConfiguredQuestion{ID: 2}, Platform: monitoringPlatformMetadata{ID: "kimi"}}, + } + + filtered := filterMonitoringDailyCandidatesByPlatform(candidates, monitoringDailyPlatformSet([]string{"qwen", "kimi"})) + + require.Len(t, filtered, 2) + assert.Equal(t, "qwen", filtered[0].Platform.ID) + assert.Equal(t, "kimi", filtered[1].Platform.ID) +} + func TestDecodeDailyMonitoringEnabledPlatformsReportsMalformedJSON(t *testing.T) { platforms, err := decodeDailyMonitoringEnabledPlatforms([]byte(`["qwen"`)) require.Error(t, err) @@ -399,6 +419,7 @@ func TestMonitoringDailyTaskMetricsRecordsSummaryCounters(t *testing.T) { FailedPlanCount: 1, FailedBrandCount: 2, PlannedTaskCount: 20, + EligibleTaskCount: 12, DueTaskCount: 8, CreatedTaskCount: 6, DesktopTaskCount: 4, @@ -415,6 +436,7 @@ func TestMonitoringDailyTaskMetricsRecordsSummaryCounters(t *testing.T) { assert.Equal(t, int64(2), snapshot.BrandFailureTotal) assert.Equal(t, int64(1), snapshot.SkippedPlanTotal) assert.Equal(t, int64(20), snapshot.PlannedTaskTotal) + assert.Equal(t, int64(12), snapshot.EligibleTaskTotal) assert.Equal(t, int64(8), snapshot.DueTaskTotal) assert.Equal(t, int64(6), snapshot.CreatedTaskTotal) assert.Equal(t, int64(4), snapshot.DesktopTaskTotal) @@ -428,6 +450,7 @@ func TestMonitoringDailyTaskMetricsRecordsSummaryCounters(t *testing.T) { prometheus := MonitoringDailyTaskMetricsPrometheus() assert.Contains(t, prometheus, `monitoring_daily_task_worker_runs_total{status="completed"} 1`) assert.Contains(t, prometheus, `monitoring_daily_task_generation_failures_total{scope="plan"} 1`) + assert.Contains(t, prometheus, `monitoring_daily_task_generation_tasks_total{stage="eligible"} 12`) assert.Contains(t, prometheus, `monitoring_daily_task_generation_tasks_total{stage="created"} 6`) assert.Contains(t, prometheus, `monitoring_daily_task_worker_last_business_date_info{business_date="2026-04-24"} 1`) } @@ -568,12 +591,15 @@ func (r emptyDailyMonitorRows) RawValues() [][]byte { r func (r emptyDailyMonitorRows) Conn() *pgx.Conn { return nil } type captureDailyMonitorTaskTx struct { - execCalled bool - execSQL string - execArgs []any - execSQLs []string - execArgses [][]any - commandTag pgconn.CommandTag + execCalled bool + execSQL string + execArgs []any + execSQLs []string + execArgses [][]any + queryCalled bool + querySQL string + queryArgs []any + commandTag pgconn.CommandTag } func (tx *captureDailyMonitorTaskTx) Begin(context.Context) (pgx.Tx, error) { return tx, nil } @@ -597,7 +623,10 @@ func (tx *captureDailyMonitorTaskTx) Exec(_ context.Context, sql string, argumen tx.execArgses = append(tx.execArgses, arguments) return tx.commandTag, nil } -func (tx *captureDailyMonitorTaskTx) Query(context.Context, string, ...any) (pgx.Rows, error) { +func (tx *captureDailyMonitorTaskTx) Query(_ context.Context, sql string, arguments ...any) (pgx.Rows, error) { + tx.queryCalled = true + tx.querySQL = sql + tx.queryArgs = arguments return emptyDailyMonitorRows{}, nil } func (tx *captureDailyMonitorTaskTx) QueryRow(context.Context, string, ...any) pgx.Row { diff --git a/server/internal/tenant/app/monitoring_metrics_prometheus.go b/server/internal/tenant/app/monitoring_metrics_prometheus.go index 41701ad..98fbf72 100644 --- a/server/internal/tenant/app/monitoring_metrics_prometheus.go +++ b/server/internal/tenant/app/monitoring_metrics_prometheus.go @@ -98,6 +98,7 @@ func (monitoringDailyTaskMetricsCollector) Collect(ch chan<- prometheus.Metric) monitoringCounter(ch, "monitoring_daily_task_generation_failures_total", "Total non-fatal daily monitoring generation failures by scope.", float64(snapshot.BrandFailureTotal), []string{"scope"}, "brand") monitoringCounter(ch, "monitoring_daily_task_generation_skipped_plans_total", "Total daily monitoring plans skipped before generation.", float64(snapshot.SkippedPlanTotal), nil) monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.PlannedTaskTotal), []string{"stage"}, "planned") + monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.EligibleTaskTotal), []string{"stage"}, "eligible") monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.DueTaskTotal), []string{"stage"}, "due") monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.CreatedTaskTotal), []string{"stage"}, "created") monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.DesktopTaskTotal), []string{"stage"}, "desktop") diff --git a/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql b/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql index ef51587..f21e641 100644 --- a/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql +++ b/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql @@ -93,7 +93,7 @@ INSERT INTO ops.scheduler_jobs (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config) VALUES ('schedule_dispatch', '内容定时任务派发', 'content', '索引驱动的延迟队列派发，仅认领到期内容任务并投递生成队列。', true, 'interval', 15, 'Asia/Shanghai', 30, 100, '{}'::jsonb), - ('monitoring_daily_task', 'AI 监控每日任务生成', 'monitoring', '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。', true, 'interval', 300, 'Asia/Shanghai', 45, 12, '{"max_desktop_client_backlog":12}'::jsonb), + ('monitoring_daily_task', 'AI 监控每日任务生成', 'monitoring', '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。', true, 'interval', 300, 'Asia/Shanghai', 45, 12, '{"max_desktop_client_backlog":12}'::jsonb), ('monitoring_result_recovery', '监控结果恢复', 'monitoring', '按回调结果队列索引恢复未成功入队的监控结果。', true, 'interval', 60, 'Asia/Shanghai', 30, 100, '{}'::jsonb), ('monitoring_lease_recovery', '监控租约回收', 'monitoring', '按过期租约索引限批回收超过租约时间仍未回调的监控任务。', true, 'interval', 1800, 'Asia/Shanghai', 30, 1000, '{}'::jsonb), ('monitoring_received_inspection', '监控 received 卡死检查', 'monitoring', '按 received watchdog 索引限批标记长时间未入队的监控任务。', true, 'interval', 300, 'Asia/Shanghai', 30, 100, '{"threshold":"15m"}'::jsonb), diff --git a/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql b/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql index bfd9d4b..de586e9 100644 --- a/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql +++ b/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql @@ -11,7 +11,7 @@ SET description = '索引驱动的延迟队列派发，仅认领到期内容任 WHERE job_key = 'schedule_dispatch'; UPDATE ops.scheduler_jobs -SET description = '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。', +SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。', timeout_seconds = 45, batch_size = 12, config = (config - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand') diff --git a/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql b/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql index 0689632..78d4b66 100644 --- a/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql +++ b/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql @@ -1,5 +1,5 @@ UPDATE ops.scheduler_jobs -SET description = '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。', +SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。', batch_size = 12, config = (config - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand') || '{"max_desktop_client_backlog":12}'::jsonb, diff --git a/server/migrations_ops/20260627184500_document_online_only_monitoring_daily.down.sql b/server/migrations_ops/20260627184500_document_online_only_monitoring_daily.down.sql new file mode 100644 index 0000000..7b1267b --- /dev/null +++ b/server/migrations_ops/20260627184500_document_online_only_monitoring_daily.down.sql @@ -0,0 +1,6 @@ +UPDATE ops.scheduler_jobs +SET description = '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。', + batch_size = 12, + config = config || '{"max_desktop_client_backlog":12}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_daily_task'; diff --git a/server/migrations_ops/20260627184500_document_online_only_monitoring_daily.up.sql b/server/migrations_ops/20260627184500_document_online_only_monitoring_daily.up.sql new file mode 100644 index 0000000..78d4b66 --- /dev/null +++ b/server/migrations_ops/20260627184500_document_online_only_monitoring_daily.up.sql @@ -0,0 +1,7 @@ +UPDATE ops.scheduler_jobs +SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。', + batch_size = 12, + config = (config - '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';