diff --git a/seaweed-worker/crates/core/src/stream.rs b/seaweed-worker/crates/core/src/stream.rs index 63eeeea696..8fe3540e64 100644 --- a/seaweed-worker/crates/core/src/stream.rs +++ b/seaweed-worker/crates/core/src/stream.rs @@ -388,6 +388,8 @@ fn spawn_detection( success: false, error_message: format!("{err:#}"), total_proposals: 0, + skipped_reason: String::new(), + eligible_after: None, }, )); "failed" diff --git a/seaweed-worker/crates/lance/src/jobs/cleanup.rs b/seaweed-worker/crates/lance/src/jobs/cleanup.rs index 8b84c4a738..c8c4a3ec36 100644 --- a/seaweed-worker/crates/lance/src/jobs/cleanup.rs +++ b/seaweed-worker/crates/lance/src/jobs/cleanup.rs @@ -210,6 +210,8 @@ impl JobHandler for CleanupVersionsHandler { success: true, error_message: String::new(), total_proposals: total, + skipped_reason: String::new(), + eligible_after: None, })?; Ok(()) } diff --git a/seaweed-worker/crates/lance/src/jobs/compact.rs b/seaweed-worker/crates/lance/src/jobs/compact.rs index da3339add4..728437420d 100644 --- a/seaweed-worker/crates/lance/src/jobs/compact.rs +++ b/seaweed-worker/crates/lance/src/jobs/compact.rs @@ -232,6 +232,8 @@ impl JobHandler for CompactHandler { success: true, error_message: String::new(), total_proposals: total, + skipped_reason: String::new(), + eligible_after: None, })?; Ok(()) } diff --git a/seaweed-worker/crates/lance/src/jobs/indices.rs b/seaweed-worker/crates/lance/src/jobs/indices.rs index 5ca17224b3..e619ff43d5 100644 --- a/seaweed-worker/crates/lance/src/jobs/indices.rs +++ b/seaweed-worker/crates/lance/src/jobs/indices.rs @@ -209,6 +209,8 @@ impl JobHandler for OptimizeIndicesHandler { success: true, error_message: String::new(), total_proposals: total, + skipped_reason: String::new(), + eligible_after: None, })?; Ok(()) } diff --git a/seaweed-worker/crates/lance/src/jobs/sort.rs b/seaweed-worker/crates/lance/src/jobs/sort.rs index 2e48eab0ae..ed8aeaa7ba 100644 --- a/seaweed-worker/crates/lance/src/jobs/sort.rs +++ b/seaweed-worker/crates/lance/src/jobs/sort.rs @@ -310,6 +310,8 @@ impl JobHandler for SortHandler { success: true, error_message: String::new(), total_proposals: total, + skipped_reason: String::new(), + eligible_after: None, })?; Ok(()) } diff --git a/weed/admin/dash/plugin_api.go b/weed/admin/dash/plugin_api.go index 934f8db7c4..93821b5c4d 100644 --- a/weed/admin/dash/plugin_api.go +++ b/weed/admin/dash/plugin_api.go @@ -604,6 +604,12 @@ func (s *AdminServer) TriggerPluginDetectionAPI(w http.ResponseWriter, r *http.R "proposals": proposalPayloads, "activities": filteredActivities, } + if report != nil && report.Complete != nil { + response["skipped_reason"] = report.Complete.SkippedReason + if at := report.Complete.EligibleAfter; at != nil { + response["eligible_after"] = at.AsTime() + } + } if err != nil { response["error"] = err.Error() diff --git a/weed/admin/plugin/config_store.go b/weed/admin/plugin/config_store.go index d6d0233e5d..ca223505e1 100644 --- a/weed/admin/plugin/config_store.go +++ b/weed/admin/plugin/config_store.go @@ -109,6 +109,10 @@ func (s *ConfigStore) SaveDescriptor(jobType string, descriptor *plugin_pb.JobTy s.mu.Lock() defer s.mu.Unlock() + current, err := s.loadDescriptor(jobType) + if err == nil && current != nil && current.DescriptorVersion > clone.DescriptorVersion { + return nil + } if !s.configured { s.memDescriptors[jobType] = clone return nil @@ -135,15 +139,18 @@ func (s *ConfigStore) LoadDescriptor(jobType string) (*plugin_pb.JobTypeDescript } s.mu.RLock() + defer s.mu.RUnlock() + return s.loadDescriptor(jobType) +} + +func (s *ConfigStore) loadDescriptor(jobType string) (*plugin_pb.JobTypeDescriptor, error) { if !s.configured { d := s.memDescriptors[jobType] - s.mu.RUnlock() if d == nil { return nil, nil } return proto.Clone(d).(*plugin_pb.JobTypeDescriptor), nil } - s.mu.RUnlock() pbPath := filepath.Join(s.baseDir, jobTypesDirName, jobType, descriptorPBFileName) data, err := os.ReadFile(pbPath) diff --git a/weed/admin/plugin/config_store_test.go b/weed/admin/plugin/config_store_test.go index c45c45a45f..00824de145 100644 --- a/weed/admin/plugin/config_store_test.go +++ b/weed/admin/plugin/config_store_test.go @@ -1,13 +1,46 @@ package plugin import ( + "os" + "path/filepath" "reflect" "testing" "time" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/stretchr/testify/require" ) +func TestConfigStoreRefreshRepairsCorruptDescriptor(t *testing.T) { + dir := t.TempDir() + store, err := NewConfigStore(dir) + require.NoError(t, err) + descriptor := &plugin_pb.JobTypeDescriptor{JobType: "vacuum", DescriptorVersion: 2, DetectionCooldownConfigField: "cooldown"} + require.NoError(t, store.SaveDescriptor("vacuum", descriptor)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "plugin", "job_types", "vacuum", "descriptor.pb"), []byte{0xff}, 0600)) + require.NoError(t, store.SaveDescriptor("vacuum", descriptor)) + got, err := store.LoadDescriptor("vacuum") + require.NoError(t, err) + require.Equal(t, descriptor.DescriptorVersion, got.DescriptorVersion) + require.Equal(t, "cooldown", got.DetectionCooldownConfigField) +} + +func TestConfigStoreDoesNotDowngradeDescriptors(t *testing.T) { + for _, dir := range []string{"", t.TempDir()} { + store, err := NewConfigStore(dir) + require.NoError(t, err) + require.NoError(t, store.SaveDescriptor("vacuum", &plugin_pb.JobTypeDescriptor{ + JobType: "vacuum", DescriptorVersion: 2, DetectionCooldownConfigField: "cooldown", + })) + require.NoError(t, store.SaveDescriptor("vacuum", &plugin_pb.JobTypeDescriptor{ + JobType: "vacuum", DescriptorVersion: 1, + })) + got, err := store.LoadDescriptor("vacuum") + require.NoError(t, err) + require.Equal(t, uint32(2), got.DescriptorVersion) + require.Equal(t, "cooldown", got.DetectionCooldownConfigField) + } +} func TestConfigStoreDescriptorRoundTrip(t *testing.T) { t.Parallel() diff --git a/weed/admin/plugin/plugin.go b/weed/admin/plugin/plugin.go index fda1a86586..9082cdc4b5 100644 --- a/weed/admin/plugin/plugin.go +++ b/weed/admin/plugin/plugin.go @@ -361,7 +361,10 @@ func (r *Plugin) RequestConfigSchema(ctx context.Context, jobType string, forceR if err != nil { return nil, err } + return r.requestConfigSchema(ctx, jobType, forceRefresh, provider.WorkerID) +} +func (r *Plugin) requestConfigSchema(ctx context.Context, jobType string, forceRefresh bool, workerID string) (*plugin_pb.JobTypeDescriptor, error) { requestID, err := newRequestID("schema") if err != nil { return nil, err @@ -388,7 +391,7 @@ func (r *Plugin) RequestConfigSchema(ctx context.Context, jobType string, forceR }, } - if err := r.sendToWorker(provider.WorkerID, requestMessage); err != nil { + if err := r.sendToWorker(workerID, requestMessage); err != nil { return nil, err } @@ -1230,6 +1233,19 @@ func (r *Plugin) handleDetectionComplete(workerID string, message *plugin_pb.Det if messageText == "" { messageText = "detection failed" } + } else if message.SkippedReason != "" { + stage = "skipped" + messageText = message.SkippedReason + } + details := map[string]interface{}{ + "success": message.Success, + "total_proposals": message.TotalProposals, + } + if message.SkippedReason != "" { + details["skipped_reason"] = message.SkippedReason + } + if message.EligibleAfter != nil { + details["eligible_after"] = message.EligibleAfter.AsTime().UTC().Format(time.RFC3339) } r.appendActivity(JobActivity{ JobType: resolvedJobType, @@ -1239,10 +1255,7 @@ func (r *Plugin) handleDetectionComplete(workerID string, message *plugin_pb.Det Stage: stage, Message: messageText, OccurredAt: timeToPtr(time.Now().UTC()), - Details: map[string]interface{}{ - "success": message.Success, - "total_proposals": message.TotalProposals, - }, + Details: details, }) r.safeSendDetectionComplete(message.RequestId, message) diff --git a/weed/admin/plugin/plugin_detection_test.go b/weed/admin/plugin/plugin_detection_test.go index ee86c353a7..bf18fc810b 100644 --- a/weed/admin/plugin/plugin_detection_test.go +++ b/weed/admin/plugin/plugin_detection_test.go @@ -6,8 +6,112 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/stretchr/testify/require" ) +func TestScheduledDetectionDistinguishesSkipFromEmptyScan(t *testing.T) { + for _, reason := range []string{"", "Cooldown after successful job (10h0m0s)"} { + t.Run(reason, func(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + p.clusterContextProvider = func(context.Context) (*plugin_pb.ClusterContext, error) { + return &plugin_pb.ClusterContext{}, nil + } + p.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-a", Capabilities: []*plugin_pb.JobTypeCapability{{JobType: "vacuum", CanDetect: true}}, + }) + session := &streamSession{workerID: "worker-a", outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 1), done: make(chan struct{})} + p.putSession(session) + done := make(chan struct{}) + go func() { + p.runJobTypeIteration("vacuum", schedulerPolicy{DetectionTimeout: time.Second, JobTypeMaxRuntime: time.Second}) + close(done) + }() + select { + case request := <-session.outgoing: + p.handleDetectionComplete("worker-a", &plugin_pb.DetectionComplete{ + RequestId: request.RequestId, JobType: "vacuum", Success: true, SkippedReason: reason, + }) + case <-time.After(2 * time.Second): + t.Fatal("no detection request") + } + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("detection did not complete") + } + info := p.snapshotSchedulerDetection("vacuum") + if reason == "" { + require.False(t, info.lastDetectedAt.IsZero()) + require.Equal(t, "success", p.snapshotSchedulerRun("vacuum").lastRunStatus) + } else { + require.True(t, info.lastDetectedAt.IsZero(), "a skipped scan must not count as an empty scan") + require.Equal(t, reason, info.lastSkippedReason) + require.Equal(t, "skipped", p.snapshotSchedulerRun("vacuum").lastRunStatus) + } + }) + } +} + +func TestManualDetectionSkipDoesNotChangeScheduledStatus(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + p.recordSchedulerDetectionSuccess("vacuum", 5) + p.pendingDetection["manual"] = &pendingDetectionState{ + jobType: "vacuum", workerID: "worker-a", complete: make(chan *plugin_pb.DetectionComplete, 1), + } + p.handleDetectionComplete("worker-a", &plugin_pb.DetectionComplete{ + RequestId: "manual", JobType: "vacuum", Success: true, SkippedReason: "Cooldown", + }) + require.Empty(t, p.snapshotSchedulerDetection("vacuum").lastSkippedReason) + require.Equal(t, 5, p.snapshotSchedulerDetection("vacuum").lastDetectedCount) + activities := p.ListActivities("vacuum", 10) + require.Len(t, activities, 1) + require.Equal(t, "skipped", activities[0].Stage) +} + +func TestPrefetchRefreshesSavedDescriptorFromReconnectingWorker(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + require.NoError(t, p.store.SaveDescriptor("vacuum", &plugin_pb.JobTypeDescriptor{JobType: "vacuum", DescriptorVersion: 1})) + require.NoError(t, p.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{ + JobType: "vacuum", WorkerConfigValues: map[string]*plugin_pb.ConfigValue{"cooldown": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 600}}}, + })) + session := &streamSession{workerID: "upgraded", outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 1), done: make(chan struct{})} + p.putSession(session) + done := make(chan struct{}) + go func() { + p.prefetchDescriptorsFromHello(&plugin_pb.WorkerHello{ + WorkerId: "upgraded", Capabilities: []*plugin_pb.JobTypeCapability{{JobType: "vacuum", CanDetect: true}}, + }) + close(done) + }() + select { + case request := <-session.outgoing: + require.True(t, request.GetRequestConfigSchema().GetForceRefresh()) + p.handleConfigSchemaResponse(&plugin_pb.ConfigSchemaResponse{ + RequestId: request.RequestId, JobType: "vacuum", Success: true, + JobTypeDescriptor: &plugin_pb.JobTypeDescriptor{JobType: "vacuum", DescriptorVersion: 2, DetectionCooldownConfigField: "cooldown"}, + }) + case <-time.After(time.Second): + t.Fatal("saved descriptor was not refreshed") + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("prefetch did not finish") + } + got, err := p.store.LoadDescriptor("vacuum") + require.NoError(t, err) + require.Equal(t, "cooldown", got.DetectionCooldownConfigField) + cfg, err := p.LoadJobTypeConfig("vacuum") + require.NoError(t, err) + require.Equal(t, int64(600), cfg.WorkerConfigValues["cooldown"].GetInt64Value()) +} + func TestRunDetectionIncludesLatestSuccessfulRun(t *testing.T) { pluginSvc, err := New(Options{}) if err != nil { diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index 58ad878d6a..759581e193 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -343,7 +343,7 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo } detectCtx, cancelDetect := context.WithTimeout(jobCtx, detectionTimeout) - proposals, err := r.RunDetection(detectCtx, jobType, clusterContext, policy.MaxResults) + report, err := r.RunDetectionWithReport(detectCtx, jobType, clusterContext, policy.MaxResults) cancelDetect() releaseDetectionLock() if err != nil { @@ -365,6 +365,12 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo return false } + if report.Complete.GetSkippedReason() != "" { + r.recordSchedulerDetectionSkip(jobType, report.Complete.GetSkippedReason()) + r.recordSchedulerRunComplete(jobType, "skipped") + return false + } + proposals := report.Proposals r.appendActivity(JobActivity{ JobType: jobType, Source: "admin_scheduler", diff --git a/weed/admin/plugin/plugin_schema_prefetch.go b/weed/admin/plugin/plugin_schema_prefetch.go index 13816f5d86..493db02f28 100644 --- a/weed/admin/plugin/plugin_schema_prefetch.go +++ b/weed/admin/plugin/plugin_schema_prefetch.go @@ -12,7 +12,7 @@ import ( const descriptorPrefetchTimeout = 20 * time.Second func (r *Plugin) prefetchDescriptorsFromHello(hello *plugin_pb.WorkerHello) { - if hello == nil || len(hello.Capabilities) == 0 { + if hello == nil || hello.WorkerId == "" || len(hello.Capabilities) == 0 { return } @@ -44,17 +44,8 @@ func (r *Plugin) prefetchDescriptorsFromHello(hello *plugin_pb.WorkerHello) { default: } - descriptor, err := r.store.LoadDescriptor(jobType) - if err != nil { - glog.Warningf("Plugin descriptor prefetch check failed for %s: %v", jobType, err) - continue - } - if descriptor != nil { - continue - } - ctx, cancel := context.WithTimeout(r.ctx, descriptorPrefetchTimeout) - _, err = r.RequestConfigSchema(ctx, jobType, false) + _, err := r.requestConfigSchema(ctx, jobType, true, hello.WorkerId) cancel() if err != nil { glog.V(1).Infof("Plugin descriptor prefetch skipped for %s: %v", jobType, err) diff --git a/weed/admin/plugin/scheduler_status.go b/weed/admin/plugin/scheduler_status.go index 7359d73449..af2710e43f 100644 --- a/weed/admin/plugin/scheduler_status.go +++ b/weed/admin/plugin/scheduler_status.go @@ -1,9 +1,12 @@ package plugin import ( + "fmt" "sort" "strings" "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" ) type SchedulerStatus struct { @@ -52,6 +55,66 @@ type SchedulerJobTypeStatus struct { LastDetectedCount int `json:"last_detected_count,omitempty"` LastDetectionError string `json:"last_detection_error,omitempty"` LastDetectionSkipped string `json:"last_detection_skipped,omitempty"` + CooldownSeconds int64 `json:"cooldown_seconds,omitempty"` + LastSuccessfulRunAt *time.Time `json:"last_successful_run_at,omitempty"` + DetectionEligibleAt *time.Time `json:"detection_eligible_at,omitempty"` + CooldownError string `json:"cooldown_error,omitempty"` +} + +func (r *Plugin) populateSchedulerCooldown(status *SchedulerJobTypeStatus) { + descriptor, err := r.store.LoadDescriptor(status.JobType) + if err != nil { + status.CooldownError = "Cannot load cooldown policy" + return + } + field := descriptor.GetDetectionCooldownConfigField() + if field == "" { + return + } + _, _, values, err := r.loadJobTypeConfigPayload(status.JobType) + if err != nil { + status.CooldownError = "Cannot load cooldown configuration" + return + } + value := values[field] + if value == nil { + value = descriptor.GetWorkerDefaultValues()[field] + } + if value == nil { + value = descriptor.GetWorkerConfigForm().GetDefaultValues()[field] + } + if value == nil { + status.CooldownError = "Cooldown configuration is missing" + return + } + if _, ok := value.Kind.(*plugin_pb.ConfigValue_Int64Value); !ok { + status.CooldownError = "Cooldown must be an integer number of seconds" + return + } + seconds := value.GetInt64Value() + if seconds < 0 || seconds > int64((1<<63-1)/time.Second) { + status.CooldownError = "Cooldown is outside the supported range" + return + } + status.CooldownSeconds = seconds + history, err := r.store.LoadRunHistory(status.JobType) + if err != nil { + status.CooldownError = "Cannot load successful job history" + return + } + if history == nil { + return + } + for _, run := range history.SuccessfulRuns { + if run.CompletedAt != nil && (status.LastSuccessfulRunAt == nil || run.CompletedAt.After(*status.LastSuccessfulRunAt)) { + at := *run.CompletedAt + status.LastSuccessfulRunAt = &at + } + } + if status.LastSuccessfulRunAt != nil && seconds > 0 { + at := status.LastSuccessfulRunAt.Add(time.Duration(seconds) * time.Second) + status.DetectionEligibleAt = &at + } } type schedulerDetectionInfo struct { @@ -349,6 +412,7 @@ func (r *Plugin) GetLaneSchedulerStatus(lane SchedulerLane) SchedulerStatus { NextDetectionAt: state.NextDetectionAt, DetectionIntervalMinutes: state.DetectionIntervalMinutes, } + r.populateSchedulerCooldown(&jobStatus) if !info.lastDetectedAt.IsZero() { jobStatus.LastDetectedAt = timeToPtr(info.lastDetectedAt) jobStatus.LastDetectedCount = info.lastDetectedCount @@ -361,6 +425,12 @@ func (r *Plugin) GetLaneSchedulerStatus(lane SchedulerLane) SchedulerStatus { } jobTypes = append(jobTypes, jobStatus) + if jobStatus.Enabled && jobStatus.DetectionEligibleAt != nil && now.Before(*jobStatus.DetectionEligibleAt) { + waiting = append(waiting, SchedulerWaitingStatus{ + Reason: "cooldown", JobType: jobType, Since: jobStatus.LastSuccessfulRunAt, Until: jobStatus.DetectionEligibleAt, + Details: map[string]interface{}{"message": fmt.Sprintf("Cooldown after successful job (%s)", time.Duration(jobStatus.CooldownSeconds)*time.Second)}, + }) + } if state.DetectionInFlight { waiting = append(waiting, SchedulerWaitingStatus{ Reason: "detection_in_flight", @@ -432,6 +502,7 @@ func (r *Plugin) GetSchedulerStatus() SchedulerStatus { NextDetectionAt: state.NextDetectionAt, DetectionIntervalMinutes: state.DetectionIntervalMinutes, } + r.populateSchedulerCooldown(&jobStatus) if !info.lastDetectedAt.IsZero() { jobStatus.LastDetectedAt = timeToPtr(info.lastDetectedAt) jobStatus.LastDetectedCount = info.lastDetectedCount @@ -444,6 +515,12 @@ func (r *Plugin) GetSchedulerStatus() SchedulerStatus { } jobTypes = append(jobTypes, jobStatus) + if jobStatus.Enabled && jobStatus.DetectionEligibleAt != nil && now.Before(*jobStatus.DetectionEligibleAt) { + waiting = append(waiting, SchedulerWaitingStatus{ + Reason: "cooldown", JobType: jobType, Since: jobStatus.LastSuccessfulRunAt, Until: jobStatus.DetectionEligibleAt, + Details: map[string]interface{}{"message": fmt.Sprintf("Cooldown after successful job (%s)", time.Duration(jobStatus.CooldownSeconds)*time.Second)}, + }) + } if state.DetectionInFlight { waiting = append(waiting, SchedulerWaitingStatus{ Reason: "detection_in_flight", diff --git a/weed/admin/plugin/scheduler_status_test.go b/weed/admin/plugin/scheduler_status_test.go index 5a0c8161ad..3218f805ed 100644 --- a/weed/admin/plugin/scheduler_status_test.go +++ b/weed/admin/plugin/scheduler_status_test.go @@ -6,8 +6,54 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/stretchr/testify/require" ) +func TestSchedulerCooldownUsesSavedPolicyAndHistory(t *testing.T) { + p, err := New(Options{}) + require.NoError(t, err) + defer p.Shutdown() + const jobType = "vacuum" + p.registry.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-a", Capabilities: []*plugin_pb.JobTypeCapability{{JobType: jobType, CanDetect: true}}, + }) + require.NoError(t, p.store.SaveDescriptor(jobType, &plugin_pb.JobTypeDescriptor{ + JobType: jobType, DetectionCooldownConfigField: "cooldown", + WorkerDefaultValues: map[string]*plugin_pb.ConfigValue{"cooldown": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 36000}}}, + })) + last := time.Now().UTC().Add(-time.Hour).Truncate(time.Second) + require.NoError(t, p.store.AppendRunRecord(jobType, &JobRunRecord{Outcome: RunOutcomeSuccess, CompletedAt: &last})) + p.markJobTypeInFlight(jobType) + for _, seconds := range []int64{36000, 600, 0, 7200} { + require.NoError(t, p.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{ + JobType: jobType, AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true, DetectionIntervalMinutes: 10}, + WorkerConfigValues: map[string]*plugin_pb.ConfigValue{"cooldown": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: seconds}}}, + })) + for _, status := range []SchedulerStatus{p.GetLaneSchedulerStatus(LaneDefault), p.GetSchedulerStatus()} { + require.Len(t, status.JobTypes, 1) + job := status.JobTypes[0] + require.Empty(t, job.CooldownError) + require.Equal(t, seconds, job.CooldownSeconds) + require.Equal(t, last, *job.LastSuccessfulRunAt) + if seconds == 0 { + require.Nil(t, job.DetectionEligibleAt) + } else { + require.Equal(t, last.Add(time.Duration(seconds)*time.Second), *job.DetectionEligibleAt) + } + var cooldown, inFlight bool + for _, wait := range status.Waiting { + cooldown = cooldown || wait.Reason == "cooldown" + inFlight = inFlight || wait.Reason == "detection_in_flight" + } + require.Equal(t, seconds > 3600, cooldown) + require.True(t, inFlight, "cooldown must not hide another blocker") + } + } + require.NoError(t, p.store.SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{JobType: jobType})) + job := SchedulerJobTypeStatus{JobType: jobType} + p.populateSchedulerCooldown(&job) + require.Equal(t, int64(36000), job.CooldownSeconds, "missing saved value uses descriptor default") +} func TestGetSchedulerStatusIncludesInProcessJobs(t *testing.T) { pluginSvc, err := New(Options{}) if err != nil { diff --git a/weed/admin/view/app/plugin.templ b/weed/admin/view/app/plugin.templ index 4cf3f1911f..166e42106d 100644 --- a/weed/admin/view/app/plugin.templ +++ b/weed/admin/view/app/plugin.templ @@ -1608,7 +1608,7 @@ templ Plugin(page string, initialJob string, lane string) { } function renderSchedulerStatus() { - window.renderPluginSchedulerStatus({scheduler: state.schedulerStatus, lock: state.schedulerLock}); + window.renderPluginSchedulerStatus({scheduler: state.schedulerStatus, lock: state.schedulerLock, job_type: state.activeTopTab === 'overview' ? '' : state.selectedJobType}); var valueNodes = document.querySelectorAll('.plugin-scheduler-next-run'); if (!valueNodes.length) { return; @@ -2803,6 +2803,22 @@ templ Plugin(page string, initialJob string, lane string) { var html = ''; html += '
Detection Run
'; + var skippedActivity = null; + for (var si = 0; si < activities.length; si++) { + if (activities[si] && activities[si].stage === 'skipped_by_interval') skippedActivity = activities[si]; + } + if (data.skipped_reason || skippedActivity) { + var eligibleAfter = data.eligible_after; + if (!eligibleAfter && skippedActivity && skippedActivity.details) { + var detailAfter = skippedActivity.details.eligible_after; + eligibleAfter = (detailAfter && detailAfter.string_value) || detailAfter; + } + html += '
Skipped: ' + escapeHtml(String(data.skipped_reason || skippedActivity.message || 'detection skipped')); + if (eligibleAfter) html += ' · Eligible after: ' + escapeHtml(parseTime(eligibleAfter) || eligibleAfter); + html += '
'; + } else if (!data.error && proposals.length === 0) { + html += '
Scan completed: no eligible volumes or jobs.
'; + } if (requestID) { html += '
request_id: ' + escapeHtml(String(requestID)) + '
'; } @@ -2928,6 +2944,7 @@ templ Plugin(page string, initialJob string, lane string) { document.getElementById('plugin-config-updated-at').textContent = 'Not saved yet'; } state.loadedJobType = jobType; + renderSchedulerStatus(); await loadRunHistory(jobType); diff --git a/weed/admin/view/app/plugin_scheduler_status.templ b/weed/admin/view/app/plugin_scheduler_status.templ index 715498eec1..936d1f58a8 100644 --- a/weed/admin/view/app/plugin_scheduler_status.templ +++ b/weed/admin/view/app/plugin_scheduler_status.templ @@ -9,7 +9,7 @@ templ PluginSchedulerStatus() { root.replaceChildren(); const lock = data.lock || {}; const waiters = lock.waiters || []; - if (!waiters.length) return; + if (waiters.length) { const alert = document.createElement('div'); alert.className = 'alert alert-warning'; const title = document.createElement('strong'); @@ -33,6 +33,33 @@ templ PluginSchedulerStatus() { note.textContent = 'This lane cannot advance until the lock becomes available. A scheduled check time is not a guaranteed start time.'; alert.appendChild(note); root.appendChild(alert); + } + const scheduler = data.scheduler || {}; + const now = new Date(scheduler.now || Date.now()); + (scheduler.job_types || []).forEach(function(job) { + if (data.job_type ? job.job_type !== data.job_type : !job.enabled) return; + const until = job.detection_eligible_at ? new Date(job.detection_eligible_at) : null; + const cooling = until && until > now; + if (!job.cooldown_error && !cooling && !(data.job_type && job.cooldown_seconds)) return; + const box = document.createElement('div'); + box.className = 'alert ' + (job.cooldown_error ? 'alert-warning' : 'alert-info'); + const title = document.createElement('strong'); + title.textContent = job.job_type + ': ' + (job.cooldown_error || (cooling ? 'Cooldown after successful job' : until ? 'Cooldown elapsed' : 'No successful job yet')); + box.appendChild(title); + const line = document.createElement('div'); + const seconds = job.cooldown_seconds || 0; + const duration = seconds >= 3600 ? (seconds / 3600) + ' hours' : seconds >= 60 ? (seconds / 60) + ' minutes' : seconds + ' seconds'; + line.textContent = 'Configured cooldown: ' + duration + ' (' + seconds + ' seconds).'; + if (job.last_successful_run_at) line.textContent += ' Last success: ' + new Date(job.last_successful_run_at).toLocaleString() + '.'; + if (until) line.textContent += ' Eligible after: ' + until.toLocaleString() + '.'; + if (cooling) line.textContent += ' Remaining: ' + Math.ceil((until - now) / 60000) + ' minutes.'; + box.appendChild(line); + const note = document.createElement('div'); + note.className = 'small mt-1'; + note.textContent = 'Applies across this job type. Locks and other jobs in the lane can delay the actual start.'; + box.appendChild(note); + root.appendChild(box); + }); }; } diff --git a/weed/admin/view/app/plugin_scheduler_status_templ.go b/weed/admin/view/app/plugin_scheduler_status_templ.go index 5b85466333..59b564d675 100644 --- a/weed/admin/view/app/plugin_scheduler_status_templ.go +++ b/weed/admin/view/app/plugin_scheduler_status_templ.go @@ -26,7 +26,7 @@ func PluginSchedulerStatus() templ.Component { } ctx = templ.InitializeContext(ctx) ctx = templ.ClearChildren(ctx) - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 1, "
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 1, "
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } diff --git a/weed/admin/view/app/plugin_templ.go b/weed/admin/view/app/plugin_templ.go index 5babafdff5..fbc54d49a8 100644 --- a/weed/admin/view/app/plugin_templ.go +++ b/weed/admin/view/app/plugin_templ.go @@ -109,7 +109,7 @@ func Plugin(page string, initialJob string, lane string) templ.Component { if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 7, "
Workers

0

Active Jobs

0

Activities (recent)

0

Next scheduled check

-

Per Job Type Summary
Job TypeActive JobsRecent Activities
Loading...
Scheduler State
Sequential scheduler with per-job runtime limits
Job TypeEnabledDetectorIn FlightMax RuntimeExec GlobalExec/WorkerExecutor WorkersEffective ExecLast Run
Loading...
Workers
WorkerAddressCapabilitiesLoad
Loading...
Job Type Configuration
Not loaded
Selected Job Type
-
Descriptor
Select a job type to load schema and config.
Admin Config Form
No admin form loaded.
Worker Config Form
No worker form loaded.
Job Scheduling Settings
How often to check for new work.
Per-attempt deadline for one job. Size-aware tasks may extend this automatically.
Next scheduled check
Scheduler
-
Not scheduled
Run History
Keep last 10 success + last 10 errors
Successful Runs
TimeJob IDWorkerDuration
No data
Error Runs
TimeJob IDWorkerError
No data
Detection Results
Run detection to see proposals.
Job Queue
States: pending/assigned/running
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Detection Jobs
Detection activities for selected job type
TimeJob TypeRequest IDWorkerStageSourceMessage
Loading...
Execution Jobs
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Execution Activities
Non-detection events only
TimeJob TypeJob IDSourceStageMessage
Loading...
Job Detail
Select a job to view details.
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 7, "
Workers

0

Active Jobs

0

Activities (recent)

0

Next scheduled check

-

Per Job Type Summary
Job TypeActive JobsRecent Activities
Loading...
Scheduler State
Sequential scheduler with per-job runtime limits
Job TypeEnabledDetectorIn FlightMax RuntimeExec GlobalExec/WorkerExecutor WorkersEffective ExecLast Run
Loading...
Workers
WorkerAddressCapabilitiesLoad
Loading...
Job Type Configuration
Not loaded
Selected Job Type
-
Descriptor
Select a job type to load schema and config.
Admin Config Form
No admin form loaded.
Worker Config Form
No worker form loaded.
Job Scheduling Settings
How often to check for new work.
Per-attempt deadline for one job. Size-aware tasks may extend this automatically.
Next scheduled check
Scheduler
-
Not scheduled
Run History
Keep last 10 success + last 10 errors
Successful Runs
TimeJob IDWorkerDuration
No data
Error Runs
TimeJob IDWorkerError
No data
Detection Results
Run detection to see proposals.
Job Queue
States: pending/assigned/running
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Detection Jobs
Detection activities for selected job type
TimeJob TypeRequest IDWorkerStageSourceMessage
Loading...
Execution Jobs
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Execution Activities
Non-detection events only
TimeJob TypeJob IDSourceStageMessage
Loading...
Job Detail
Select a job to view details.
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } diff --git a/weed/pb/plugin.proto b/weed/pb/plugin.proto index ffef11de56..a0b562c558 100644 --- a/weed/pb/plugin.proto +++ b/weed/pb/plugin.proto @@ -130,6 +130,8 @@ message JobTypeDescriptor { AdminRuntimeDefaults admin_runtime_defaults = 8; map worker_default_values = 9; + // Worker config field containing a cooldown in seconds after the latest successful job. + string detection_cooldown_config_field = 10; } message ConfigForm { @@ -282,6 +284,8 @@ message DetectionComplete { bool success = 3; string error_message = 4; int32 total_proposals = 5; + string skipped_reason = 6; + google.protobuf.Timestamp eligible_after = 7; } message JobProposal { diff --git a/weed/pb/plugin_pb/plugin.pb.go b/weed/pb/plugin_pb/plugin.pb.go index dd25056889..0652d87d7f 100644 --- a/weed/pb/plugin_pb/plugin.pb.go +++ b/weed/pb/plugin_pb/plugin.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v6.33.4 +// protoc v7.35.0 // source: plugin.proto package plugin_pb @@ -1490,8 +1490,10 @@ type JobTypeDescriptor struct { WorkerConfigForm *ConfigForm `protobuf:"bytes,7,opt,name=worker_config_form,json=workerConfigForm,proto3" json:"worker_config_form,omitempty"` AdminRuntimeDefaults *AdminRuntimeDefaults `protobuf:"bytes,8,opt,name=admin_runtime_defaults,json=adminRuntimeDefaults,proto3" json:"admin_runtime_defaults,omitempty"` WorkerDefaultValues map[string]*ConfigValue `protobuf:"bytes,9,rep,name=worker_default_values,json=workerDefaultValues,proto3" json:"worker_default_values,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Worker config field containing a cooldown in seconds after the latest successful job. + DetectionCooldownConfigField string `protobuf:"bytes,10,opt,name=detection_cooldown_config_field,json=detectionCooldownConfigField,proto3" json:"detection_cooldown_config_field,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *JobTypeDescriptor) Reset() { @@ -1587,6 +1589,13 @@ func (x *JobTypeDescriptor) GetWorkerDefaultValues() map[string]*ConfigValue { return nil } +func (x *JobTypeDescriptor) GetDetectionCooldownConfigField() string { + if x != nil { + return x.DetectionCooldownConfigField + } + return "" +} + type ConfigForm struct { state protoimpl.MessageState `protogen:"open.v1"` FormId string `protobuf:"bytes,1,opt,name=form_id,json=formId,proto3" json:"form_id,omitempty"` @@ -2948,6 +2957,8 @@ type DetectionComplete struct { Success bool `protobuf:"varint,3,opt,name=success,proto3" json:"success,omitempty"` ErrorMessage string `protobuf:"bytes,4,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` TotalProposals int32 `protobuf:"varint,5,opt,name=total_proposals,json=totalProposals,proto3" json:"total_proposals,omitempty"` + SkippedReason string `protobuf:"bytes,6,opt,name=skipped_reason,json=skippedReason,proto3" json:"skipped_reason,omitempty"` + EligibleAfter *timestamppb.Timestamp `protobuf:"bytes,7,opt,name=eligible_after,json=eligibleAfter,proto3" json:"eligible_after,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3017,6 +3028,20 @@ func (x *DetectionComplete) GetTotalProposals() int32 { return 0 } +func (x *DetectionComplete) GetSkippedReason() string { + if x != nil { + return x.SkippedReason + } + return "" +} + +func (x *DetectionComplete) GetEligibleAfter() *timestamppb.Timestamp { + if x != nil { + return x.EligibleAfter + } + return nil +} + type JobProposal struct { state protoimpl.MessageState `protogen:"open.v1"` ProposalId string `protobuf:"bytes,1,opt,name=proposal_id,json=proposalId,proto3" json:"proposal_id,omitempty"` @@ -4415,7 +4440,7 @@ const file_plugin_proto_rawDesc = "" + "\bjob_type\x18\x02 \x01(\tR\ajobType\x12\x18\n" + "\asuccess\x18\x03 \x01(\bR\asuccess\x12#\n" + "\rerror_message\x18\x04 \x01(\tR\ferrorMessage\x12I\n" + - "\x13job_type_descriptor\x18\x05 \x01(\v2\x19.plugin.JobTypeDescriptorR\x11jobTypeDescriptor\"\xd1\x04\n" + + "\x13job_type_descriptor\x18\x05 \x01(\v2\x19.plugin.JobTypeDescriptorR\x11jobTypeDescriptor\"\x98\x05\n" + "\x11JobTypeDescriptor\x12\x19\n" + "\bjob_type\x18\x01 \x01(\tR\ajobType\x12!\n" + "\fdisplay_name\x18\x02 \x01(\tR\vdisplayName\x12 \n" + @@ -4425,7 +4450,9 @@ const file_plugin_proto_rawDesc = "" + "\x11admin_config_form\x18\x06 \x01(\v2\x12.plugin.ConfigFormR\x0fadminConfigForm\x12@\n" + "\x12worker_config_form\x18\a \x01(\v2\x12.plugin.ConfigFormR\x10workerConfigForm\x12R\n" + "\x16admin_runtime_defaults\x18\b \x01(\v2\x1c.plugin.AdminRuntimeDefaultsR\x14adminRuntimeDefaults\x12f\n" + - "\x15worker_default_values\x18\t \x03(\v22.plugin.JobTypeDescriptor.WorkerDefaultValuesEntryR\x13workerDefaultValues\x1a[\n" + + "\x15worker_default_values\x18\t \x03(\v22.plugin.JobTypeDescriptor.WorkerDefaultValuesEntryR\x13workerDefaultValues\x12E\n" + + "\x1fdetection_cooldown_config_field\x18\n" + + " \x01(\tR\x1cdetectionCooldownConfigField\x1a[\n" + "\x18WorkerDefaultValuesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12)\n" + "\x05value\x18\x02 \x01(\v2\x13.plugin.ConfigValueR\x05value:\x028\x01\"\xb5\x02\n" + @@ -4563,14 +4590,16 @@ const file_plugin_proto_rawDesc = "" + "request_id\x18\x01 \x01(\tR\trequestId\x12\x19\n" + "\bjob_type\x18\x02 \x01(\tR\ajobType\x121\n" + "\tproposals\x18\x03 \x03(\v2\x13.plugin.JobProposalR\tproposals\x12\x19\n" + - "\bhas_more\x18\x04 \x01(\bR\ahasMore\"\xb5\x01\n" + + "\bhas_more\x18\x04 \x01(\bR\ahasMore\"\x9f\x02\n" + "\x11DetectionComplete\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x12\x19\n" + "\bjob_type\x18\x02 \x01(\tR\ajobType\x12\x18\n" + "\asuccess\x18\x03 \x01(\bR\asuccess\x12#\n" + "\rerror_message\x18\x04 \x01(\tR\ferrorMessage\x12'\n" + - "\x0ftotal_proposals\x18\x05 \x01(\x05R\x0etotalProposals\"\xce\x04\n" + + "\x0ftotal_proposals\x18\x05 \x01(\x05R\x0etotalProposals\x12%\n" + + "\x0eskipped_reason\x18\x06 \x01(\tR\rskippedReason\x12A\n" + + "\x0eeligible_after\x18\a \x01(\v2\x1a.google.protobuf.TimestampR\religibleAfter\"\xce\x04\n" + "\vJobProposal\x12\x1f\n" + "\vproposal_id\x18\x01 \x01(\tR\n" + "proposalId\x12\x1d\n" + @@ -4948,64 +4977,65 @@ var file_plugin_proto_depIdxs = []int32{ 41, // 54: plugin.RunDetectionRequest.cluster_context:type_name -> plugin.ClusterContext 72, // 55: plugin.RunDetectionRequest.last_successful_run:type_name -> google.protobuf.Timestamp 35, // 56: plugin.DetectionProposals.proposals:type_name -> plugin.JobProposal - 1, // 57: plugin.JobProposal.priority:type_name -> plugin.JobPriority - 59, // 58: plugin.JobProposal.parameters:type_name -> plugin.JobProposal.ParametersEntry - 60, // 59: plugin.JobProposal.labels:type_name -> plugin.JobProposal.LabelsEntry - 72, // 60: plugin.JobProposal.not_before:type_name -> google.protobuf.Timestamp - 72, // 61: plugin.JobProposal.expires_at:type_name -> google.protobuf.Timestamp - 37, // 62: plugin.ExecuteJobRequest.job:type_name -> plugin.JobSpec - 31, // 63: plugin.ExecuteJobRequest.admin_runtime:type_name -> plugin.AdminRuntimeConfig - 61, // 64: plugin.ExecuteJobRequest.admin_config_values:type_name -> plugin.ExecuteJobRequest.AdminConfigValuesEntry - 62, // 65: plugin.ExecuteJobRequest.worker_config_values:type_name -> plugin.ExecuteJobRequest.WorkerConfigValuesEntry - 41, // 66: plugin.ExecuteJobRequest.cluster_context:type_name -> plugin.ClusterContext - 1, // 67: plugin.JobSpec.priority:type_name -> plugin.JobPriority - 63, // 68: plugin.JobSpec.parameters:type_name -> plugin.JobSpec.ParametersEntry - 64, // 69: plugin.JobSpec.labels:type_name -> plugin.JobSpec.LabelsEntry - 72, // 70: plugin.JobSpec.created_at:type_name -> google.protobuf.Timestamp - 72, // 71: plugin.JobSpec.scheduled_at:type_name -> google.protobuf.Timestamp - 2, // 72: plugin.JobProgressUpdate.state:type_name -> plugin.JobState - 65, // 73: plugin.JobProgressUpdate.metrics:type_name -> plugin.JobProgressUpdate.MetricsEntry - 42, // 74: plugin.JobProgressUpdate.activities:type_name -> plugin.ActivityEvent - 72, // 75: plugin.JobProgressUpdate.updated_at:type_name -> google.protobuf.Timestamp - 40, // 76: plugin.JobCompleted.result:type_name -> plugin.JobResult - 42, // 77: plugin.JobCompleted.activities:type_name -> plugin.ActivityEvent - 72, // 78: plugin.JobCompleted.completed_at:type_name -> google.protobuf.Timestamp - 66, // 79: plugin.JobResult.output_values:type_name -> plugin.JobResult.OutputValuesEntry - 67, // 80: plugin.ClusterContext.metadata:type_name -> plugin.ClusterContext.MetadataEntry - 6, // 81: plugin.ActivityEvent.source:type_name -> plugin.ActivitySource - 68, // 82: plugin.ActivityEvent.details:type_name -> plugin.ActivityEvent.DetailsEntry - 72, // 83: plugin.ActivityEvent.created_at:type_name -> google.protobuf.Timestamp - 0, // 84: plugin.CancelRequest.target_kind:type_name -> plugin.WorkKind - 69, // 85: plugin.PersistedJobTypeConfig.admin_config_values:type_name -> plugin.PersistedJobTypeConfig.AdminConfigValuesEntry - 70, // 86: plugin.PersistedJobTypeConfig.worker_config_values:type_name -> plugin.PersistedJobTypeConfig.WorkerConfigValuesEntry - 31, // 87: plugin.PersistedJobTypeConfig.admin_runtime:type_name -> plugin.AdminRuntimeConfig - 72, // 88: plugin.PersistedJobTypeConfig.updated_at:type_name -> google.protobuf.Timestamp - 47, // 89: plugin.WorkerObservations.observations:type_name -> plugin.ObjectObservation - 71, // 90: plugin.ObjectObservation.attributes:type_name -> plugin.ObjectObservation.AttributesEntry - 72, // 91: plugin.ObjectObservation.observed_at:type_name -> google.protobuf.Timestamp - 50, // 92: plugin.ObjectPreviewResponse.rows:type_name -> plugin.PreviewRow - 23, // 93: plugin.JobTypeDescriptor.WorkerDefaultValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 94: plugin.ConfigForm.DefaultValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 95: plugin.ValueMap.FieldsEntry.value:type_name -> plugin.ConfigValue - 23, // 96: plugin.RunDetectionRequest.AdminConfigValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 97: plugin.RunDetectionRequest.WorkerConfigValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 98: plugin.JobProposal.ParametersEntry.value:type_name -> plugin.ConfigValue - 23, // 99: plugin.ExecuteJobRequest.AdminConfigValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 100: plugin.ExecuteJobRequest.WorkerConfigValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 101: plugin.JobSpec.ParametersEntry.value:type_name -> plugin.ConfigValue - 23, // 102: plugin.JobProgressUpdate.MetricsEntry.value:type_name -> plugin.ConfigValue - 23, // 103: plugin.JobResult.OutputValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 104: plugin.ActivityEvent.DetailsEntry.value:type_name -> plugin.ConfigValue - 23, // 105: plugin.PersistedJobTypeConfig.AdminConfigValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 106: plugin.PersistedJobTypeConfig.WorkerConfigValuesEntry.value:type_name -> plugin.ConfigValue - 23, // 107: plugin.ObjectObservation.AttributesEntry.value:type_name -> plugin.ConfigValue - 7, // 108: plugin.PluginControlService.WorkerStream:input_type -> plugin.WorkerToAdminMessage - 8, // 109: plugin.PluginControlService.WorkerStream:output_type -> plugin.AdminToWorkerMessage - 109, // [109:110] is the sub-list for method output_type - 108, // [108:109] is the sub-list for method input_type - 108, // [108:108] is the sub-list for extension type_name - 108, // [108:108] is the sub-list for extension extendee - 0, // [0:108] is the sub-list for field type_name + 72, // 57: plugin.DetectionComplete.eligible_after:type_name -> google.protobuf.Timestamp + 1, // 58: plugin.JobProposal.priority:type_name -> plugin.JobPriority + 59, // 59: plugin.JobProposal.parameters:type_name -> plugin.JobProposal.ParametersEntry + 60, // 60: plugin.JobProposal.labels:type_name -> plugin.JobProposal.LabelsEntry + 72, // 61: plugin.JobProposal.not_before:type_name -> google.protobuf.Timestamp + 72, // 62: plugin.JobProposal.expires_at:type_name -> google.protobuf.Timestamp + 37, // 63: plugin.ExecuteJobRequest.job:type_name -> plugin.JobSpec + 31, // 64: plugin.ExecuteJobRequest.admin_runtime:type_name -> plugin.AdminRuntimeConfig + 61, // 65: plugin.ExecuteJobRequest.admin_config_values:type_name -> plugin.ExecuteJobRequest.AdminConfigValuesEntry + 62, // 66: plugin.ExecuteJobRequest.worker_config_values:type_name -> plugin.ExecuteJobRequest.WorkerConfigValuesEntry + 41, // 67: plugin.ExecuteJobRequest.cluster_context:type_name -> plugin.ClusterContext + 1, // 68: plugin.JobSpec.priority:type_name -> plugin.JobPriority + 63, // 69: plugin.JobSpec.parameters:type_name -> plugin.JobSpec.ParametersEntry + 64, // 70: plugin.JobSpec.labels:type_name -> plugin.JobSpec.LabelsEntry + 72, // 71: plugin.JobSpec.created_at:type_name -> google.protobuf.Timestamp + 72, // 72: plugin.JobSpec.scheduled_at:type_name -> google.protobuf.Timestamp + 2, // 73: plugin.JobProgressUpdate.state:type_name -> plugin.JobState + 65, // 74: plugin.JobProgressUpdate.metrics:type_name -> plugin.JobProgressUpdate.MetricsEntry + 42, // 75: plugin.JobProgressUpdate.activities:type_name -> plugin.ActivityEvent + 72, // 76: plugin.JobProgressUpdate.updated_at:type_name -> google.protobuf.Timestamp + 40, // 77: plugin.JobCompleted.result:type_name -> plugin.JobResult + 42, // 78: plugin.JobCompleted.activities:type_name -> plugin.ActivityEvent + 72, // 79: plugin.JobCompleted.completed_at:type_name -> google.protobuf.Timestamp + 66, // 80: plugin.JobResult.output_values:type_name -> plugin.JobResult.OutputValuesEntry + 67, // 81: plugin.ClusterContext.metadata:type_name -> plugin.ClusterContext.MetadataEntry + 6, // 82: plugin.ActivityEvent.source:type_name -> plugin.ActivitySource + 68, // 83: plugin.ActivityEvent.details:type_name -> plugin.ActivityEvent.DetailsEntry + 72, // 84: plugin.ActivityEvent.created_at:type_name -> google.protobuf.Timestamp + 0, // 85: plugin.CancelRequest.target_kind:type_name -> plugin.WorkKind + 69, // 86: plugin.PersistedJobTypeConfig.admin_config_values:type_name -> plugin.PersistedJobTypeConfig.AdminConfigValuesEntry + 70, // 87: plugin.PersistedJobTypeConfig.worker_config_values:type_name -> plugin.PersistedJobTypeConfig.WorkerConfigValuesEntry + 31, // 88: plugin.PersistedJobTypeConfig.admin_runtime:type_name -> plugin.AdminRuntimeConfig + 72, // 89: plugin.PersistedJobTypeConfig.updated_at:type_name -> google.protobuf.Timestamp + 47, // 90: plugin.WorkerObservations.observations:type_name -> plugin.ObjectObservation + 71, // 91: plugin.ObjectObservation.attributes:type_name -> plugin.ObjectObservation.AttributesEntry + 72, // 92: plugin.ObjectObservation.observed_at:type_name -> google.protobuf.Timestamp + 50, // 93: plugin.ObjectPreviewResponse.rows:type_name -> plugin.PreviewRow + 23, // 94: plugin.JobTypeDescriptor.WorkerDefaultValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 95: plugin.ConfigForm.DefaultValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 96: plugin.ValueMap.FieldsEntry.value:type_name -> plugin.ConfigValue + 23, // 97: plugin.RunDetectionRequest.AdminConfigValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 98: plugin.RunDetectionRequest.WorkerConfigValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 99: plugin.JobProposal.ParametersEntry.value:type_name -> plugin.ConfigValue + 23, // 100: plugin.ExecuteJobRequest.AdminConfigValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 101: plugin.ExecuteJobRequest.WorkerConfigValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 102: plugin.JobSpec.ParametersEntry.value:type_name -> plugin.ConfigValue + 23, // 103: plugin.JobProgressUpdate.MetricsEntry.value:type_name -> plugin.ConfigValue + 23, // 104: plugin.JobResult.OutputValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 105: plugin.ActivityEvent.DetailsEntry.value:type_name -> plugin.ConfigValue + 23, // 106: plugin.PersistedJobTypeConfig.AdminConfigValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 107: plugin.PersistedJobTypeConfig.WorkerConfigValuesEntry.value:type_name -> plugin.ConfigValue + 23, // 108: plugin.ObjectObservation.AttributesEntry.value:type_name -> plugin.ConfigValue + 7, // 109: plugin.PluginControlService.WorkerStream:input_type -> plugin.WorkerToAdminMessage + 8, // 110: plugin.PluginControlService.WorkerStream:output_type -> plugin.AdminToWorkerMessage + 110, // [110:111] is the sub-list for method output_type + 109, // [109:110] is the sub-list for method input_type + 109, // [109:109] is the sub-list for extension type_name + 109, // [109:109] is the sub-list for extension extendee + 0, // [0:109] is the sub-list for field type_name } func init() { file_plugin_proto_init() } diff --git a/weed/plugin/worker/admin_script_handler.go b/weed/plugin/worker/admin_script_handler.go index d6ff69cca9..5df6fbe3f6 100644 --- a/weed/plugin/worker/admin_script_handler.go +++ b/weed/plugin/worker/admin_script_handler.go @@ -142,28 +142,8 @@ func (h *AdminScriptHandler) Detect(ctx context.Context, request *plugin_pb.RunD script := normalizeAdminScript(ReadStringConfig(request.GetAdminConfigValues(), "script", "")) scriptName := strings.TrimSpace(ReadStringConfig(request.GetAdminConfigValues(), "script_name", "")) runIntervalMinutes := readAdminScriptRunIntervalMinutes(request.GetAdminConfigValues()) - if ShouldSkipDetectionByInterval(request.GetLastSuccessfulRun(), runIntervalMinutes*60) { - _ = sender.SendActivity(BuildDetectorActivity( - "skipped_by_interval", - fmt.Sprintf("ADMIN SCRIPT: Detection skipped due to run interval (%dm)", runIntervalMinutes), - map[string]*plugin_pb.ConfigValue{ - "run_interval_minutes": { - Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(runIntervalMinutes)}, - }, - }, - )) - if err := sender.SendProposals(&plugin_pb.DetectionProposals{ - JobType: adminScriptJobType, - Proposals: []*plugin_pb.JobProposal{}, - HasMore: false, - }); err != nil { - return err - } - return sender.SendComplete(&plugin_pb.DetectionComplete{ - JobType: adminScriptJobType, - Success: true, - TotalProposals: 0, - }) + if skipped, err := SkipDetectionByInterval(request, sender, runIntervalMinutes*60); skipped || err != nil { + return err } commands := parseAdminScriptCommands(script) diff --git a/weed/plugin/worker/admin_script_handler_test.go b/weed/plugin/worker/admin_script_handler_test.go index 5f134078b1..235cff769a 100644 --- a/weed/plugin/worker/admin_script_handler_test.go +++ b/weed/plugin/worker/admin_script_handler_test.go @@ -92,11 +92,14 @@ func TestAdminScriptDetectSkipsByRunInterval(t *testing.T) { if sender.complete == nil || !sender.complete.Success { t.Fatalf("expected successful completion message") } + if sender.complete.SkippedReason == "" || sender.complete.EligibleAfter == nil { + t.Fatalf("expected skipped completion metadata, got %+v", sender.complete) + } if len(sender.events) == 0 { t.Fatalf("expected detector activity events") } - if !strings.Contains(sender.events[0].Message, "run interval") { - t.Fatalf("unexpected skip message: %q", sender.events[0].Message) + if sender.events[0].Stage != "skipped_by_interval" { + t.Fatalf("unexpected skip stage: %q", sender.events[0].Stage) } } diff --git a/weed/plugin/worker/interval.go b/weed/plugin/worker/interval.go index 880d8fe127..1144a23376 100644 --- a/weed/plugin/worker/interval.go +++ b/weed/plugin/worker/interval.go @@ -1,8 +1,10 @@ package pluginworker import ( + "fmt" "time" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -18,3 +20,25 @@ func ShouldSkipDetectionByInterval(lastSuccessfulRun *timestamppb.Timestamp, min } return time.Since(lastRun) < time.Duration(minIntervalSeconds)*time.Second } + +func SkipDetectionByInterval(request *plugin_pb.RunDetectionRequest, sender DetectionSender, minIntervalSeconds int) (bool, error) { + if !ShouldSkipDetectionByInterval(request.GetLastSuccessfulRun(), minIntervalSeconds) { + return false, nil + } + until := request.GetLastSuccessfulRun().AsTime().Add(time.Duration(minIntervalSeconds) * time.Second) + reason := fmt.Sprintf("Cooldown after last completed job (%s)", time.Duration(minIntervalSeconds)*time.Second) + if err := sender.SendActivity(BuildDetectorActivity("skipped_by_interval", reason, map[string]*plugin_pb.ConfigValue{ + "eligible_after": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: until.UTC().Format(time.RFC3339)}}, + })); err != nil { + return true, err + } + if err := sender.SendProposals(&plugin_pb.DetectionProposals{JobType: request.GetJobType()}); err != nil { + return true, err + } + return true, sender.SendComplete(&plugin_pb.DetectionComplete{ + JobType: request.GetJobType(), + Success: true, + SkippedReason: reason, + EligibleAfter: timestamppb.New(until), + }) +} diff --git a/weed/plugin/worker/worker_test.go b/weed/plugin/worker/worker_test.go index d540d048d5..a519d8568d 100644 --- a/weed/plugin/worker/worker_test.go +++ b/weed/plugin/worker/worker_test.go @@ -8,8 +8,47 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/types/known/timestamppb" ) +func TestSkipDetectionByInterval(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + for _, tc := range []struct { + name string + last *timestamppb.Timestamp + seconds int + skip bool + }{ + {"no history", nil, 36000, false}, + {"disabled", timestamppb.New(now), 0, false}, + {"cooldown", timestamppb.New(now), 36000, true}, + {"expired", timestamppb.New(now.Add(-11 * time.Hour)), 36000, false}, + } { + t.Run(tc.name, func(t *testing.T) { + sender := &recordingDetectionSender{} + skipped, err := SkipDetectionByInterval(&plugin_pb.RunDetectionRequest{JobType: "vacuum", LastSuccessfulRun: tc.last}, sender, tc.seconds) + if err != nil || skipped != tc.skip { + t.Fatalf("skip=%v err=%v", skipped, err) + } + if !skipped { + if sender.complete != nil { + t.Fatal("unexpected completion for eligible detection") + } + return + } + if sender.complete.GetSkippedReason() == "" || !sender.complete.GetSuccess() || sender.proposals == nil { + t.Fatalf("missing skipped outcome: %v", sender.complete) + } + if len(sender.events) != 1 || sender.events[0].Stage != "skipped_by_interval" { + t.Fatal("older admins must still receive a cooldown activity") + } + if want := now.Add(10 * time.Hour); !sender.complete.EligibleAfter.AsTime().Equal(want) { + t.Fatalf("eligible after = %v, want %v", sender.complete.EligibleAfter, want) + } + }) + } +} + func TestWorkerBuildHelloUsesConfiguredConcurrency(t *testing.T) { handler := &testJobHandler{ capability: &plugin_pb.JobTypeCapability{