diff --git a/weed/admin/dash/plugin_api.go b/weed/admin/dash/plugin_api.go index 93821b5c4d..58afa7e977 100644 --- a/weed/admin/dash/plugin_api.go +++ b/weed/admin/dash/plugin_api.go @@ -152,12 +152,21 @@ func (s *AdminServer) GetPluginJobsAPI(w http.ResponseWriter, r *http.Request) { state := strings.TrimSpace(query.Get("state")) laneFilter := strings.TrimSpace(query.Get("lane")) limit := parsePositiveInt(query.Get("limit"), 200) - jobs := s.ListPluginJobs(jobType, state, limit) + // Apply the lane before the limit, or a busy lane's newest rows crowd out a quiet lane. + listLimit := limit + if laneFilter != "" { + listLimit = 0 + } + jobs := s.ListPluginJobs(jobType, state, listLimit) if jobs == nil { writeJSON(w, http.StatusOK, []interface{}{}) return } - writeJSON(w, http.StatusOK, filterTrackedJobsByLane(jobs, laneFilter)) + jobs = filterTrackedJobsByLane(jobs, laneFilter) + if len(jobs) > limit { + jobs = jobs[:limit] + } + writeJSON(w, http.StatusOK, jobs) } // GetPluginJobAPI returns one tracked job. @@ -248,12 +257,20 @@ func (s *AdminServer) GetPluginActivitiesAPI(w http.ResponseWriter, r *http.Requ jobType := strings.TrimSpace(query.Get("job_type")) laneFilter := strings.TrimSpace(query.Get("lane")) limit := parsePositiveInt(query.Get("limit"), 500) - activities := s.ListPluginActivities(jobType, limit) + listLimit := limit + if laneFilter != "" { + listLimit = 0 + } + activities := s.ListPluginActivities(jobType, listLimit) if activities == nil { writeJSON(w, http.StatusOK, []interface{}{}) return } - writeJSON(w, http.StatusOK, filterActivitiesByLane(activities, laneFilter)) + activities = filterActivitiesByLane(activities, laneFilter) + if len(activities) > limit { + activities = activities[:limit] + } + writeJSON(w, http.StatusOK, activities) } // GetPluginSchedulerStatesAPI returns per-job-type scheduler status for monitoring. diff --git a/weed/admin/dash/plugin_api_test.go b/weed/admin/dash/plugin_api_test.go index bc74188aa3..a88e215a5b 100644 --- a/weed/admin/dash/plugin_api_test.go +++ b/weed/admin/dash/plugin_api_test.go @@ -3,11 +3,13 @@ package dash import ( "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" "strings" "testing" + "time" "github.com/gorilla/mux" "github.com/seaweedfs/seaweedfs/weed/admin/plugin" @@ -230,3 +232,61 @@ func TestFilterActivitiesByLane(t *testing.T) { t.Fatalf("expected lifecycle activity to be retained, got %+v", filtered[0]) } } + +func TestPluginJobsAndActivitiesAPILimitAfterLaneFilter(t *testing.T) { + dataDir := t.TempDir() + store, err := plugin.NewConfigStore(dataDir) + if err != nil { + t.Fatalf("NewConfigStore: %v", err) + } + now := time.Now() + var jobs []plugin.TrackedJob + var activities []plugin.JobActivity + for i := 0; i < 5; i++ { + jobType, jobID := "vacuum", fmt.Sprintf("vacuum-%d", i) + if i == 4 { + jobType, jobID = "iceberg_maintenance", "iceberg-1" + } + at := now.Add(-time.Duration(i) * time.Minute) + jobs = append(jobs, plugin.TrackedJob{JobID: jobID, JobType: jobType, State: plugin.StateSucceeded, UpdatedAt: &at}) + activities = append(activities, plugin.JobActivity{JobID: jobID, JobType: jobType, Message: "done", OccurredAt: &at}) + } + if err := store.SaveTrackedJobs(jobs); err != nil { + t.Fatalf("SaveTrackedJobs: %v", err) + } + if err := store.SaveActivities(activities); err != nil { + t.Fatalf("SaveActivities: %v", err) + } + p, err := plugin.New(plugin.Options{DataDir: dataDir}) + if err != nil { + t.Fatalf("plugin.New: %v", err) + } + defer p.Shutdown() + s := &AdminServer{plugin: p} + + get := func(handler http.HandlerFunc, url string) []map[string]any { + recorder := httptest.NewRecorder() + handler(recorder, httptest.NewRequest(http.MethodGet, url, nil)) + if recorder.Code != http.StatusOK { + t.Fatalf("%s: status %d", url, recorder.Code) + } + var rows []map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &rows); err != nil { + t.Fatalf("%s: %v", url, err) + } + return rows + } + + if rows := get(s.GetPluginJobsAPI, "/api/plugin/jobs?limit=2&lane=iceberg"); len(rows) != 1 || rows[0]["job_id"] != "iceberg-1" { + t.Fatalf("iceberg lane jobs: got %v, want the one iceberg job", rows) + } + if rows := get(s.GetPluginJobsAPI, "/api/plugin/jobs?limit=2&lane=default"); len(rows) != 2 { + t.Fatalf("default lane jobs: got %d rows, want the limit of 2", len(rows)) + } + if rows := get(s.GetPluginActivitiesAPI, "/api/plugin/activities?limit=2&lane=iceberg"); len(rows) != 1 || rows[0]["job_id"] != "iceberg-1" { + t.Fatalf("iceberg lane activities: got %v, want the one iceberg activity", rows) + } + if rows := get(s.GetPluginActivitiesAPI, "/api/plugin/activities?limit=2&lane=default"); len(rows) != 2 { + t.Fatalf("default lane activities: got %d rows, want the limit of 2", len(rows)) + } +} diff --git a/weed/admin/view/app/plugin.templ b/weed/admin/view/app/plugin.templ index 166e42106d..f0518c95bc 100644 --- a/weed/admin/view/app/plugin.templ +++ b/weed/admin/view/app/plugin.templ @@ -620,6 +620,8 @@ templ Plugin(page string, initialJob string, lane string) { schedulerLock: null, allJobs: [], allActivities: [], + monitorRowsScope: null, + refreshError: '', loadedJobType: '', descriptor: null, config: null, @@ -633,6 +635,13 @@ templ Plugin(page string, initialJob string, lane string) { return url + sep + 'lane=' + encodeURIComponent(pluginLane); } + // appendJobTypeParam appends ?job_type= or &job_type= to a URL if a job type is given. + function appendJobTypeParam(url, jobType) { + if (!jobType) return url; + var sep = url.indexOf('?') === -1 ? '?' : '&'; + return url + sep + 'job_type=' + encodeURIComponent(jobType); + } + function getCSRFToken() { var meta = document.querySelector('meta[name="csrf-token"]'); if (!meta) { @@ -787,6 +796,10 @@ templ Plugin(page string, initialJob string, lane string) { return { jobType: jobType, displayName: displayName }; } + function monitorJobTypeScope() { + return state.activeTopTab === 'overview' ? '' : state.selectedJobType; + } + function topTabKeyForJobType(jobType) { var normalized = String(jobType || '').trim(); if (!normalized) { @@ -1257,11 +1270,16 @@ templ Plugin(page string, initialJob string, lane string) { } else { notify('Job expired: ' + normalizedJobID, 'success'); } - await refreshJobsAndActivities(); - await openJobDetail(normalizedJobID); } catch (e) { notify('Failed to expire job: ' + e.message, 'error'); + return; } + try { + await refreshJobsAndActivities(); + } catch (refreshErr) { + notify('Monitoring refresh failed: ' + refreshErr.message, 'error'); + } + await openJobDetail(normalizedJobID); } async function openJobDetail(jobID) { @@ -1608,7 +1626,7 @@ templ Plugin(page string, initialJob string, lane string) { } function renderSchedulerStatus() { - window.renderPluginSchedulerStatus({scheduler: state.schedulerStatus, lock: state.schedulerLock, job_type: state.activeTopTab === 'overview' ? '' : state.selectedJobType}); + window.renderPluginSchedulerStatus({scheduler: state.schedulerStatus, lock: state.schedulerLock, job_type: monitorJobTypeScope()}); var valueNodes = document.querySelectorAll('.plugin-scheduler-next-run'); if (!valueNodes.length) { return; @@ -1735,9 +1753,9 @@ templ Plugin(page string, initialJob string, lane string) { } var filteredJobs = []; - var allJobs = Array.isArray(state.allJobs) ? state.allJobs : []; - for (var i = 0; i < allJobs.length; i++) { - var job = allJobs[i] || {}; + var jobs = Array.isArray(state.jobs) ? state.jobs : []; + for (var i = 0; i < jobs.length; i++) { + var job = jobs[i] || {}; var jobType = String(job.job_type || '').trim(); if (state.selectedJobType && jobType !== state.selectedJobType) { continue; @@ -1801,9 +1819,9 @@ templ Plugin(page string, initialJob string, lane string) { } var filteredJobs = []; - var allJobs = Array.isArray(state.allJobs) ? state.allJobs : []; - for (var i = 0; i < allJobs.length; i++) { - var queueJob = allJobs[i] || {}; + var jobs = Array.isArray(state.jobs) ? state.jobs : []; + for (var i = 0; i < jobs.length; i++) { + var queueJob = jobs[i] || {}; var queueJobType = String(queueJob.job_type || '').trim(); if (state.selectedJobType && queueJobType !== state.selectedJobType) { continue; @@ -1858,9 +1876,9 @@ templ Plugin(page string, initialJob string, lane string) { } var filteredActivities = []; - var allActivities = Array.isArray(state.allActivities) ? state.allActivities : []; - for (var i = 0; i < allActivities.length; i++) { - var activity = allActivities[i] || {}; + var activities = Array.isArray(state.activities) ? state.activities : []; + for (var i = 0; i < activities.length; i++) { + var activity = activities[i] || {}; var jobType = String(activity.job_type || '').trim(); if (state.selectedJobType && jobType !== state.selectedJobType) { continue; @@ -1900,9 +1918,9 @@ templ Plugin(page string, initialJob string, lane string) { } var filteredActivities = []; - var allActivities = Array.isArray(state.allActivities) ? state.allActivities : []; - for (var i = 0; i < allActivities.length; i++) { - var activity = allActivities[i] || {}; + var activities = Array.isArray(state.activities) ? state.activities : []; + for (var i = 0; i < activities.length; i++) { + var activity = activities[i] || {}; if (isDetectionActivitySource(activity.source)) { continue; } @@ -3064,9 +3082,14 @@ templ Plugin(page string, initialJob string, lane string) { '
Success: ' + escapeHtml(String(data.success_count || 0)) + '
' + '
Errors: ' + escapeHtml(String(data.error_count || 0)) + '
'; await loadRunHistory(jobType); - await refreshJobsAndActivities(); } catch (e) { notify('Workflow run failed: ' + e.message, 'error'); + return; + } + try { + await refreshJobsAndActivities(); + } catch (refreshErr) { + notify('Workflow run succeeded, but monitoring refresh failed: ' + refreshErr.message, 'error'); } } @@ -3075,13 +3098,31 @@ templ Plugin(page string, initialJob string, lane string) { state.monitorExecutionJobState = executionStateFilter ? String(executionStateFilter.value || '').trim() : ''; - var allJobsPromise = pluginRequest('GET', appendLaneParam('/api/plugin/jobs?limit=500')); - var allActivitiesPromise = pluginRequest('GET', appendLaneParam('/api/plugin/activities?limit=500')); - var schedulerPromise = pluginRequest('GET', appendLaneParam('/api/plugin/scheduler-states')); + // The newest 500 rows cluster-wide may hold none of the selected job type on a busy + // cluster, so the job-type sub-tabs fetch their own server-side filtered window. + var scopedJobType = monitorJobTypeScope(); + var sameScope = state.monitorRowsScope === scopedJobType; + + // A failed request keeps its last rows; the rest still land and the failure is rethrown. + var failure = null; + function keepOnFailure(promise, previous) { + return promise.catch(function(e) { + failure = failure || e; + return previous; + }); + } + + var allJobsPromise = keepOnFailure(pluginRequest('GET', appendLaneParam('/api/plugin/jobs?limit=500')), state.allJobs); + var allActivitiesPromise = keepOnFailure(pluginRequest('GET', appendLaneParam('/api/plugin/activities?limit=500')), state.allActivities); + var jobsPromise = scopedJobType ? keepOnFailure(pluginRequest('GET', appendJobTypeParam(appendLaneParam('/api/plugin/jobs?limit=500'), scopedJobType)), sameScope ? state.jobs : null) : allJobsPromise; + var activitiesPromise = scopedJobType ? keepOnFailure(pluginRequest('GET', appendJobTypeParam(appendLaneParam('/api/plugin/activities?limit=500'), scopedJobType)), sameScope ? state.activities : null) : allActivitiesPromise; + var schedulerPromise = keepOnFailure(pluginRequest('GET', appendLaneParam('/api/plugin/scheduler-states')), state.schedulerStates); var schedulerStatusPromise = pluginRequest('GET', appendLaneParam('/api/plugin/scheduler-status')); var allJobs = await allJobsPromise; var allActivities = await allActivitiesPromise; + var jobs = await jobsPromise; + var activities = await activitiesPromise; var schedulerStates = await schedulerPromise; var schedulerStatus = null; try { @@ -3089,11 +3130,15 @@ templ Plugin(page string, initialJob string, lane string) { } catch (e) { schedulerStatus = null; } + if (monitorJobTypeScope() !== scopedJobType) { + return; + } - state.jobs = Array.isArray(allJobs) ? allJobs : []; - state.activities = Array.isArray(allActivities) ? allActivities : []; - state.allJobs = state.jobs; - state.allActivities = state.activities; + state.allJobs = Array.isArray(allJobs) ? allJobs : []; + state.allActivities = Array.isArray(allActivities) ? allActivities : []; + state.jobs = Array.isArray(jobs) ? jobs : state.allJobs; + state.activities = Array.isArray(activities) ? activities : state.allActivities; + state.monitorRowsScope = scopedJobType; state.schedulerStates = Array.isArray(schedulerStates) ? schedulerStates : []; state.schedulerStatus = schedulerStatus && schedulerStatus.scheduler ? schedulerStatus.scheduler : null; state.schedulerLock = schedulerStatus && schedulerStatus.lock ? schedulerStatus.lock : null; @@ -3105,6 +3150,9 @@ templ Plugin(page string, initialJob string, lane string) { renderSchedulerStatus(); renderStatus(); renderJobTypeSummary(); + if (failure) { + throw failure; + } } async function refreshAll() { @@ -3133,8 +3181,13 @@ templ Plugin(page string, initialJob string, lane string) { } await refreshJobsAndActivities(); + state.refreshError = ''; } catch (e) { - notify('Refresh failed: ' + e.message, 'error'); + // The 3s timer would otherwise reopen the same alert on every tick. + if (e.message !== state.refreshError) { + state.refreshError = e.message; + notify('Refresh failed: ' + e.message, 'error'); + } } } @@ -3231,10 +3284,11 @@ templ Plugin(page string, initialJob string, lane string) { } await loadRunHistory(jobType); } - renderQueueJobs(); - renderDetectionJobs(); - renderExecutionJobs(); - renderExecutionActivities(); + try { + await refreshJobsAndActivities(); + } catch (e) { + notify('Refresh failed: ' + e.message, 'error'); + } return; } @@ -3300,10 +3354,11 @@ templ Plugin(page string, initialJob string, lane string) { console.error('Failed to load data on popstate navigation:', e); } } - renderQueueJobs(); - renderDetectionJobs(); - renderExecutionJobs(); - renderExecutionActivities(); + try { + await refreshJobsAndActivities(); + } catch (e) { + console.error('Failed to refresh jobs on popstate navigation:', e); + } }); window.addEventListener('beforeunload', function() { diff --git a/weed/admin/view/app/plugin_templ.go b/weed/admin/view/app/plugin_templ.go index fbc54d49a8..621a1031a8 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 }