mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-13 15:29:12 +01:00
admin: filter plugin jobs and activities by lane before the limit
/api/plugin/jobs and /api/plugin/activities took the newest N rows across every lane and then dropped the other lanes' rows, so on a busy default lane a quieter lane's page could come back empty. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
86d7a9b71d
commit
e73684cb95
2 files changed
+81
-4
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user