14 Commits
Author SHA1 Message Date
tao.chenandClaude 6f4c3a3e21 ci: Bark 推送逻辑提到独立脚本, 两 job 复用
CI / Release / CI (push) Successful in 3m15s
CI / Release / Release (push) Successful in 4m43s
之前 ci/release 各自 inline 拼 payload + curl, 重复 25 行 x 2.
device_key / endpoint / group / ttl 硬编码两次, 改一处忘另一处
会出现两个不同 group 的推送.

新建 .github/scripts/bark-notify.sh:
  bash .github/scripts/bark-notify.sh "<title>" "<body>"

脚本封装:
  - endpoint / device_key / group (Gitea) / isArchive / ttl 单一来源
  - heredoc 拼 JSON, 不依赖 jq
  - 错误用法打印 usage 并 exit 1 (set -euo pipefail)

workflow 两处都改成一行:
  bash .github/scripts/bark-notify.sh "CI \${REF_NAME} \${STATUS}" "\${BODY}"
  bash .github/scripts/bark-notify.sh "\${TAG_NAME} \${STATUS}" "\${BODY}"

emoji (/) 跟 status 映射仍在 workflow 里, 因为不同 job 拼
不同的 title prefix (CI vs release), 脚本不管这个, 调用方传什么
就推什么.

chmod +x 加在 commit 里, 但 git mode bit 在 .gitattributes 或
core.fileMode 关时可能丢. runner 一般会按 working tree 模式跑,
不需要额外 chmod 在 step 里.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-15 16:36:07 +08:00
tao.chenandClaude 44a30c4a8a ci: Bark 通知 group 改 Gitea, CI/Release 成功失败都推
CI / Release / CI (push) Successful in 2m52s
CI / Release / Release (push) Has been skipped
group 从 "release" 改为 "Gitea" — 不再按事件分类, 跟 Bark 设备
group 设置对齐.

两个 job (ci / release) 都加 if: always() Bark step, 任何结束
状态 (success / failure / cancelled) 都推送. 之前 release job
只在 success() 触发, 失败时收不到通知, 调试慢.

通知内容带 job.status:
  - 成功: 标题 "v1.0.0 success", body 加  emoji + release URL
  - 失败: 标题 "v1.0.0 failure", body 加  emoji + release URL
           (release job 失败时) 或 actions run URL (ci job 失败时)

Bark 端 group 分类: "Gitea" group 收所有 CI/release 推送, 用户
在 Bark 设备上一目了然.

\${{ job.status }} 是 GitHub Actions 标准 context, Gitea 兼容
(API-compatible).

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-15 16:33:41 +08:00
tao.chenandClaude 287d098e06 ci: release upload 改用 Gitea API 直调
CI / Release / CI (push) Successful in 1m55s
CI / Release / Release (push) Successful in 6m28s
softprops/action-gh-release@v2 在 Gitea 上偶发不工作, 改用
Gitea 原生 API:
  POST /repos/{owner}/{repo}/releases   <- 创建 release
  POST /repos/{owner}/{repo}/releases/{id}/assets  <- 上传 binary

token 走 secrets.GITHUB_TOKEN (Gitea 自动注入平台 token),
api_url 走 \${{ github.api_url }} (Gitea 自动解析).

RELEASE_ID 从 POST /releases 响应 grep 出来 (用 -oP "id":\s*\K[0-9]+).
- PCRE 模式 (grep -P) 依赖 GNU grep, ubuntu 镜像自带有, alpine
  runner 需要补 pcre 工具. 跟之前去 jq 一样是 runner 镜像依赖,
  这次保留因为已经在用 \${{ secrets.GITHUB_TOKEN }} 这种 GitHub
  风格 secret, runner 假定 ubuntu-family.

asset 上传字段是 attachment (Gitea API 文档要求, 不是 name).

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-15 09:37:57 +08:00
tao.chenandClaude e2bbbd5c45 ci: Bark 通知 group 改 release
CI / Release / CI (push) Successful in 1m49s
CI / Release / Release (push) Failing after 1m31s
之前用 "backup" 是从另一个 workflow 抄过来, 不准确. release 通
知归档到 "release" group, Bark 端按 group 分类更直观.

只改一处, 不动其他逻辑.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 19:22:00 +08:00
tao.chenandClaude 1cf4f90452 ci: release 通知 step 去掉 jq 依赖
CI / Release / CI (push) Successful in 5m42s
CI / Release / Release (push) Has been skipped
之前用 jq -n --arg 拼 JSON payload, 依赖 runner 镜像预装 jq.
Gitea act_runner 镜像可以自定义 (alpine / distroless / 极简
ubuntu), 不一定带 jq. 一旦缺, release 通知 step 静默失败
(curl 仍然发但 payload 是空字符串, Bark 端按 400/500 处理),
release 仍然成功上传但 user 收不到通知.

改用 heredoc 直接展开 shell 变量拼 JSON:
  PAYLOAD=$(cat <<EOF
  { "device_key": "...", "title": "${TITLE}", ... }
  EOF
  )

TITLE (tag 名, 形如 v0.0.1-beta) 和 BODY (release URL, 只含
https URL + tag 字符串) 都没有 ", \, 控制字符, heredoc 展开
安全. 不需要 jq 转义.

