From 25f93dfb59b89e3eb515800f0add4ef5fa63f568 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:45:11 +0800 Subject: [PATCH] admin: submissions page, download endpoint, uploads search by app_id Co-Authored-By: Claude --- internal/admin/router.go | 81 +++++- internal/admin/router_test.go | 2 +- internal/admin/submissions_test.go | 155 ++++++++++++ internal/admin/uploads_test.go | 54 +++- internal/admin/web.go | 9 + internal/admin/web/cluster.html | 9 +- internal/admin/web/submissions.html | 367 ++++++++++++++++++++++++++++ internal/admin/web/uploads.html | 9 +- internal/storage/upload_repo.go | 20 +- main.go | 3 +- 10 files changed, 683 insertions(+), 26 deletions(-) create mode 100644 internal/admin/submissions_test.go create mode 100644 internal/admin/web/submissions.html diff --git a/internal/admin/router.go b/internal/admin/router.go index dd2e7fc..178aecb 100644 --- a/internal/admin/router.go +++ b/internal/admin/router.go @@ -10,6 +10,7 @@ import ( "os" "path/filepath" "strconv" + "strings" "github.com/gin-gonic/gin" @@ -22,7 +23,7 @@ import ( // Mount attaches the /admin sub-router to r, protecting every route with // bearer-token admin authentication. -func Mount(r *gin.Engine, repo *storage.ClusterRepo, uploadRepo *storage.UploadRepo, auditRepo *audit.Repo, uploadStore *uploads.Store, adminTokens []string) { +func Mount(r *gin.Engine, repo *storage.ClusterRepo, uploadRepo *storage.UploadRepo, submissionRepo *storage.SubmissionRepo, auditRepo *audit.Repo, uploadStore *uploads.Store, adminTokens []string) { // HTML page is public — its own modal prompts for the token. // All other /admin/* endpoints (API + OpenAPI docs) still require auth. gPublic := r.Group("/admin") @@ -38,7 +39,10 @@ func Mount(r *gin.Engine, repo *storage.ClusterRepo, uploadRepo *storage.UploadR g.GET("/audit", listAudit(auditRepo)) g.GET("/uploads", uploadsWebHandler) g.GET("/uploads/api", listUploads(uploadRepo)) + g.GET("/uploads/:id/download", downloadUpload(uploadRepo, uploadStore)) g.DELETE("/uploads/:id", deleteUpload(uploadRepo, auditRepo, uploadStore)) + g.GET("/submissions", submissionsWebHandler) + g.GET("/submissions/api", listSubmissions(submissionRepo)) g.GET("/docs", DocsHandler) g.GET("/docs/spec", OpenAPISpecHandler) } @@ -300,6 +304,81 @@ func listUploads(repo *storage.UploadRepo) gin.HandlerFunc { } } +func listSubmissions(repo *storage.SubmissionRepo) gin.HandlerFunc { + return func(c *gin.Context) { + limitStr := c.DefaultQuery("limit", "100") + limit, err := strconv.Atoi(limitStr) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid limit"}) + return + } + offsetStr := c.DefaultQuery("offset", "0") + offset, err := strconv.Atoi(offsetStr) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid offset"}) + return + } + + subs, err := repo.List(c.Request.Context(), storage.SubmissionFilter{ + Search: c.Query("search"), + FileID: c.Query("file_id"), + ClusterID: c.Query("cluster_id"), + AppID: c.Query("app_id"), + Limit: limit, + Offset: offset, + }) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, subs) + } +} + +func downloadUpload(repo *storage.UploadRepo, store *uploads.Store) gin.HandlerFunc { + return func(c *gin.Context) { + id := c.Param("id") + ctx := c.Request.Context() + + meta, err := repo.Get(ctx, id) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "upload not found"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + dataPath, err := store.AbsPath(id) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + data, err := os.ReadFile(dataPath) + if err != nil { + if os.IsNotExist(err) { + c.JSON(http.StatusNotFound, gin.H{"error": "upload file missing"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + c.Header("Content-Disposition", "attachment; filename=\""+escapeDispositionFilename(meta.Name)+"\"") + c.Data(http.StatusOK, "application/octet-stream", data) + } +} + +// escapeDispositionFilename escapes quotes and backslashes for the +// Content-Disposition filename parameter. The upload validator already rejects +// names containing quotes or newlines; this is defense in depth. +func escapeDispositionFilename(name string) string { + name = strings.ReplaceAll(name, "\\", "\\\\") + name = strings.ReplaceAll(name, "\"", "\\\"") + return name +} + func deleteUpload(repo *storage.UploadRepo, auditRepo *audit.Repo, store *uploads.Store) gin.HandlerFunc { return func(c *gin.Context) { id := c.Param("id") diff --git a/internal/admin/router_test.go b/internal/admin/router_test.go index 7e824d0..33b0183 100644 --- a/internal/admin/router_test.go +++ b/internal/admin/router_test.go @@ -35,7 +35,7 @@ func newTestAdmin(t *testing.T) (*gin.Engine, *storage.ClusterRepo, *audit.Repo) if err != nil { t.Fatalf("new upload store: %v", err) } - Mount(r, db.Clusters(), db.Uploads(), audit.NewRepo(db), &uploadStore, []string{"good-token"}) + Mount(r, db.Clusters(), db.Uploads(), db.Submissions(), audit.NewRepo(db), &uploadStore, []string{"good-token"}) return r, db.Clusters(), audit.NewRepo(db) } diff --git a/internal/admin/submissions_test.go b/internal/admin/submissions_test.go new file mode 100644 index 0000000..903db92 --- /dev/null +++ b/internal/admin/submissions_test.go @@ -0,0 +1,155 @@ +package admin + +import ( + "context" + "encoding/json" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" + + "spark-mcp-go/internal/storage" + "spark-mcp-go/internal/uploads" +) + +func newTestAdminSubmissions(t *testing.T) (*gin.Engine, *storage.SubmissionRepo, *storage.UploadRepo, *uploads.Store) { + t.Helper() + db, err := storage.Open(":memory:") + if err != nil { + t.Fatalf("open: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + uploadStore, err := uploads.New(filepath.Join(t.TempDir(), "uploads")) + if err != nil { + t.Fatalf("new upload store: %v", err) + } + + r := gin.New() + Mount(r, db.Clusters(), db.Uploads(), db.Submissions(), nil, &uploadStore, []string{"good-token"}) + return r, db.Submissions(), db.Uploads(), &uploadStore +} + +func TestSubmissionsWeb_RequiresAuth(t *testing.T) { + r, _, _, _ := newTestAdminSubmissions(t) + + w := doReq(t, r, "GET", "/admin/submissions", "", nil) + if w.Code != http.StatusUnauthorized { + t.Errorf("GET /admin/submissions without auth: got %d, want %d", w.Code, http.StatusUnauthorized) + } +} + +func TestSubmissionsWeb_WithAuth(t *testing.T) { + r, _, _, _ := newTestAdminSubmissions(t) + + w := doReq(t, r, "GET", "/admin/submissions", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("GET /admin/submissions with auth: got %d, want %d", w.Code, http.StatusOK) + } + ct := w.Header().Get("Content-Type") + if !strings.Contains(ct, "text/html") { + t.Errorf("Content-Type = %q, want text/html", ct) + } + body := w.Body.String() + if !strings.Contains(body, "href=\"/admin/uploads\"") && !strings.Contains(body, "href='/admin/uploads'") { + t.Errorf("body missing uploads nav link") + } + if !strings.Contains(body, "href=\"/admin/submissions\"") && !strings.Contains(body, "href='/admin/submissions'") { + t.Errorf("body missing submissions nav link") + } + if !strings.Contains(body, "submissions-body") { + t.Errorf("body missing submissions table") + } +} + +func TestListSubmissions_SearchByAppID(t *testing.T) { + r, subRepo, uploadRepo, _ := newTestAdminSubmissions(t) + ctx := context.Background() + + uploadedAt := time.Unix(0, time.Now().UnixNano()) + if err := uploadRepo.Create(ctx, "00000000000000000000000000000001", "alpha.py", 10, "deadbeef", uploadedAt); err != nil { + t.Fatalf("create upload: %v", err) + } + if err := uploadRepo.Create(ctx, "00000000000000000000000000000002", "beta.py", 20, "cafebabe", uploadedAt.Add(time.Second)); err != nil { + t.Fatalf("create upload: %v", err) + } + + submissions := []storage.Submission{ + {FileID: "00000000000000000000000000000001", AppID: "application_111_0001", TrackingURL: "http://a", ClusterID: "c1", ExitCode: 0, DurationMS: 100, SubmittedAt: time.Now()}, + {FileID: "00000000000000000000000000000002", AppID: "application_222_0001", TrackingURL: "http://b", ClusterID: "c1", ExitCode: 0, DurationMS: 100, SubmittedAt: time.Now().Add(time.Second)}, + } + for _, s := range submissions { + if err := subRepo.Create(ctx, s); err != nil { + t.Fatalf("create submission: %v", err) + } + } + + w := doReq(t, r, "GET", "/admin/submissions/api?search=application_111", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("search: got status %d: %s", w.Code, w.Body.String()) + } + var list []map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &list); err != nil { + t.Fatalf("unmarshal search: %v", err) + } + if len(list) != 1 { + t.Errorf("search result: got %d, want 1", len(list)) + } + if len(list) > 0 && list[0]["app_id"] != "application_111_0001" { + t.Errorf("app_id=%v, want application_111_0001", list[0]["app_id"]) + } + + w = doReq(t, r, "GET", "/admin/submissions/api?search=beta.py", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("search by name: got status %d: %s", w.Code, w.Body.String()) + } + if err := json.Unmarshal(w.Body.Bytes(), &list); err != nil { + t.Fatalf("unmarshal search: %v", err) + } + if len(list) != 1 { + t.Errorf("search by name result: got %d, want 1", len(list)) + } +} + +func TestDownloadUpload(t *testing.T) { + r, _, uploadRepo, store := newTestAdminSubmissions(t) + ctx := context.Background() + + id := "00000000000000000000000000000003" + payload := []byte("hello payload") + if err := uploadRepo.Create(ctx, id, "report.csv", int64(len(payload)), "feedface", time.Now()); err != nil { + t.Fatalf("create upload: %v", err) + } + dataPath, err := store.AbsPath(id) + if err != nil { + t.Fatalf("abspath: %v", err) + } + if err := os.WriteFile(dataPath, payload, 0o640); err != nil { + t.Fatalf("write data: %v", err) + } + + w := doReq(t, r, "GET", "/admin/uploads/"+id+"/download", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("download: got status %d: %s", w.Code, w.Body.String()) + } + if string(w.Body.Bytes()) != string(payload) { + t.Errorf("body = %q, want %q", w.Body.Bytes(), payload) + } + cd := w.Header().Get("Content-Disposition") + if !strings.Contains(cd, `filename="report.csv"`) { + t.Errorf("Content-Disposition = %q, want filename=\"report.csv\"", cd) + } +} + +func TestDownloadUpload_NotFound(t *testing.T) { + r, _, _, _ := newTestAdminSubmissions(t) + + w := doReq(t, r, "GET", "/admin/uploads/00000000000000000000000000000099/download", "good-token", nil) + if w.Code != http.StatusNotFound { + t.Errorf("download unknown: got %d, want %d", w.Code, http.StatusNotFound) + } +} diff --git a/internal/admin/uploads_test.go b/internal/admin/uploads_test.go index 657eb39..aae5d23 100644 --- a/internal/admin/uploads_test.go +++ b/internal/admin/uploads_test.go @@ -18,7 +18,7 @@ import ( "spark-mcp-go/internal/uploads" ) -func newTestAdminUploads(t *testing.T) (*gin.Engine, *storage.UploadRepo, *audit.Repo, *uploads.Store) { +func newTestAdminUploads(t *testing.T) (*gin.Engine, *storage.UploadRepo, *storage.SubmissionRepo, *audit.Repo, *uploads.Store) { t.Helper() db, err := storage.Open(":memory:") if err != nil { @@ -32,12 +32,12 @@ func newTestAdminUploads(t *testing.T) (*gin.Engine, *storage.UploadRepo, *audit } r := gin.New() - Mount(r, db.Clusters(), db.Uploads(), audit.NewRepo(db), &uploadStore, []string{"good-token"}) - return r, db.Uploads(), audit.NewRepo(db), &uploadStore + Mount(r, db.Clusters(), db.Uploads(), db.Submissions(), audit.NewRepo(db), &uploadStore, []string{"good-token"}) + return r, db.Uploads(), db.Submissions(), audit.NewRepo(db), &uploadStore } func TestUploadsWeb_RequiresAuth(t *testing.T) { - r, _, _, _ := newTestAdminUploads(t) + r, _, _, _, _ := newTestAdminUploads(t) w := doReq(t, r, "GET", "/admin/uploads", "", nil) if w.Code != http.StatusUnauthorized { @@ -62,7 +62,7 @@ func TestUploadsWeb_RequiresAuth(t *testing.T) { } func TestListUploads(t *testing.T) { - r, repo, _, _ := newTestAdminUploads(t) + r, repo, _, _, _ := newTestAdminUploads(t) ctx := context.Background() uploadedAt := time.Unix(0, time.Now().UnixNano()) @@ -98,7 +98,7 @@ func TestListUploads(t *testing.T) { } func TestDeleteUpload(t *testing.T) { - r, repo, auditRepo, store := newTestAdminUploads(t) + r, repo, _, auditRepo, store := newTestAdminUploads(t) ctx := context.Background() id := "00000000000000000000000000000003" @@ -152,3 +152,45 @@ func TestDeleteUpload(t *testing.T) { t.Errorf("missing upload.delete audit entry") } } + +func TestListUploads_SearchByAppID(t *testing.T) { + r, uploadRepo, subRepo, _, _ := newTestAdminUploads(t) + ctx := context.Background() + + uploadedAt := time.Unix(0, time.Now().UnixNano()) + if err := uploadRepo.Create(ctx, "00000000000000000000000000000010", "gamma.py", 10, "deadbeef", uploadedAt); err != nil { + t.Fatalf("create upload: %v", err) + } + if err := uploadRepo.Create(ctx, "00000000000000000000000000000011", "delta.py", 20, "cafebabe", uploadedAt.Add(time.Second)); err != nil { + t.Fatalf("create upload: %v", err) + } + + // Only gamma.py has an associated submission; searching by app_id should + // still return it even though the filename doesn't match the query. + if err := subRepo.Create(ctx, storage.Submission{ + FileID: "00000000000000000000000000000010", + AppID: "application_gamma_0001", + TrackingURL: "http://gamma", + ClusterID: "c1", + ExitCode: 0, + DurationMS: 100, + SubmittedAt: time.Now(), + }); err != nil { + t.Fatalf("create submission: %v", err) + } + + w := doReq(t, r, "GET", "/admin/uploads/api?search=application_gamma", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("search by app_id: got status %d: %s", w.Code, w.Body.String()) + } + var list []map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &list); err != nil { + t.Fatalf("unmarshal search: %v", err) + } + if len(list) != 1 { + t.Errorf("search by app_id result: got %d, want 1", len(list)) + } + if len(list) > 0 && list[0]["file_id"] != "00000000000000000000000000000010" { + t.Errorf("file_id=%v, want upload 10", list[0]["file_id"]) + } +} diff --git a/internal/admin/web.go b/internal/admin/web.go index 5d8e02d..d1fedc9 100644 --- a/internal/admin/web.go +++ b/internal/admin/web.go @@ -27,3 +27,12 @@ func uploadsWebHandler(c *gin.Context) { } c.Data(http.StatusOK, "text/html; charset=utf-8", data) } + +func submissionsWebHandler(c *gin.Context) { + data, err := webFS.ReadFile("web/submissions.html") + if err != nil { + c.String(http.StatusInternalServerError, "submissions.html: %v", err) + return + } + c.Data(http.StatusOK, "text/html; charset=utf-8", data) +} diff --git a/internal/admin/web/cluster.html b/internal/admin/web/cluster.html index d0d94d4..fedbcaf 100644 --- a/internal/admin/web/cluster.html +++ b/internal/admin/web/cluster.html @@ -117,10 +117,11 @@

spark-mcp-go Admin

- +
diff --git a/internal/admin/web/submissions.html b/internal/admin/web/submissions.html new file mode 100644 index 0000000..ee847a7 --- /dev/null +++ b/internal/admin/web/submissions.html @@ -0,0 +1,367 @@ + + + + + + Submissions - spark-mcp-go Admin + + + +
+
+

spark-mcp-go Admin

+ +
+
+ + +
+
+ +
+ + +
+ + + +
+ +
+ + + + + + + + + + + + + + + + + +
IDFile IDFile NameApp IDTracking URLClusterExitDuration (ms)Submitted At
Loading...
+
+
+ + +

Admin Token Required

+ +
+ + +
+ + +
+
+ + + + + diff --git a/internal/admin/web/uploads.html b/internal/admin/web/uploads.html index acd4de2..55c550e 100644 --- a/internal/admin/web/uploads.html +++ b/internal/admin/web/uploads.html @@ -103,10 +103,11 @@

spark-mcp-go Admin

- +
diff --git a/internal/storage/upload_repo.go b/internal/storage/upload_repo.go index e693a9f..7b09224 100644 --- a/internal/storage/upload_repo.go +++ b/internal/storage/upload_repo.go @@ -68,7 +68,8 @@ func (r *UploadRepo) Get(ctx context.Context, fileID string) (UploadMeta, error) // List returns upload index records ordered by uploaded_at descending. // -// If search is non-empty, name is filtered with a case-insensitive LIKE. +// If search is non-empty, file_name or any associated spark_submit app_id is +// filtered with a case-insensitive LIKE. // A limit of zero or less falls back to 100; values above 500 are capped. func (r *UploadRepo) List(ctx context.Context, search string, limit int) ([]UploadMeta, error) { if limit <= 0 { @@ -82,18 +83,19 @@ func (r *UploadRepo) List(ctx context.Context, search string, limit int) ([]Uplo var err error if search != "" { rows, err = r.db.sqlDB.QueryContext(ctx, ` - SELECT file_id, name, size, sha256, uploaded_at - FROM upload_files - WHERE name LIKE ? - ORDER BY uploaded_at DESC + SELECT DISTINCT u.file_id, u.name, u.size, u.sha256, u.uploaded_at + FROM upload_files u + LEFT JOIN spark_submissions s ON u.file_id = s.file_id + WHERE u.name LIKE ? OR s.app_id LIKE ? + ORDER BY u.uploaded_at DESC LIMIT ?`, - "%"+search+"%", limit, + "%"+search+"%", "%"+search+"%", limit, ) } else { rows, err = r.db.sqlDB.QueryContext(ctx, ` - SELECT file_id, name, size, sha256, uploaded_at - FROM upload_files - ORDER BY uploaded_at DESC + SELECT u.file_id, u.name, u.size, u.sha256, u.uploaded_at + FROM upload_files u + ORDER BY u.uploaded_at DESC LIMIT ?`, limit) } if err != nil { diff --git a/main.go b/main.go index f311459..c4d9682 100644 --- a/main.go +++ b/main.go @@ -105,7 +105,7 @@ func run() error { c.JSON(http.StatusOK, gin.H{"ok": true, "version": "0.0.0"}) }) - admin.Mount(r, db.Clusters(), db.Uploads(), audit.NewRepo(db), &uploadStore, cfg.AdminTokens) + admin.Mount(r, db.Clusters(), db.Uploads(), db.Submissions(), audit.NewRepo(db), &uploadStore, cfg.AdminTokens) mcpHandler, err := mcpsrv.Handler(&tools.Deps{ Logger: logger.With("component", "mcp"), @@ -119,6 +119,7 @@ func run() error { MaxResponseBytes: cfg.MaxResponseBytes, DataDir: cfg.DataDir, UploadStore: uploadStore, + SubmissionRepo: db.Submissions(), AnalyzerThresholds: analyzer.Thresholds{ DataSkewRatio: cfg.AnalyzerDataSkewRatio, GCPressureRatio: cfg.AnalyzerGCPressureRatio,