这一步跟 release upload 串行 (release job 本身顺序 steps), Bark
通知仍然是 release 成功的最后一步, if: success() 保留.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 19:21:18 +08:00
tao.chenandClaude 2bd67a5afc ci: release 后调 Bark 推送 release 通知
CI / Release / CI (push) Failing after 16m37s
CI / Release / Release (push) Has been skipped
release job 在 softprops/action-gh-release@v2 上传完
spark-mcp-linux-amd64 后, 调 Bark (http://101.43.40.124:81/push) 推
一条 device_key 绑定的通知, 标题用 github.ref_name (e.g.
v0.0.1-beta), body 带 release 页面 URL.

device_key 直接写进 workflow, 是公开的推送 key, group=backup
归档, ttl=3600 (1 小时后从 Bark 设备上消失).

if: success() 保证 release upload 失败时不会发通知, 避免推送
"已发布" 但实际没有的误导信息.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 19:19:35 +08:00
tao.chen 1460303b3e update: GOPROXY in release workflow
CI / Release / CI (push) Successful in 3m37s
CI / Release / Release (push) Failing after 1m27s
2026-07-14 19:11:43 +08:00
tao.chen 02499922b4 update: GOPROXY in building workflow
CI / Release / Release (push) Has been cancelled
CI / Release / CI (push) Has been cancelled
2026-07-14 19:06:53 +08:00
tao.chenandClaude 25c2fc66be ci: 改 workflow 用国内镜像装 Go + goproxy.cn 加速
CI / Release / CI (push) Successful in 7m45s
CI / Release / Release (push) Has been skipped
之前 ci/release 两个 job 用 actions/setup-go@v5 从 go.mod 装 Go,
内置 cache. 在 Gitea runner 所在网络区域 (ping 23ms 短, 推测
国内) 跑得动但慢.

改用用户提供的脚本手动装 Go 1.25.5 (NJU 镜像) + goproxy.cn 代理
+ GITHUB_PATH/GITHUB_ENV (Gitea runner 也兼容这俩 env var).
actions/setup-go 整个去掉, 因为手动装后 PATH 跟 GOPROXY 都设好,
go build 直接可跑.

GITBUB_PATH/GITHUB_ENV 是 GitHub Actions 的特殊 file path env var,
Gitea Actions 兼容 (api-compatible). 写到这两个文件等价于在
后续 step 设置 PATH 和 环境变量.

保留 go build / go vet / go test / gofmt -l . 四项 CI gate.
保留 release job 的 tag 触发 + softprops/action-gh-release@v2 上传.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 19:03:29 +08:00
tao.chenandClaude 80e644483a Add Gitea Actions CI/release workflow
CI / Release / Release (push) Has been cancelled
CI / Release / CI (push) Has been cancelled
The ci job triggers on every push to main and every pull request; the release job triggers only on tag pushes matching v* and needs the ci job to pass first.

We place the workflow under .github/workflows/ because Gitea Actions is API-compatible with GitHub Actions and discovers workflows in that directory.

The release build command is GOOS=linux GOARCH=amd64 go build -o spark-mcp-linux-amd64 ./main.go, producing the Linux amd64 artifact requested for deployment.

CI runs go build ./..., go vet ./..., go test ./..., and fails if gofmt -l . reports any unformatted files.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 16:11:27 +08:00
tao.chenandClaude 25f93dfb59 admin: submissions page, download endpoint, uploads search by app_id
Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 13:45:11 +08:00
tao.chenandClaude 573397dbb4 tools: persist spark_submit calls to SubmissionRepo
Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 13:45:07 +08:00
tao.chenandClaude a2232cecc4 storage: add spark_submissions table and SubmissionRepo
Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 13:45:03 +08:00
tao.chenandClaude d6e3952a4b executor: parse tracking_url and error on missing application_id
Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 13:45:01 +08:00
22 changed files with 1532 additions and 87 deletions
+37
View File
@@ -0,0 +1,37 @@
#!/usr/bin/env bash
# bark-notify.sh — push a notification to the local Bark server.
#
# Usage: bark-notify.sh <title> <body>
#
# The Bark endpoint, device_key, and group are pinned in this script
# (it is a deployment detail, not a secret). The group is "Gitea" —
# change it here if you want to sort notifications by category.
set -euo pipefail
if [ "$#" -ne 2 ]; then
echo "usage: $0 <title> <body>" >&2
exit 1
fi
TITLE="$1"
BODY="$2"
PAYLOAD=$(cat <<EOF
{
"device_key": "6ZFVsxM95cuAjYoNLWSWaN",
"title": "${TITLE}",
"body": "${BODY}",
"group": "Gitea",
"isArchive": 1,
"ttl": 3600
}
EOF
)
curl -s \
-X POST http://101.43.40.124:81/push \
-H "Content-Type: application/json; charset=utf-8" \
-d "${PAYLOAD}"
echo "bark notified: ${TITLE}"
+132
View File
@@ -0,0 +1,132 @@
# This workflow runs on Gitea Actions, which is API-compatible with GitHub Actions.
# It performs CI on every push to main and every pull request, and automatically
# builds and attaches a Linux amd64 release binary when a version tag (v*) is pushed.
#
# Go is installed manually from a Chinese mirror (NJU) instead of using
# actions/setup-go, because the runner is in a region where the official Go
# download endpoint is slow. GOPROXY is set to goproxy.cn for the same reason.
name: CI / Release
on:
push:
branches:
- main
tags:
- 'v*'
pull_request:
jobs:
ci:
name: CI
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Install Go (Mirror in China)
run: |
if [ ! -d "/usr/local/go" ]; then
wget https://mirrors.nju.edu.cn/golang/go1.25.5.linux-amd64.tar.gz
sudo tar -C /usr/local -xzf go1.25.5.linux-amd64.tar.gz
rm go1.25.5.linux-amd64.tar.gz
fi
echo "/usr/local/go/bin" >> $GITHUB_PATH
echo "GOPROXY=https://goproxy.cn/,https://goproxy.io,direct" >> $GITHUB_ENV
- name: Build
run: go build ./...
- name: Vet
run: go vet ./...
- name: Test
run: go test ./...
- name: Check formatting
run: |
if [ -n "$(gofmt -l .)" ]; then
gofmt -l .
exit 1
fi
- name: Notify CI via Bark
if: always()
run: |
REF_NAME="${{ github.ref_name }}"
STATUS="${{ job.status }}"
if [ "${STATUS}" = "success" ]; then
EMOJI="✅"
else
EMOJI="❌"
fi
BODY="${EMOJI} ${REF_NAME} ${STATUS}: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${{ github.run_id }}"
bash .github/scripts/bark-notify.sh "CI ${REF_NAME} ${STATUS}" "${BODY}"
release:
name: Release
needs: ci
if: startsWith(github.ref, 'refs/tags/v')
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Install Go (Mirror in China)
run: |
if [ ! -d "/usr/local/go" ]; then
wget https://mirrors.nju.edu.cn/golang/go1.25.5.linux-amd64.tar.gz
sudo tar -C /usr/local -xzf go1.25.5.linux-amd64.tar.gz
rm go1.25.5.linux-amd64.tar.gz
fi
echo "/usr/local/go/bin" >> $GITHUB_PATH
echo "GOPROXY=https://goproxy.cn/,https://goproxy.io,direct" >> $GITHUB_ENV
- name: Build release binary
run: GOOS=linux GOARCH=amd64 go build -o spark-mcp-linux-amd64 ./main.go
- name: Upload release asset via Gitea API
env:
GITEA_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
# 1. 获取当前 Tag 名字 (例如 v0.0.1-beta)
TAG_NAME="${{ github.ref_name }}"
echo "正在为 Tag ${TAG_NAME} 创建 Gitea Release..."
# 2. 调用 Gitea API 创建一个 Release 并获取其 ID
# 注意:${{ github.api_url }} 会自动解析为你本地 Gitea 的 API 地址 (如 http://your-gitea/api/v1)
RELEASE_ID=$(curl -X POST "${{ github.api_url }}/repos/${{ github.repository }}/releases" \
-H "accept: application/json" \
-H "authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d "{
\"tag_name\": \"${TAG_NAME}\",
\"target_commitish\": \"${{ github.sha }}\",
\"name\": \"Release ${TAG_NAME}\",
\"body\": \"Auto generated release assets.\",
\"draft\": false,
\"prerelease\": false
}" | grep -oP '"id":\s*\K[0-9]+' | head -n 1)
echo "Release 创建成功,ID 为: ${RELEASE_ID}"
# 3. 将你的二进制文件上传到该 Release 中
echo "开始上传文件 spark-mcp-linux-amd64 ..."
curl -X POST "${{ github.api_url }}/repos/${{ github.repository }}/releases/${RELEASE_ID}/assets" \
-H "accept: application/json" \
-H "authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: multipart/form-data" \
-F "attachment=@spark-mcp-linux-amd64"
- name: Notify release via Bark
if: always()
run: |
TITLE="${{ github.ref_name }}"
STATUS="${{ job.status }}"
if [ "${STATUS}" = "success" ]; then
EMOJI="✅"
else
EMOJI="❌"
fi
BODY="${EMOJI} spark-mcp ${TITLE} ${STATUS}: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/releases/tag/${TITLE}"
bash .github/scripts/bark-notify.sh "${TITLE} ${STATUS}" "${BODY}"
+80 -1
View File
@@ -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")
+1 -1
View File
@@ -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)
}
+155
View File
@@ -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)
}
}
+48 -6
View File
@@ -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"])
}
}
+9
View File
@@ -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)
}
+5 -4
View File
@@ -117,10 +117,11 @@
<header>
<div style='display:flex;align-items:center;gap:1rem;'>
<h1>spark-mcp-go Admin</h1>
<nav>
<a href='/admin' class='active'>Clusters</a>
<a href='/admin/uploads'>Uploads</a>
</nav>
<nav>
<a href='/admin' class='active'>Clusters</a>
<a href='/admin/uploads'>Uploads</a>
<a href='/admin/submissions'>Submissions</a>
</nav>
</div>
<div class='auth'>
<button id='set-token' class='secondary'>Set Token</button>
+367
View File
@@ -0,0 +1,367 @@
<!DOCTYPE html>
<html lang='en'>
<head>
<meta charset='utf-8'>
<meta name='viewport' content='width=device-width, initial-scale=1'>
<title>Submissions - spark-mcp-go Admin</title>
<style>
:root {
--bg: #0d1117;
--panel: #161b22;
--border: #30363d;
--text: #c9d1d9;
--muted: #8b949e;
--accent: #58a6ff;
--danger: #f85149;
--ok: #3fb950;
}
* { box-sizing: border-box; }
body {
margin: 0;
font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, 'Liberation Mono', monospace;
background: var(--bg);
color: var(--text);
line-height: 1.5;
}
header {
display: flex;
align-items: center;
justify-content: space-between;
gap: 1rem;
padding: 1rem;
border-bottom: 1px solid var(--border);
background: var(--panel);
flex-wrap: wrap;
}
header h1 { margin: 0; font-size: 1.1rem; color: var(--accent); }
nav { display: flex; gap: .75rem; }
nav a { color: var(--muted); text-decoration: none; }
nav a:hover { color: var(--text); }
nav a.active { color: var(--accent); }
.auth { display: flex; align-items: center; gap: .5rem; flex-wrap: wrap; }
input[type='text'], input[type='password'] {
background: var(--bg);
color: var(--text);
border: 1px solid var(--border);
border-radius: 4px;
padding: .4rem .5rem;
font: inherit;
}
input[type='text'] { min-width: 260px; }
button {
background: var(--accent);
color: #fff;
border: none;
border-radius: 4px;
padding: .4rem .8rem;
font: inherit;
cursor: pointer;
}
button:hover { opacity: .9; }
button.danger { background: var(--danger); }
button.secondary { background: var(--border); color: var(--text); }
main { padding: 1rem; max-width: 1400px; margin: 0 auto; }
.toolbar { display: flex; gap: .5rem; margin-bottom: 1rem; align-items: center; flex-wrap: wrap; }
.toolbar span { color: var(--muted); }
.error {
background: rgba(248, 81, 73, .15);
border: 1px solid var(--danger);
color: var(--danger);
padding: .75rem;
border-radius: 4px;
margin-bottom: 1rem;
white-space: pre-wrap;
}
.hidden { display: none; }
table { width: 100%; border-collapse: collapse; }
th, td { text-align: left; padding: .5rem; border-bottom: 1px solid var(--border); }
th { color: var(--muted); font-weight: 600; user-select: none; }
th.sortable { cursor: pointer; }
th.sortable:hover { color: var(--text); }
td { vertical-align: middle; font-size: .9rem; }
td.mono { font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; }
td.wrap { word-break: break-all; }
.actions { display: flex; gap: .4rem; }
dialog.modal {
border: 1px solid var(--border);
background: var(--panel);
color: var(--text);
border-radius: 6px;
padding: 1.5rem;
min-width: 360px;
max-width: 90vw;
}
dialog.modal::backdrop { background: rgba(0, 0, 0, .65); }
dialog.modal h2 { margin: 0 0 .5rem; font-size: 1.1rem; color: var(--accent); }
dialog.modal form { display: flex; flex-direction: column; gap: .75rem; }
dialog.modal label { color: var(--muted); font-size: .9rem; }
dialog.modal input { width: 100%; }
.modal-error { color: var(--danger); font-size: .9rem; margin: 0; }
.modal-actions { display: flex; gap: .5rem; justify-content: flex-end; }
</style>
</head>
<body>
<header>
<div style='display:flex;align-items:center;gap:1rem;'>
<h1>spark-mcp-go Admin</h1>
<nav>
<a href='/admin'>Clusters</a>
<a href='/admin/uploads'>Uploads</a>
<a href='/admin/submissions' class='active'>Submissions</a>
</nav>
</div>
<div class='auth'>
<button id='set-token' class='secondary'>Set Token</button>
<button id='logout' class='danger'>Logout</button>
</div>
</header>
<main>
<div id='error' class='error hidden'></div>
<section class='toolbar'>
<input id='search' type='text' placeholder='Search by file name or app id...'>
<button id='refresh' class='secondary'>Refresh</button>
<span id='count'></span>
</section>
<section>
<table>
<thead>
<tr>
<th class='sortable' data-key='id'>ID</th>
<th class='sortable' data-key='file_id'>File ID</th>
<th class='sortable' data-key='file_name'>File Name</th>
<th class='sortable' data-key='app_id'>App ID</th>
<th>Tracking URL</th>
<th class='sortable' data-key='cluster_id'>Cluster</th>
<th class='sortable' data-key='exit_code'>Exit</th>
<th class='sortable' data-key='duration_ms'>Duration (ms)</th>
<th class='sortable' data-key='submitted_at'>Submitted At</th>
</tr>
</thead>
<tbody id='submissions-body'>
<tr><td colspan='9'>Loading...</td></tr>
</tbody>
</table>
</section>
</main>
<dialog id='token-modal' class='modal'>
<h2>Admin Token Required</h2>
<p id='modal-error' class='modal-error hidden'></p>
<form id='token-form'>
<label for='modal-token'>Bearer token</label>
<input id='modal-token' type='password' autocomplete='off' required>
<div class='modal-actions'>
<button type='submit' id='modal-save'>Save</button>
<button type='button' id='modal-cancel' class='secondary hidden'>Cancel</button>
</div>
</form>
</dialog>
<script>
const $ = (sel) => document.querySelector(sel);
const API = '/admin/submissions';
let submissions = [];
let sortKey = 'submitted_at';
let sortDir = -1;
let pendingRetry = null;
let cancelBlocker = null;
function headers() {
return {
'Authorization': 'Bearer ' + (localStorage.getItem('adminToken') || ''),
'Content-Type': 'application/json'
};
}
function showError(msg) {
const el = $('#error');
el.textContent = msg || '';
el.classList.toggle('hidden', !msg);
}
function showModalError(msg) {
const el = $('#modal-error');
el.textContent = msg || '';
el.classList.toggle('hidden', !msg);
}
function openTokenModal(opts = {}) {
const dialog = $('#token-modal');
const cancelBtn = $('#modal-cancel');
showModalError(opts.error || '');
cancelBtn.classList.toggle('hidden', !opts.allowCancel);
if (cancelBlocker) {
dialog.removeEventListener('cancel', cancelBlocker);
cancelBlocker = null;
}
if (!opts.allowCancel) {
cancelBlocker = (e) => e.preventDefault();
dialog.addEventListener('cancel', cancelBlocker);
}
dialog.showModal();
$('#modal-token').focus();
}
function closeTokenModal() {
const dialog = $('#token-modal');
showModalError('');
if (cancelBlocker) {
dialog.removeEventListener('cancel', cancelBlocker);
cancelBlocker = null;
}
dialog.close();
$('#modal-token').value = '';
}
function saveTokenFromModal(e) {
e.preventDefault();
localStorage.setItem('adminToken', $('#modal-token').value.trim());
closeTokenModal();
if (pendingRetry) {
const fn = pendingRetry;
pendingRetry = null;
fn();
} else {
loadSubmissions();
}
}
async function api(method, path, body) {
const opts = { method, headers: headers() };
if (body !== undefined) opts.body = JSON.stringify(body);
const res = await fetch(API + path, opts);
if (!res.ok) {
if (res.status === 401) {
localStorage.removeItem('adminToken');
pendingRetry = () => api(method, path, body);
openTokenModal({
error: 'Token rejected by server (401). Enter a valid token.',
allowCancel: false
});
throw new Error('401 Unauthorized');
}
let detail = res.statusText;
try {
const j = await res.json();
if (j.error) detail = j.error;
} catch (_) {}
throw new Error(`${res.status} ${detail}`);
}
if (res.status === 204) return null;
return res.json();
}
function formatTime(ts) {
if (!ts) return '';
const d = new Date(ts);
return isNaN(d) ? ts : d.toISOString();
}
function truncate(s, n) {
if (!s || s.length <= n) return s || '';
return s.slice(0, n) + '…';
}
function escapeHtml(s) {
return String(s)
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
.replace(/"/g, '&quot;')
.replace(/'/g, '&#39;');
}
async function loadSubmissions() {
showError('');
try {
const search = $('#search').value.trim();
const path = search ? `/api?search=${encodeURIComponent(search)}` : '/api';
submissions = await api('GET', path);
renderList();
} catch (e) {
showError('Failed to load submissions: ' + e.message);
}
}
function compare(a, b) {
let av = a[sortKey];
let bv = b[sortKey];
if (sortKey === 'id' || sortKey === 'exit_code' || sortKey === 'duration_ms') {
av = Number(av); bv = Number(bv);
} else if (sortKey === 'submitted_at') {
av = new Date(av).getTime(); bv = new Date(bv).getTime();
} else {
av = String(av).toLowerCase(); bv = String(bv).toLowerCase();
}
if (av < bv) return -sortDir;
if (av > bv) return sortDir;
return 0;
}
function renderList() {
const tbody = $('#submissions-body');
const list = submissions.slice().sort(compare);
$('#count').textContent = list.length + ' submission' + (list.length === 1 ? '' : 's');
tbody.innerHTML = '';
if (!list.length) {
tbody.innerHTML = '<tr><td colspan="9">No submissions.</td></tr>';
return;
}
for (const s of list) {
const tr = document.createElement('tr');
const fileLink = s.file_id ? `<a href='/admin/uploads/${encodeURIComponent(s.file_id)}'>${escapeHtml(truncate(s.file_id, 8))}</a>` : escapeHtml(truncate(s.file_id, 8));
tr.innerHTML = `
<td class='mono'>${escapeHtml(s.id)}</td>
<td class='mono' title='${escapeHtml(s.file_id)}'>${fileLink}</td>
<td>${escapeHtml(s.file_name || '')}</td>
<td class='mono'>${escapeHtml(s.app_id)}</td>
<td class='wrap'>${escapeHtml(s.tracking_url || '')}</td>
<td>${escapeHtml(s.cluster_id)}</td>
<td>${escapeHtml(s.exit_code)}</td>
<td>${escapeHtml(s.duration_ms)}</td>
<td>${escapeHtml(formatTime(s.submitted_at))}</td>
`;
tbody.appendChild(tr);
}
}
function handleSort(e) {
const key = e.target.dataset.key;
if (!key) return;
if (sortKey === key) {
sortDir = -sortDir;
} else {
sortKey = key;
sortDir = key === 'submitted_at' ? -1 : 1;
}
document.querySelectorAll('th.sortable').forEach(th => {
th.textContent = th.textContent.replace(/ [▲▼]$/, '');
});
const marker = sortDir > 0 ? ' ▲' : ' ▼';
e.target.textContent = e.target.textContent.replace(/ [▲▼]$/, '') + marker;
renderList();
}
$('#search').addEventListener('input', loadSubmissions);
$('#refresh').addEventListener('click', loadSubmissions);
$('#set-token').addEventListener('click', () => openTokenModal({ allowCancel: true }));
$('#logout').addEventListener('click', () => {
localStorage.removeItem('adminToken');
openTokenModal({ error: 'Token cleared. Enter a new token to continue.', allowCancel: false });
});
$('#token-form').addEventListener('submit', saveTokenFromModal);
$('#modal-cancel').addEventListener('click', closeTokenModal);
document.querySelectorAll('th.sortable').forEach(th => th.addEventListener('click', handleSort));
if (!localStorage.getItem('adminToken')) {
openTokenModal({ allowCancel: false });
} else {
loadSubmissions();
}
</script>
</body>
</html>
+5 -4
View File
@@ -103,10 +103,11 @@
<header>
<div style='display:flex;align-items:center;gap:1rem;'>
<h1>spark-mcp-go Admin</h1>
<nav>
<a href='/admin'>Clusters</a>
<a href='/admin/uploads' class='active'>Uploads</a>
</nav>
<nav>
<a href='/admin'>Clusters</a>
<a href='/admin/uploads' class='active'>Uploads</a>
<a href='/admin/submissions'>Submissions</a>
</nav>
</div>
<div class='auth'>
<button id='set-token' class='secondary'>Set Token</button>
+68 -21
View File
@@ -23,10 +23,13 @@ import (
// application application_xxx" line and a few progress lines.
const stdoutCap = 10 << 20
// appIDPattern matches the YARN line printed on successful submission:
//
// "Submitted application application_12345_0001"
var appIDPattern = regexp.MustCompile(`application_\d+_\d+`)
// submittedAppPattern matches the explicit success line, including the
// application id captured in group 1. This mirrors the Python reference.
var submittedAppPattern = regexp.MustCompile(`Submitted application (\S+)`)
// trackingURLPattern matches "tracking URL: <url>" on stderr. The URL tail
// is used as a fallback source for the application id.
var trackingURLPattern = regexp.MustCompile(`tracking URL:\s+(\S+)`)
// SparkSubmitOpts is the resolved command line for a single spark-submit run.
//
@@ -43,11 +46,12 @@ type SparkSubmitOpts struct {
// Result is what spark_submit returns to the LLM.
type Result struct {
AppID string `json:"app_id,omitempty"`
ExitCode int `json:"exit_code"`
StdoutTail string `json:"stdout_tail"`
StderrTail string `json:"stderr_tail"`
DurationMS int64 `json:"duration_ms"`
AppID string `json:"app_id,omitempty"`
TrackingURL string `json:"tracking_url,omitempty"`
ExitCode int `json:"exit_code"`
StdoutTail string `json:"stdout_tail"`
StderrTail string `json:"stderr_tail"`
DurationMS int64 `json:"duration_ms"`
}
// ErrBinaryEmpty is returned when the cluster has no spark-submit binary
@@ -90,27 +94,70 @@ func Run(ctx context.Context, opts SparkSubmitOpts) (*Result, error) {
DurationMS: dur.Milliseconds(),
}
// app_id: prefer stdout, fall back to stderr. Both are searched only on
// the tail (cap-bounded) so we don't pay for the full output.
if id := matchAppID(res.StdoutTail); id != "" {
res.AppID = id
} else if id := matchAppID(res.StderrTail); id != "" {
res.AppID = id
}
appID, trackingURL, ok := parseSparkSubmitOutput(res.StdoutTail, res.StderrTail)
res.AppID = appID
res.TrackingURL = trackingURL
if !ok {
return res, fmt.Errorf("executor: spark-submit ran but no application_id in output (exit_code=%d)", res.ExitCode)
}
if err != nil {
// Wrap the underlying error but keep the parsed result so the LLM
// can see exit_code and stderr even on failure.
return res, fmt.Errorf("executor: spark-submit failed: %w", err)
}
return res, nil
}
func matchAppID(s string) string {
if s == "" {
// parseSparkSubmitOutput extracts the application id and tracking URL from
// spark-submit output. It mirrors the Python reference implementation: first
// look for "Submitted application <id>", then fall back to "tracking URL:",
// deriving the id from the last path segment. The returned ok is false when
// no application id could be determined.
func parseSparkSubmitOutput(stdout, stderr string) (appID, trackingURL string, ok bool) {
// Stage 1: explicit success line, prefer stdout then stderr.
var trackingFound bool
if m := submittedAppPattern.FindStringSubmatch(stdout); m != nil {
appID = m[1]
ok = true
} else if m := submittedAppPattern.FindStringSubmatch(stderr); m != nil {
appID = m[1]
ok = true
}
// Stage 2: tracking URL fallback (stderr only, like the Python reference).
if !ok {
if m := trackingURLPattern.FindStringSubmatch(stderr); m != nil {
trackingURL = m[1]
trackingFound = true
appID = extractAppIDFromTrackingURL(trackingURL)
ok = appID != ""
if !ok {
trackingURL = ""
}
}
}
// Tracking URL is independent: if present anywhere on stderr, capture it.
if trackingURL == "" && !trackingFound {
if m := trackingURLPattern.FindStringSubmatch(stderr); m != nil {
trackingURL = m[1]
}
}
return
}
// extractAppIDFromTrackingURL returns the last path segment if it starts with
// "application_", otherwise an empty string.
func extractAppIDFromTrackingURL(rawURL string) string {
trimmed := strings.TrimRight(rawURL, "/")
idx := strings.LastIndex(trimmed, "/")
if idx < 0 {
return ""
}
return appIDPattern.FindString(s)
tail := trimmed[idx+1:]
if strings.HasPrefix(tail, "application_") {
return tail
}
return ""
}
// exitCodeFromErr returns 0 on nil, the process's exit code on *exec.ExitError,
@@ -0,0 +1,79 @@
package executor
import "testing"
func TestParseAppID_FromStdout(t *testing.T) {
stdout := "some prefix\nSubmitted application application_123_456\n"
stderr := "random log\n"
appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr)
if !ok {
t.Fatal("expected ok")
}
if appID != "application_123_456" {
t.Errorf("appID=%q, want application_123_456", appID)
}
if trackingURL != "" {
t.Errorf("trackingURL=%q, want empty", trackingURL)
}
}
func TestParseAppID_FromStderr(t *testing.T) {
stdout := ""
stderr := "Submitted application application_123_456\n"
appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr)
if !ok {
t.Fatal("expected ok")
}
if appID != "application_123_456" {
t.Errorf("appID=%q, want application_123_456", appID)
}
if trackingURL != "" {
t.Errorf("trackingURL=%q, want empty", trackingURL)
}
}
func TestParseAppID_FromTrackingURL(t *testing.T) {
stdout := ""
stderr := "Log line\ntracking URL: http://rm.example.com:8088/proxy/application_123_456/\n"
appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr)
if !ok {
t.Fatal("expected ok")
}
if appID != "application_123_456" {
t.Errorf("appID=%q, want application_123_456", appID)
}
wantURL := "http://rm.example.com:8088/proxy/application_123_456/"
if trackingURL != wantURL {
t.Errorf("trackingURL=%q, want %q", trackingURL, wantURL)
}
}
func TestParseAppID_NoAppID(t *testing.T) {
stdout := ""
stderr := "some random log"
appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr)
if ok {
t.Fatal("expected not ok")
}
if appID != "" {
t.Errorf("appID=%q, want empty", appID)
}
if trackingURL != "" {
t.Errorf("trackingURL=%q, want empty", trackingURL)
}
}
func TestParseAppID_BadTrackingURL(t *testing.T) {
stdout := ""
stderr := "tracking URL: http://rm.example.com:8088/proxy/not-an-app/\n"
appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr)
if ok {
t.Fatal("expected not ok")
}
if appID != "" {
t.Errorf("appID=%q, want empty", appID)
}
if trackingURL != "" {
t.Errorf("trackingURL=%q, want empty", trackingURL)
}
}
+13 -9
View File
@@ -30,7 +30,7 @@ func TestRun_FakeBinary(t *testing.T) {
res, err := Run(ctx, SparkSubmitOpts{
Binary: "/bin/echo",
Args: []string{"--master", "yarn; rm -rf /tmp/this-should-not-exist"},
Args: []string{"--master", "yarn; rm -rf /tmp/this-should-not-exist", "Submitted application application_12345_0001"},
Timeout: 3 * time.Second,
})
if err != nil {
@@ -43,8 +43,8 @@ func TestRun_FakeBinary(t *testing.T) {
if !strings.Contains(res.StdoutTail, "yarn; rm -rf /tmp/this-should-not-exist") {
t.Errorf("stdout missing literal arg, got: %q", res.StdoutTail)
}
if res.AppID != "" {
t.Errorf("AppID should be empty for echo, got %q", res.AppID)
if res.AppID != "application_12345_0001" {
t.Errorf("AppID = %q, want application_12345_0001", res.AppID)
}
}
@@ -66,22 +66,26 @@ func TestRun_NonZeroExit(t *testing.T) {
}
}
// TestMatchAppID locks down the regex so a YARN output tweak doesn't
// TestSubmittedAppPattern locks down the regex so a YARN output tweak doesn't
// silently break app_id extraction.
func TestMatchAppID(t *testing.T) {
func TestSubmittedAppPattern(t *testing.T) {
cases := []struct {
in, want string
}{
{"Submitted application application_12345_0001", "application_12345_0001"},
{"... some prefix application_999_42 ... tail", "application_999_42"},
{"Submitted application application_999_42 extra", "application_999_42"},
{"no app id here", ""},
{"", ""},
{"application_1_2 application_3_4", "application_1_2"}, // first match
{"Submitted application app_1_2 Submitted application app_3_4", "app_1_2"}, // first match
}
for _, tc := range cases {
got := matchAppID(tc.in)
m := submittedAppPattern.FindStringSubmatch(tc.in)
var got string
if m != nil {
got = m[1]
}
if got != tc.want {
t.Errorf("matchAppID(%q) = %q, want %q", tc.in, got, tc.want)
t.Errorf("submittedAppPattern(%q) = %q, want %q", tc.in, got, tc.want)
}
}
}
+1
View File
@@ -16,6 +16,7 @@ type Deps struct {
Logger *slog.Logger
AuditRepo *audit.Repo
ClusterRepo *storage.ClusterRepo
SubmissionRepo *storage.SubmissionRepo
SparkSubmitTimeout time.Duration
HTTPClient *httpclient.Client
MaxResponseBytes int64
+8 -7
View File
@@ -25,7 +25,7 @@ import (
// testDepsWithDataDir returns dependencies backed by an in-memory DB and a
// temporary data directory.
func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo, *audit.Repo) {
func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo, *storage.SubmissionRepo, *audit.Repo) {
t.Helper()
db, err := storage.Open(":memory:")
if err != nil {
@@ -46,10 +46,11 @@ func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo, *audit.Repo
}),
AuditRepo: audit.NewRepo(db),
ClusterRepo: db.Clusters(),
SubmissionRepo: db.Submissions(),
MaxResponseBytes: 1 << 20,
DataDir: dataDir,
UploadStore: uploadStore,
}, db.Clusters(), audit.NewRepo(db)
}, db.Clusters(), db.Submissions(), audit.NewRepo(db)
}
// createCluster creates a cluster in the repository with the given fields.
@@ -100,7 +101,7 @@ func TestFetchURL(t *testing.T) {
}))
defer srv.Close()
deps, repo, _ := testDepsWithDataDir(t)
deps, repo, _, _ := testDepsWithDataDir(t)
createCluster(t, repo, &cluster.Cluster{
ID: "cluster-a",
Name: "Cluster A",
@@ -294,7 +295,7 @@ func TestFetchURL_RedirectPreservesAuth(t *testing.T) {
}))
defer server1.Close()
deps, repo, _ := testDepsWithDataDir(t)
deps, repo, _, _ := testDepsWithDataDir(t)
createCluster(t, repo, &cluster.Cluster{
ID: "cluster-redirect",
Name: "Redirect",
@@ -323,7 +324,7 @@ func TestFetchURL_RedirectPreservesAuth(t *testing.T) {
}
func TestUploadFile(t *testing.T) {
deps, _, auditRepo := testDepsWithDataDir(t)
deps, _, _, auditRepo := testDepsWithDataDir(t)
req := newToolRequest(UploadFileName, map[string]any{
"filename": "hello.txt",
@@ -387,7 +388,7 @@ func TestUploadFile(t *testing.T) {
}
func TestUploadFile_PathTraversal(t *testing.T) {
deps, _, _ := testDepsWithDataDir(t)
deps, _, _, _ := testDepsWithDataDir(t)
cases := []string{
"../../../etc/passwd",
@@ -418,7 +419,7 @@ func TestUploadFile_PathTraversal(t *testing.T) {
}
func TestUploadFile_Base64(t *testing.T) {
deps, _, _ := testDepsWithDataDir(t)
deps, _, _, _ := testDepsWithDataDir(t)
req := newToolRequest(UploadFileName, map[string]any{
"filename": "hello.bin",
+44 -18
View File
@@ -6,10 +6,12 @@ import (
"log/slog"
"sort"
"strconv"
"time"
"github.com/mark3labs/mcp-go/mcp"
"spark-mcp-go/internal/executor"
"spark-mcp-go/internal/storage"
)
const SparkSubmitName = "spark_submit"
@@ -98,7 +100,8 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
// letting spark-submit produce a confusing FileNotFoundException later.
// Placed after all required-string parses so error messages surface in
// field order (cluster_id, master, ..., script_path, queue, ...).
if _, err := d.UploadStore.Validate(scriptPath); err != nil {
fileID, err := d.UploadStore.Validate(scriptPath)
if err != nil {
if d.Logger != nil {
d.Logger.Warn("spark_submit.unminted_path_rejected", slog.String("path", scriptPath), slog.String("err", err.Error()))
}
@@ -173,35 +176,58 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
})
if err != nil {
callLog.WithError(err)
// Even on error, result carries ExitCode/Stderr for the LLM.
if result != nil {
callLog.WithResult(map[string]any{
"binary": binary,
"argv": cmd,
"app_id": result.AppID,
"exit_code": result.ExitCode,
"duration_ms": result.DurationMS,
})
callLog.WithResult(sparkSubmitResultLog(binary, cmd, result))
return textResult(encodeJSON(map[string]any{
"app_id": result.AppID,
"exit_code": result.ExitCode,
"stdout_tail": result.StdoutTail,
"stderr_tail": result.StderrTail,
"duration_ms": result.DurationMS,
"error": err.Error(),
"app_id": result.AppID,
"tracking_url": result.TrackingURL,
"exit_code": result.ExitCode,
"stdout_tail": result.StdoutTail,
"stderr_tail": result.StderrTail,
"duration_ms": result.DurationMS,
"error": err.Error(),
})), nil
}
return errResult("spark_submit: " + err.Error()), nil
}
callLog.WithResult(map[string]any{
if result.AppID != "" && d.SubmissionRepo != nil {
sub := storage.Submission{
FileID: fileID,
AppID: result.AppID,
TrackingURL: result.TrackingURL,
ClusterID: clusterID,
ExitCode: result.ExitCode,
DurationMS: result.DurationMS,
SubmittedAt: time.Now(),
}
if createErr := d.SubmissionRepo.Create(ctx, sub); createErr != nil {
if d.Logger != nil {
d.Logger.Warn("spark_submit.persist_failed", slog.String("file_id", fileID), slog.String("app_id", result.AppID), slog.String("err", createErr.Error()))
} else {
slog.Default().Warn("spark_submit.persist_failed", slog.String("file_id", fileID), slog.String("app_id", result.AppID), slog.String("err", createErr.Error()))
}
}
}
callLog.WithResult(sparkSubmitResultLog(binary, cmd, result))
return textResult(encodeJSON(result)), nil
}
// sparkSubmitResultLog builds the standard result map used for both logging
// and the admin audit trail.
func sparkSubmitResultLog(binary string, cmd []string, result *executor.Result) map[string]any {
m := map[string]any{
"binary": binary,
"argv": cmd,
"app_id": result.AppID,
"exit_code": result.ExitCode,
"duration_ms": result.DurationMS,
})
return textResult(encodeJSON(result)), nil
}
if result.TrackingURL != "" {
m["tracking_url"] = result.TrackingURL
}
return m
}
// SparkSubmitCommandOpts holds the structured arguments used to build the
+109 -6
View File
@@ -10,11 +10,15 @@ import (
"github.com/mark3labs/mcp-go/mcp"
"spark-mcp-go/internal/audit"
"spark-mcp-go/internal/cluster"
"spark-mcp-go/internal/httpclient"
"spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
)
func TestSparkSubmit_StructuredCommand(t *testing.T) {
deps, repo, _ := testDepsWithDataDir(t)
deps, repo, _, _ := testDepsWithDataDir(t)
deps.SparkSubmitTimeout = 5 * time.Second
store := deps.UploadStore
@@ -25,7 +29,7 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
echoed := filepath.Join(t.TempDir(), "echoed")
echoScript := filepath.Join(t.TempDir(), "echo-args.sh")
if err := os.WriteFile(echoScript, []byte("#!/bin/sh\nfor arg do\n echo \"$arg\" >> \"$ECHO_FILE\"\ndone\n"), 0o755); err != nil {
if err := os.WriteFile(echoScript, []byte("#!/bin/sh\nfor arg do\n echo \"$arg\" >> \"$ECHO_FILE\"\ndone\necho 'Submitted application application_test_0001'\n"), 0o755); err != nil {
t.Fatalf("write echo script: %v", err)
}
@@ -97,8 +101,107 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
}
}
func TestSparkSubmit_PersistsSubmission(t *testing.T) {
// Open a dedicated DB so we can insert upload_files metadata for the join.
db, err := storage.Open(":memory:")
if err != nil {
t.Fatalf("open db: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
dataDir := t.TempDir()
store, err := uploads.New(filepath.Join(dataDir, "uploads"))
if err != nil {
t.Fatalf("new upload store: %v", err)
}
deps := &Deps{
HTTPClient: httpclient.New(httpclient.Config{Timeout: 5 * time.Second, MaxResponseBytes: 1 << 20}),
AuditRepo: audit.NewRepo(db),
ClusterRepo: db.Clusters(),
SubmissionRepo: db.Submissions(),
MaxResponseBytes: 1 << 20,
DataDir: dataDir,
UploadStore: store,
SparkSubmitTimeout: 5 * time.Second,
}
repo := db.Clusters()
subRepo := db.Submissions()
fileID, _, _, _, absPath, err := store.Save([]byte("print('hi')\n"), "hello.py")
if err != nil {
t.Fatalf("save upload: %v", err)
}
if err := db.Uploads().Create(context.Background(), fileID, "hello.py", 14, "dummy-sha", time.Now()); err != nil {
t.Fatalf("create upload record: %v", err)
}
echoScript := filepath.Join(t.TempDir(), "echo-submit.sh")
if err := os.WriteFile(echoScript, []byte("#!/bin/sh\necho 'tracking URL: http://rm.example.com:8088/proxy/application_persist_0001/' >&2\necho 'Submitted application application_persist_0001'\n"), 0o755); err != nil {
t.Fatalf("write echo script: %v", err)
}
createCluster(t, repo, &cluster.Cluster{
ID: "cluster-persist",
Name: "Persist",
IsActive: true,
AuthType: cluster.AuthNone,
SparkSubmitExecuteBin: echoScript,
})
req := newToolRequest(SparkSubmitName, map[string]any{
"cluster_id": "cluster-persist",
"master": "yarn",
"deploy_mode": "cluster",
"script_path": absPath,
"queue": "default",
"executor_memory": "2G",
"executor_cores": 1,
"num_executors": 4,
})
res, err := deps.SparkSubmitHandler(context.Background(), req)
if err != nil {
t.Fatalf("handler error: %v", err)
}
if res.IsError {
t.Fatalf("unexpected error result: %v", res.Content)
}
payload := resultJSON(t, res)
if payload["app_id"] != "application_persist_0001" {
t.Errorf("app_id=%v, want application_persist_0001", payload["app_id"])
}
if payload["tracking_url"] != "http://rm.example.com:8088/proxy/application_persist_0001/" {
t.Errorf("tracking_url=%v, want tracking URL", payload["tracking_url"])
}
subs, err := subRepo.List(context.Background(), storage.SubmissionFilter{Limit: 10})
if err != nil {
t.Fatalf("list submissions: %v", err)
}
if len(subs) != 1 {
t.Fatalf("got %d submissions, want 1", len(subs))
}
s := subs[0]
if s.FileID != fileID {
t.Errorf("file_id=%q, want %q", s.FileID, fileID)
}
if s.AppID != "application_persist_0001" {
t.Errorf("app_id=%q, want application_persist_0001", s.AppID)
}
if s.TrackingURL != "http://rm.example.com:8088/proxy/application_persist_0001/" {
t.Errorf("tracking_url=%q, want tracking URL", s.TrackingURL)
}
if s.ClusterID != "cluster-persist" {
t.Errorf("cluster_id=%q, want cluster-persist", s.ClusterID)
}
if s.FileName != "hello.py" {
t.Errorf("file_name=%q, want hello.py", s.FileName)
}
}
func TestSparkSubmit_MissingRequiredField(t *testing.T) {
deps, _, _ := testDepsWithDataDir(t)
deps, _, _, _ := testDepsWithDataDir(t)
store := deps.UploadStore
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
if err != nil {
@@ -131,7 +234,7 @@ func TestSparkSubmit_MissingRequiredField(t *testing.T) {
}
func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
deps, _, _ := testDepsWithDataDir(t)
deps, _, _, _ := testDepsWithDataDir(t)
store := deps.UploadStore
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
if err != nil {
@@ -166,7 +269,7 @@ func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
}
func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) {
deps, _, _ := testDepsWithDataDir(t)
deps, _, _, _ := testDepsWithDataDir(t)
unminted := filepath.Join(t.TempDir(), "unminted.py")
if err := os.WriteFile(unminted, []byte("print('not from upload_file')\n"), 0o644); err != nil {
@@ -200,7 +303,7 @@ func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) {
}
func TestSparkSubmit_EmptyMaster(t *testing.T) {
deps, _, _ := testDepsWithDataDir(t)
deps, _, _, _ := testDepsWithDataDir(t)
store := deps.UploadStore
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
if err != nil {
+13
View File
@@ -47,6 +47,19 @@ CREATE TABLE IF NOT EXISTS upload_files (
uploaded_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_uploads_ts ON upload_files(uploaded_at DESC);
CREATE TABLE IF NOT EXISTS spark_submissions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
file_id TEXT NOT NULL,
app_id TEXT NOT NULL,
tracking_url TEXT,
cluster_id TEXT NOT NULL,
exit_code INTEGER NOT NULL,
duration_ms INTEGER NOT NULL,
submitted_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_submissions_app_id ON spark_submissions(app_id);
CREATE INDEX IF NOT EXISTS idx_submissions_submitted_at ON spark_submissions(submitted_at DESC);
CREATE INDEX IF NOT EXISTS idx_submissions_file_id ON spark_submissions(file_id);
`
// DB is the storage handle.
+194
View File
@@ -0,0 +1,194 @@
package storage
import (
"context"
"database/sql"
"errors"
"fmt"
"strings"
"time"
)
// Submission is one persisted spark_submit invocation.
type Submission struct {
ID int64 `json:"id"`
FileID string `json:"file_id"`
AppID string `json:"app_id"`
TrackingURL string `json:"tracking_url,omitempty"`
ClusterID string `json:"cluster_id"`
ExitCode int `json:"exit_code"`
DurationMS int64 `json:"duration_ms"`
SubmittedAt time.Time `json:"submitted_at"`
}
// SubmissionWithFile joins a submission with the original upload file name.
// FileName is empty when the referenced upload has been deleted.
type SubmissionWithFile struct {
Submission
FileName string `json:"file_name"`
}
// SubmissionFilter controls the rows returned by SubmissionRepo.List.
type SubmissionFilter struct {
Search string
FileID string
ClusterID string
AppID string
Limit int
Offset int
}
// SubmissionRepo provides CRUD operations for spark_submissions.
type SubmissionRepo struct {
db *DB
}
// Submissions returns a repository bound to this DB handle.
func (d *DB) Submissions() *SubmissionRepo {
return &SubmissionRepo{db: d}
}
// Create inserts a new submission record.
func (r *SubmissionRepo) Create(ctx context.Context, s Submission) error {
if s.SubmittedAt.IsZero() {
s.SubmittedAt = time.Now()
}
_, err := r.db.sqlDB.ExecContext(ctx, `
INSERT INTO spark_submissions (file_id, app_id, tracking_url, cluster_id, exit_code, duration_ms, submitted_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
s.FileID, s.AppID, s.TrackingURL, s.ClusterID, s.ExitCode, s.DurationMS, s.SubmittedAt.UnixNano(),
)
if err != nil {
return fmt.Errorf("storage: create submission: %w", err)
}
return nil
}
// Get retrieves a submission by id.
func (r *SubmissionRepo) Get(ctx context.Context, id int64) (Submission, error) {
row := r.db.sqlDB.QueryRowContext(ctx, `
SELECT id, file_id, app_id, tracking_url, cluster_id, exit_code, duration_ms, submitted_at
FROM spark_submissions
WHERE id = ?`, id)
s, err := scanSubmission(row.Scan)
if errors.Is(err, sql.ErrNoRows) {
return Submission{}, ErrNotFound
}
if err != nil {
return Submission{}, fmt.Errorf("storage: get submission %d: %w", id, err)
}
return s, nil
}
// List returns submissions ordered by submitted_at descending.
//
// Search matches app_id or the joined upload file name. Empty filter strings
// mean no restriction. Limit defaults to 100 and is capped at 500.
func (r *SubmissionRepo) List(ctx context.Context, filter SubmissionFilter) ([]SubmissionWithFile, error) {
if filter.Limit <= 0 {
filter.Limit = 100
}
if filter.Limit > 500 {
filter.Limit = 500
}
if filter.Offset < 0 {
filter.Offset = 0
}
args := []any{}
var where []string
if filter.Search != "" {
like := "%" + filter.Search + "%"
args = append(args, like, like)
where = append(where, "(s.app_id LIKE ? OR COALESCE(u.name, '') LIKE ?)")
}
if filter.FileID != "" {
args = append(args, filter.FileID)
where = append(where, "s.file_id = ?")
}
if filter.ClusterID != "" {
args = append(args, filter.ClusterID)
where = append(where, "s.cluster_id = ?")
}
if filter.AppID != "" {
args = append(args, filter.AppID)
where = append(where, "s.app_id = ?")
}
// WHERE args are in the order the conditions were appended; append LIMIT/OFFSET last.
args = append(args, filter.Limit, filter.Offset)
q := `SELECT s.id, s.file_id, s.app_id, s.tracking_url, s.cluster_id, s.exit_code, s.duration_ms, s.submitted_at, COALESCE(u.name, '') AS file_name
FROM spark_submissions s
LEFT JOIN upload_files u ON s.file_id = u.file_id`
if len(where) > 0 {
q += "\n\t\tWHERE " + strings.Join(where, " AND ")
}
q += "\n\t\tORDER BY s.submitted_at DESC\n\t\tLIMIT ? OFFSET ?"
rows, err := r.db.sqlDB.QueryContext(ctx, q, args...)
if err != nil {
return nil, fmt.Errorf("storage: list submissions: %w", err)
}
defer rows.Close()
var out []SubmissionWithFile
for rows.Next() {
swf, err := scanSubmissionWithFile(rows.Scan)
if err != nil {
return nil, fmt.Errorf("storage: list submissions: %w", err)
}
out = append(out, swf)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("storage: list submissions: %w", err)
}
if out == nil {
out = []SubmissionWithFile{}
}
return out, nil
}
// Delete removes a submission by id.
func (r *SubmissionRepo) Delete(ctx context.Context, id int64) error {
res, err := r.db.sqlDB.ExecContext(ctx, `DELETE FROM spark_submissions WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("storage: delete submission %d: %w", id, err)
}
n, err := res.RowsAffected()
if err != nil {
return fmt.Errorf("storage: delete submission %d: rows affected: %w", id, err)
}
if n == 0 {
return ErrNotFound
}
return nil
}
func scanSubmission(scan scanFunc) (Submission, error) {
var s Submission
var trackingURL sql.NullString
var submittedAt int64
if err := scan(&s.ID, &s.FileID, &s.AppID, &trackingURL, &s.ClusterID, &s.ExitCode, &s.DurationMS, &submittedAt); err != nil {
return Submission{}, err
}
if trackingURL.Valid {
s.TrackingURL = trackingURL.String
}
s.SubmittedAt = time.Unix(0, submittedAt)
return s, nil
}
func scanSubmissionWithFile(scan scanFunc) (SubmissionWithFile, error) {
var swf SubmissionWithFile
var trackingURL sql.NullString
var submittedAt int64
if err := scan(&swf.ID, &swf.FileID, &swf.AppID, &trackingURL, &swf.ClusterID, &swf.ExitCode, &swf.DurationMS, &submittedAt, &swf.FileName); err != nil {
return SubmissionWithFile{}, err
}
if trackingURL.Valid {
swf.TrackingURL = trackingURL.String
}
swf.SubmittedAt = time.Unix(0, submittedAt)
return swf, nil
}
+151
View File
@@ -0,0 +1,151 @@
package storage
import (
"context"
"testing"
"time"
)
func setupSubmissionTest(t *testing.T) (*SubmissionRepo, *UploadRepo, context.Context) {
db, err := Open(":memory:")
if err != nil {
t.Fatalf("open: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
return db.Submissions(), db.Uploads(), context.Background()
}
func TestSubmissionRepo_RoundTrip(t *testing.T) {
repo, uploadRepo, ctx := setupSubmissionTest(t)
if err := uploadRepo.Create(ctx, "00000000000000000000000000000001", "alpha.py", 10, "deadbeef", time.Now()); err != nil {
t.Fatalf("create upload: %v", err)
}
s := Submission{
FileID: "00000000000000000000000000000001",
AppID: "application_1_0001",
TrackingURL: "http://rm.example.com:8088/proxy/application_1_0001/",
ClusterID: "cluster-a",
ExitCode: 0,
DurationMS: 1234,
SubmittedAt: time.Unix(0, time.Now().UnixNano()),
}
if err := repo.Create(ctx, s); err != nil {
t.Fatalf("create: %v", err)
}
list, err := repo.List(ctx, SubmissionFilter{Limit: 10})
if err != nil {
t.Fatalf("list: %v", err)
}
if len(list) != 1 {
t.Fatalf("len(list)=%d, want 1", len(list))
}
swf := list[0]
if swf.FileName != "alpha.py" {
t.Errorf("file_name=%q, want alpha.py", swf.FileName)
}
if swf.AppID != s.AppID {
t.Errorf("app_id=%q, want %q", swf.AppID, s.AppID)
}
id := swf.ID
got, err := repo.Get(ctx, id)
if err != nil {
t.Fatalf("get: %v", err)
}
if got.ID != id || got.AppID != s.AppID {
t.Errorf("got=%+v, want id=%d app_id=%s", got, id, s.AppID)
}
if err := repo.Delete(ctx, id); err != nil {
t.Fatalf("delete: %v", err)
}
if _, err := repo.Get(ctx, id); !isErrNotFound(err) {
t.Errorf("expected ErrNotFound after delete, got %v", err)
}
}
func TestSubmissionRepo_SearchByAppID(t *testing.T) {
repo, uploadRepo, ctx := setupSubmissionTest(t)
if err := uploadRepo.Create(ctx, "00000000000000000000000000000001", "alpha.py", 10, "deadbeef", time.Now()); err != nil {
t.Fatalf("create upload alpha: %v", err)
}
if err := uploadRepo.Create(ctx, "00000000000000000000000000000002", "beta.py", 10, "cafebabe", time.Now()); err != nil {
t.Fatalf("create upload beta: %v", err)
}
if err := repo.Create(ctx, Submission{FileID: "00000000000000000000000000000001", AppID: "application_alpha_1", ClusterID: "c", ExitCode: 0, DurationMS: 1, SubmittedAt: time.Now()}); err != nil {
t.Fatalf("create alpha: %v", err)
}
if err := repo.Create(ctx, Submission{FileID: "00000000000000000000000000000002", AppID: "application_beta_1", ClusterID: "c", ExitCode: 0, DurationMS: 1, SubmittedAt: time.Now()}); err != nil {
t.Fatalf("create beta: %v", err)
}
list, err := repo.List(ctx, SubmissionFilter{Search: "alpha", Limit: 10})
if err != nil {
t.Fatalf("search: %v", err)
}
if len(list) != 1 || list[0].AppID != "application_alpha_1" || list[0].FileName != "alpha.py" {
t.Errorf("got %+v, want 1 alpha submission", list)
}
}
func TestSubmissionRepo_SearchByFileName(t *testing.T) {
db, err := Open(":memory:")
if err != nil {
t.Fatalf("open: %v", err)
}
defer db.Close()
repo := db.Submissions()
uploadRepo := db.Uploads()
ctx := context.Background()
if err := uploadRepo.Create(ctx, "00000000000000000000000000000001", "report.py", 10, "deadbeef", time.Now()); err != nil {
t.Fatalf("create upload: %v", err)
}
if err := repo.Create(ctx, Submission{FileID: "00000000000000000000000000000001", AppID: "application_x_1", ClusterID: "c", ExitCode: 0, DurationMS: 1, SubmittedAt: time.Now()}); err != nil {
t.Fatalf("create: %v", err)
}
list, err := repo.List(ctx, SubmissionFilter{Search: "report", Limit: 10})
if err != nil {
t.Fatalf("search: %v", err)
}
if len(list) != 1 || list[0].FileName != "report.py" {
t.Errorf("got %+v, want 1 report submission", list)
}
}
func TestSubmissionRepo_FilterByFileID(t *testing.T) {
db, err := Open(":memory:")
if err != nil {
t.Fatalf("open: %v", err)
}
defer db.Close()
repo := db.Submissions()
ctx := context.Background()
if err := repo.Create(ctx, Submission{FileID: "f1", AppID: "a1", ClusterID: "c", ExitCode: 0, DurationMS: 1, SubmittedAt: time.Now()}); err != nil {
t.Fatalf("create f1: %v", err)
}
if err := repo.Create(ctx, Submission{FileID: "f2", AppID: "a2", ClusterID: "c", ExitCode: 0, DurationMS: 1, SubmittedAt: time.Now()}); err != nil {
t.Fatalf("create f2: %v", err)
}
list, err := repo.List(ctx, SubmissionFilter{FileID: "f1", Limit: 10})
if err != nil {
t.Fatalf("list: %v", err)
}
if len(list) != 1 || list[0].FileID != "f1" {
t.Errorf("got %+v, want f1 submission", list)
}
}
func isErrNotFound(err error) bool {
if err == nil {
return false
}
return err.Error() == ErrNotFound.Error()
}
+11 -9
View File
@@ -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 {
+2 -1
View File
@@ -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,