23 Commits
Author SHA1 Message Date
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
tao.chenandClaude df9b01272f admin: uploads list view with search, sort, and delete
Add /admin/uploads HTML page and JSON API backed by the upload_files

DB index. The page lists file_id, name, size (KB), uploaded_at, and

sha256 with client-side search by name, column sort, and per-row delete.

DELETE /admin/uploads/:id removes the DB index row and the on-disk

data file plus .meta.json sidecar. Admin deletes are written to the

audit log.

Wire the new dependencies through admin.Mount and main.go, and run

the backfill step on startup so pre-existing sidecars are indexed.

Tests cover auth, HTML rendering, JSON search, delete removing all

artifacts, and audit entry creation.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 11:39:56 +08:00
tao.chenandClaude 876f1c9edb uploads: dual-write uploads to DB, startup backfill, and audit log
Store now indexes every Save into the upload_files table via SetRepo.

DB failures are logged as warnings and do not fail the upload because

.meta.json remains the source of truth and startup backfill recovers.

Add Store.Backfill to walk Root at startup and insert index rows for

any pre-existing .meta.json sidecars, swallowing duplicate-key races.

The upload_file MCP Tool now writes an audit_log entry on success.

Tests cover dual-write args, repo-error non-failure, and backfill

skipping existing rows.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 11:30:39 +08:00
tao.chenandClaude 1796b5b872 storage: add upload_files table and UploadRepo
Add the upload_files SQLite table and UploadRepo CRUD methods

(Create/Get/List/Delete). The table stores file_id, name, size,

sha256, and uploaded_at as a queryable index.

The .meta.json sidecar remains the source of truth for uploads;

the DB is only an index for admin visibility. On startup the next

commit will backfill any pre-existing sidecars into this table.

Tests cover Create/Get/List/Delete round-trip, name search

filtering, descending time ordering, and limit defaults/cap.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 11:21:38 +08:00
tao.chenandClaude 4af166ea48 tools: spark_submit 记录拼装命令到 per-tool log
之前 d.Logger.Debug("spark_submit.built", "argv", cmd) 在 spark-submit
执行前 log 一次, 但 LOG_LEVEL=info 默认看不到, 拼装结果只活在
运行时 stack. 现在把 binary + argv 合并进 callLog.WithResult 写入
per-tool 文件 (data/logs/tools/<ts>_spark_submit_<id>.log), 每次
执行都在 tail -f 的现场能 grep 出来.

   tail -f data/logs/tools/*.log | grep -A1 '"argv"'
   -> 看每次实际传给 spark-submit 的 argv, 含 script_path 在最后

保留 Debug 调用: verbose 模式下用 base.Logger 直接看到完整 argv 跟
cluster_id, 不依赖 file tail.

两份记录 (Debug + WithResult) 字段一致, 都是 binary + argv, 排查
spark-submit 行为时可以交叉对照.

测试:
  - 已有 TestSparkSubmit_StructuredCommand 跑通, argv 字段被
    JSON marshal 进 resultRaw, 不需要新增测试
  - go test ./... 全绿

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 10:51:03 +08:00
tao.chenandClaude 157848d87a main: bound startup sweep with context timeout
Sweep() used to run synchronously at startup without any time bound.
On a large uploads directory this can block the server for 30s or more,
keeping /healthz returning 503 while the reaper works. Add a context
timeout so startup continues if the sweep cannot finish within budget.

The timeout is set to 5 seconds: long enough to clear typical leftover
state, short enough that a stuck sweep will not meaningfully delay
health checks or startup readiness.

Change Store.Sweep to accept a context.Context and check ctx.Err() at
the top of each iteration. If the context is cancelled or times out,
Sweep returns the files deleted so far and ctx.Err().

Add TestStore_Sweep_HonorsContextCancel to ensure a cancelled context
causes Sweep to exit early instead of running to completion.

Deliberately unchanged: there is still no periodic reaper. The server
only sweeps once at startup; this change just bounds that single sweep.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 10:18:13 +08:00
tao.chenandClaude 97f6184f65 cluster: deprecate DefaultSubmitArgs (no longer prepended)
cluster.DefaultSubmitArgs is now deprecated. It used to be prepended to
spark_submit Tool calls, but since 7920dc9 the spark-submit builder
constructs argv from the structured fields in the tool call. The field is
still accepted by the admin API and still persisted to the DB, so this is
a non-breaking change.

This commit deliberately does NOT remove:
- the Cluster.DefaultSubmitArgs field in internal/cluster/types.go
- the default_submit_args DB column
- the admin API read/write support in internal/admin/router.go
- existing storage/repo tests

Runtime now logs a one-time slog.Info when the PUT handler receives a
non-empty default_submit_args value, warning operators that the field is
ignored by spark_submit and that they should use spark_conf / extra_args
instead.

Follow-up removal should touch:
- the Cluster struct field and its JSON tag
- storage/cluster_repo.go scan/insert/update SQL and params
- admin/router.go JSON round-tripping
- storage/cluster_repo_test.go assertions
- the DB migration dropping the default_submit_args column

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 10:11:39 +08:00
tao.chen dab4cd062e 修复 spark_submit 参数重复、上传校验绕过等 8 处缺陷
对应代码审查发现的问题(#1, #2, #4-#8, #10):

#1 CRITICAL:spark_submit 在 argv 中重复拼接 binary。此前
    cmd := []string{binary} 后又把 cmd 作为 Args 传给 executor.Run,
    而 executor 会再拼一次 Binary,导致 OS argv 为 [binary, binary, ...],
    spark-submit 会把自身当作应用 jar。现在 cmd 从 --master 开始,
    executor.Run 使用 Binary + Args,argv 正确。
#2:Validate 曾接受 .meta.json 路径本身。现在显式拒绝 sidecar 路径,
    要求传入数据文件路径。
#4:Sweep 对 sidecar 损坏的数据文件跳过清理。现在损坏 sidecar 会回退
    到数据文件 mtime,超期即删除。
#5:upload_file 描述仍引用已移除的 args 字段,已改为引用 script_path
    及结构化字段。
#6:Deps.UploadStore 改为值类型 uploads.Store,避免 nil 绕过上传校验;
    移除 spark_submit/upload_file 中的 nil 检查。
#7:master/queue/executor_memory 增加空字符串校验。
#8:提取 buildSparkSubmitCommand 构建 argv,消除双写参数的结构性根因。
#10:Validate 失败时记录 slog.Warn("spark_submit.unminted_path_rejected")。

新增测试:
- TestSparkSubmit_StructuredCommand:断言 argv 首行为 --master,末行
  仍为 script_path。
- TestSparkSubmit_EmptyMaster:空 master 返回错误。
- TestStore_Validate_RejectsSidecarPath:拒绝 .meta.json 路径。
- TestStore_Sweep_DeletesDataWithCorruptSidecar:损坏 sidecar 的数据文件
  被清理。

未在本提交处理:
- #9 cluster.DefaultSubmitArgs 弃用留作后续批次。

Co-Authored-By: tao.chen <93983997+taochen-ct@users.noreply.github.com>
2026-07-14 10:09:48 +08:00
tao.chenandClaude 4a28886502 tools: 清理 fetch_url.go S1017 lint 警告
matchAllowedHost 里两处 if strings.HasPrefix(...) pattern = pattern[N:]
替成无条件 strings.TrimPrefix, 行为完全等价 (前缀不匹配时
TrimPrefix 是 no-op). 触发 S1017 (Should replace this if statement
with an unconditional strings.TrimPrefix).

跟 475e353 在 httpclient/ssrf.go 上做的清理一模一样, 当时
只清了一处, fetch_url.go 里 copy 了同款 pattern 漏了.

净 -4 行, 函数语义不变, 现有测试 (fetch_url_test.go 覆盖的精确
/ 后缀 / wildcard 三种 host 匹配路径) 不变.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-13 19:56:24 +08:00
tao.chenandClaude 67d1e9a6c5 gofmt: 清理 admin/executor 测试文件的对齐空格
这两个文件在 HEAD 上就未通过 gofmt, 但 go test 不检查 gofmt,
所以一直没暴露. 修一下让 gofmt -l . 干净, 否则新加的 commit
之前都得先 stash.

纯格式变更, 无逻辑变化.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-13 19:55:26 +08:00
tao.chenandClaude 44c9f415b7 tools: 修 spark_submit 校验的两处小问题
c9e54a3 引入了 server-minted 校验, 但有两处需要调整:

1. 错误信息用 fmt.Errorf(... %w).Error() 构造, %w 包装语义被丢
   掉 (errResult 接收 string, Error() 之后 %w 已经不可达). 改用
   fmt.Sprintf 拼接, %q 引用路径, 错误信息保持不变但代码不再误
   导.

2. 校验放在 scriptPath 解析之后、queue 解析之前. 错误信息会按字
   段出现顺序报 (cluster_id, master, deploy_mode, script_path,
   queue, ...), 但当前顺序是 script_path 校验先报, 然后才报 queue
   缺失. 把校验挪到所有 RequireString 之后、parseStringMap 之前,
   LLM 看错误时字段顺序跟 schema 顺序一致.

nil-safe 校验逻辑不变 (d.UploadStore == nil 时跳过). 现有测试
全部通过:
  - TestSparkSubmit_RejectsNonMintedPath 仍通过 (校验位置不影响
    行为)
  - TestSparkSubmit_StructuredCommand 仍通过 (走 mint 路径)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-13 19:55:16 +08:00
tao.chenandClaude c9e54a3f27 tools: 强制 spark_submit 的 script_path 必须来自 upload_file
在 spark_submit handler 中增加 server-minted 校验:script_path 必须通过
本服务器 upload_file 的 UploadStore.Validate,否则立即返回清晰错误,避免
远程 agent 把自身本地路径转发给服务器侧 spark-submit 时出现难以定位的
FileNotFoundException。

校验对 nil UploadStore 是安全的:当 d.UploadStore 为 nil 时直接跳过,保留
不强制依赖 store 的测试或降级场景。当前 testDepsWithDataDir 始终会注入
UploadStore,所以测试走的是真实校验路径。

同步更新了 script_path 的工具描述,明确声明非 upload_file 铸造的路径会被
拒绝。这是对 7920dc9 中“path 永远在最后”规则的收紧——现在不仅位置固定,
而且必须是本机 uploads 目录下的有效上传文件。

测试调整:
- 新增 TestSparkSubmit_RejectsNonMintedPath:使用 t.TempDir() 下未通过
  upload_file 写入的文件作为 script_path,断言返回错误并包含
  "not from upload_file"。
- TestSparkSubmit_MissingRequiredField / TestSparkSubmit_BadSparkConfValue
  原来使用 "/tmp/script.py" 作为占位路径,现在会先通过 UploadStore.Save
  生成一个铸造路径,确保它们继续分别验证 master 缺失和 spark_conf 值类型
  错误,而不是被新的路径守卫拦截。

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-13 19:49:26 +08:00
28 changed files with 3061 additions and 153 deletions
+127
View File
@@ -0,0 +1,127 @@
# 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
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: success()
run: |
TITLE="${{ github.ref_name }}"
BODY="spark-mcp ${TITLE} released: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/releases/tag/${TITLE}"
PAYLOAD=$(cat <<EOF
{
"device_key": "6ZFVsxM95cuAjYoNLWSWaN",
"title": "${TITLE}",
"body": "${BODY}",
"group": "release",
"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}"
+151 -1
View File
@@ -5,8 +5,12 @@ package admin
import (
"encoding/json"
"errors"
"log/slog"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"github.com/gin-gonic/gin"
@@ -14,11 +18,12 @@ import (
"spark-mcp-go/internal/cluster"
"spark-mcp-go/internal/middleware"
"spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
)
// Mount attaches the /admin sub-router to r, protecting every route with
// bearer-token admin authentication.
func Mount(r *gin.Engine, repo *storage.ClusterRepo, auditRepo *audit.Repo, 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")
@@ -32,6 +37,12 @@ func Mount(r *gin.Engine, repo *storage.ClusterRepo, auditRepo *audit.Repo, admi
g.PUT("/clusters/:id", updateCluster(repo, auditRepo))
g.DELETE("/clusters/:id", deleteCluster(repo, auditRepo))
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)
}
@@ -170,6 +181,10 @@ func updateCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.Handler
merged.DefaultSubmitArgs = decodeStringSlice(v)
}
if len(merged.DefaultSubmitArgs) > 0 {
slog.Default().Info("admin: cluster sets default_submit_args which is deprecated; the spark_submit Tool no longer prepends these args. Configure flags via spark_conf / extra_args in the structured tool call.", "cluster_id", merged.ID)
}
// Re-validate after merging so a missing auth_type on a fresh
// cluster (defaulted by DB to "none") doesn't get clobbered.
if err := validateClusterUpdate(&merged); err != nil {
@@ -270,6 +285,141 @@ func deleteCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.Handler
}
}
func listUploads(repo *storage.UploadRepo) gin.HandlerFunc {
return func(c *gin.Context) {
search := c.Query("search")
limitStr := c.DefaultQuery("limit", "100")
limit, err := strconv.Atoi(limitStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid limit"})
return
}
uploads, err := repo.List(c.Request.Context(), search, limit)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, uploads)
}
}
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")
ctx := c.Request.Context()
meta, err := repo.Get(ctx, id)
if err != nil && !errors.Is(err, storage.ErrNotFound) {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
if err := repo.Delete(ctx, id); err != nil && !errors.Is(err, storage.ErrNotFound) {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
// Always attempt to remove the on-disk files; the DB is only an index.
if store != nil {
dataPath := filepath.Join(store.Root, id)
_ = os.Remove(dataPath)
_ = os.Remove(dataPath + ".meta.json")
}
if meta.FileID != "" {
details, _ := audit.MarshalDetails(map[string]any{
"file_id": id,
"name": meta.Name,
"size": meta.Size,
"sha256": meta.SHA256,
})
_ = auditRepo.Insert(ctx, &audit.Entry{
Actor: actor(c),
Action: audit.ActionUploadDelete,
ClusterID: id,
Details: details,
})
}
c.Status(http.StatusNoContent)
}
}
func listAudit(auditRepo *audit.Repo) gin.HandlerFunc {
return func(c *gin.Context) {
limitStr := c.DefaultQuery("limit", "100")
+21 -15
View File
@@ -7,6 +7,7 @@ import (
"io"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
@@ -15,6 +16,7 @@ import (
"spark-mcp-go/internal/audit"
"spark-mcp-go/internal/cluster"
"spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
)
func init() {
@@ -29,7 +31,11 @@ func newTestAdmin(t *testing.T) (*gin.Engine, *storage.ClusterRepo, *audit.Repo)
}
t.Cleanup(func() { _ = db.Close() })
r := gin.New()
Mount(r, db.Clusters(), audit.NewRepo(db), []string{"good-token"})
uploadStore, err := uploads.New(filepath.Join(t.TempDir(), "uploads"))
if err != nil {
t.Fatalf("new upload store: %v", err)
}
Mount(r, db.Clusters(), db.Uploads(), db.Submissions(), audit.NewRepo(db), &uploadStore, []string{"good-token"})
return r, db.Clusters(), audit.NewRepo(db)
}
@@ -54,20 +60,20 @@ func doReq(t *testing.T, r *gin.Engine, method, path, token string, body any) *h
func fullClusterMap(id string) map[string]any {
return map[string]any{
"id": id,
"name": id + "-name",
"rm_url": "http://rm.example.com:8088",
"shs_url": "http://shs.example.com:18080",
"spark_submit_execute_bin": "/usr/bin/spark-submit",
"is_active": true,
"auth_type": "basic",
"auth_username": "admin",
"auth_password": "real-secret",
"ssl_verify": false,
"ssl_ca_bundle": "",
"url_allowlist": []string{"*.example.com"},
"default_submit_args": []string{"--master", "yarn"},
"rate_limit_per_min": 10,
"id": id,
"name": id + "-name",
"rm_url": "http://rm.example.com:8088",
"shs_url": "http://shs.example.com:18080",
"spark_submit_execute_bin": "/usr/bin/spark-submit",
"is_active": true,
"auth_type": "basic",
"auth_username": "admin",
"auth_password": "real-secret",
"ssl_verify": false,
"ssl_ca_bundle": "",
"url_allowlist": []string{"*.example.com"},
"default_submit_args": []string{"--master", "yarn"},
"rate_limit_per_min": 10,
}
}
+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)
}
}
+196
View File
@@ -0,0 +1,196 @@
package admin
import (
"context"
"encoding/json"
"errors"
"net/http"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/gin-gonic/gin"
"spark-mcp-go/internal/audit"
"spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
)
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 {
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(), 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)
w := doReq(t, r, "GET", "/admin/uploads", "", nil)
if w.Code != http.StatusUnauthorized {
t.Errorf("GET /admin/uploads without auth: got %d, want %d", w.Code, http.StatusUnauthorized)
}
w = doReq(t, r, "GET", "/admin/uploads", "good-token", nil)
if w.Code != http.StatusOK {
t.Errorf("GET /admin/uploads 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, "uploads-body") {
t.Errorf("body missing uploads table")
}
}
func TestListUploads(t *testing.T) {
r, repo, _, _, _ := newTestAdminUploads(t)
ctx := context.Background()
uploadedAt := time.Unix(0, time.Now().UnixNano())
if err := repo.Create(ctx, "00000000000000000000000000000001", "alpha.txt", 10, "deadbeef", uploadedAt); err != nil {
t.Fatalf("create upload: %v", err)
}
if err := repo.Create(ctx, "00000000000000000000000000000002", "beta.txt", 20, "cafebabe", uploadedAt.Add(time.Second)); err != nil {
t.Fatalf("create upload: %v", err)
}
w := doReq(t, r, "GET", "/admin/uploads/api", "good-token", nil)
if w.Code != http.StatusOK {
t.Fatalf("got status %d, want %d", w.Code, http.StatusOK)
}
var list []map[string]any
if err := json.Unmarshal(w.Body.Bytes(), &list); err != nil {
t.Fatalf("unmarshal list: %v", err)
}
if len(list) != 2 {
t.Errorf("got %d uploads, want 2", len(list))
}
w = doReq(t, r, "GET", "/admin/uploads/api?search=alpha", "good-token", nil)
if w.Code != http.StatusOK {
t.Fatalf("search: got status %d", w.Code)
}
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))
}
}
func TestDeleteUpload(t *testing.T) {
r, repo, _, auditRepo, store := newTestAdminUploads(t)
ctx := context.Background()
id := "00000000000000000000000000000003"
dataPath := filepath.Join(store.Root, id)
metaPath := dataPath + ".meta.json"
if err := os.WriteFile(dataPath, []byte("payload"), 0o640); err != nil {
t.Fatalf("create data file: %v", err)
}
sc := map[string]any{
"name": "delete-me.txt",
"size": 7,
"sha256": "feedface",
"uploaded_at": time.Now().Format(time.RFC3339Nano),
}
b, _ := json.Marshal(sc)
if err := os.WriteFile(metaPath, b, 0o600); err != nil {
t.Fatalf("create sidecar: %v", err)
}
if err := repo.Create(ctx, id, "delete-me.txt", 7, "feedface", time.Now()); err != nil {
t.Fatalf("create db row: %v", err)
}
w := doReq(t, r, "DELETE", "/admin/uploads/"+id, "good-token", nil)
if w.Code != http.StatusNoContent {
t.Fatalf("delete: got status %d, want %d: %s", w.Code, http.StatusNoContent, w.Body.String())
}
if _, err := repo.Get(ctx, id); !errors.Is(err, storage.ErrNotFound) {
t.Errorf("db row still exists: %v", err)
}
if _, err := os.Stat(dataPath); !os.IsNotExist(err) {
t.Errorf("data file still exists")
}
if _, err := os.Stat(metaPath); !os.IsNotExist(err) {
t.Errorf("sidecar still exists")
}
entries, err := auditRepo.List(ctx, 100)
if err != nil {
t.Fatalf("list audit: %v", err)
}
found := false
for _, e := range entries {
if e.Action == audit.ActionUploadDelete && e.ClusterID == id {
found = true
break
}
}
if !found {
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"])
}
}
+19 -1
View File
@@ -7,7 +7,7 @@ import (
"github.com/gin-gonic/gin"
)
//go:embed web/cluster.html
//go:embed web
var webFS embed.FS
func webHandler(c *gin.Context) {
@@ -18,3 +18,21 @@ func webHandler(c *gin.Context) {
}
c.Data(http.StatusOK, "text/html; charset=utf-8", data)
}
func uploadsWebHandler(c *gin.Context) {
data, err := webFS.ReadFile("web/uploads.html")
if err != nil {
c.String(http.StatusInternalServerError, "uploads.html: %v", err)
return
}
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)
}
+12 -1
View File
@@ -34,6 +34,10 @@
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'], input[type='number'], select, textarea {
background: var(--bg);
@@ -111,7 +115,14 @@
</head>
<body>
<header>
<h1>spark-mcp-go Admin</h1>
<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>
<a href='/admin/submissions'>Submissions</a>
</nav>
</div>
<div class='auth'>
<button id='set-token' class='secondary'>Set Token</button>
<button id='logout' class='danger'>Logout</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>
+378
View File
@@ -0,0 +1,378 @@
<!DOCTYPE html>
<html lang='en'>
<head>
<meta charset='utf-8'>
<meta name='viewport' content='width=device-width, initial-scale=1'>
<title>Uploads - 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: 220px; }
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: 1200px; 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; }
.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' class='active'>Uploads</a>
<a href='/admin/submissions'>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 name...'>
<button id='refresh' class='secondary'>Refresh</button>
<span id='count'></span>
</section>
<section>
<table>
<thead>
<tr>
<th class='sortable' data-key='file_id'>File ID</th>
<th class='sortable' data-key='name'>Name</th>
<th class='sortable' data-key='size'>Size (KB)</th>
<th class='sortable' data-key='uploaded_at'>Uploaded At</th>
<th>SHA256</th>
<th>Actions</th>
</tr>
</thead>
<tbody id='uploads-body'>
<tr><td colspan='6'>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/uploads';
let uploads = [];
let sortKey = 'uploaded_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 {
loadUploads();
}
}
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 formatKB(bytes) {
return (bytes / 1024).toFixed(2);
}
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;');
}
function filteredUploads() {
const q = $('#search').value.trim().toLowerCase();
if (!q) return uploads.slice();
return uploads.filter(u => (u.name || '').toLowerCase().includes(q));
}
function compare(a, b) {
let av = a[sortKey];
let bv = b[sortKey];
if (sortKey === 'size') { av = Number(av); bv = Number(bv); }
else if (sortKey === 'uploaded_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 = $('#uploads-body');
const list = filteredUploads().sort(compare);
$('#count').textContent = list.length + ' file' + (list.length === 1 ? '' : 's');
tbody.innerHTML = '';
if (!list.length) {
tbody.innerHTML = '<tr><td colspan="6">No uploads.</td></tr>';
return;
}
for (const u of list) {
const tr = document.createElement('tr');
tr.innerHTML = `
<td class='mono' title='${escapeHtml(u.file_id)}'>${escapeHtml(truncate(u.file_id, 8))}</td>
<td>${escapeHtml(u.name)}</td>
<td>${escapeHtml(formatKB(u.size))}</td>
<td>${escapeHtml(formatTime(u.uploaded_at))}</td>
<td class='mono' title='${escapeHtml(u.sha256)}'>${escapeHtml(truncate(u.sha256, 8))}</td>
<td class='actions'>
<button data-id='${escapeHtml(u.file_id)}' class='danger delete'>Delete</button>
</td>
`;
tbody.appendChild(tr);
}
tbody.querySelectorAll('.delete').forEach(b => b.addEventListener('click', handleDelete));
}
async function loadUploads() {
showError('');
try {
uploads = await api('GET', '/api');
renderList();
} catch (e) {
showError('Failed to load uploads: ' + e.message);
}
}
async function handleDelete(e) {
const id = e.target.dataset.id;
if (!confirm(`Delete upload ${id}?`)) return;
showError('');
try {
await api('DELETE', `/${encodeURIComponent(id)}`);
await loadUploads();
} catch (err) {
showError('Failed to delete upload: ' + err.message);
}
}
function handleSort(e) {
const key = e.target.dataset.key;
if (!key) return;
if (sortKey === key) {
sortDir = -sortDir;
} else {
sortKey = key;
sortDir = key === 'uploaded_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', renderList);
$('#refresh').addEventListener('click', loadUploads);
$('#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 {
loadUploads();
}
</script>
</body>
</html>
+2
View File
@@ -18,6 +18,8 @@ const (
ActionClusterCreate Action = "cluster.create"
ActionClusterUpdate Action = "cluster.update"
ActionClusterDelete Action = "cluster.delete"
ActionUploadCreate Action = "upload.create"
ActionUploadDelete Action = "upload.delete"
)
// Entry is one admin write operation recorded for accountability.
+22 -17
View File
@@ -34,23 +34,28 @@ const (
// password" — keeps the existing encrypted blob intact.
// - URLAllowlist: extra glob patterns beyond the RM/SHS hosts; fetch_url
// uses this to permit long-tail SHS endpoints the agent may construct.
// - DefaultSubmitArgs: prepended to every spark_submit Tool call's args.
// - DefaultSubmitArgs: deprecated; no longer prepended by spark_submit
// (post-7920dc9 the builder constructs argv from structured fields).
// Kept for backward compatibility with the admin API and DB schema;
// actual removal is a follow-up. New cluster configurations should
// leave this empty.
// - RateLimitPerMin: 0 disables the per-cluster limiter.
type Cluster struct {
ID string `json:"id"`
Name string `json:"name"`
RMURL string `json:"rm_url"`
SHSURL string `json:"shs_url"`
SparkSubmitExecuteBin string `json:"spark_submit_execute_bin"`
IsActive bool `json:"is_active"`
AuthType AuthType `json:"auth_type"`
AuthUsername string `json:"auth_username,omitempty"`
AuthPassword string `json:"-"` // never serialized; encrypted at rest in auth_password_enc
SSLVerify bool `json:"ssl_verify"`
SSLCABundle string `json:"ssl_ca_bundle,omitempty"`
URLAllowlist []string `json:"url_allowlist,omitempty"`
DefaultSubmitArgs []string `json:"default_submit_args,omitempty"`
RateLimitPerMin int `json:"rate_limit_per_min"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
ID string `json:"id"`
Name string `json:"name"`
RMURL string `json:"rm_url"`
SHSURL string `json:"shs_url"`
SparkSubmitExecuteBin string `json:"spark_submit_execute_bin"`
IsActive bool `json:"is_active"`
AuthType AuthType `json:"auth_type"`
AuthUsername string `json:"auth_username,omitempty"`
AuthPassword string `json:"-"` // never serialized; encrypted at rest in auth_password_enc
SSLVerify bool `json:"ssl_verify"`
SSLCABundle string `json:"ssl_ca_bundle,omitempty"`
URLAllowlist []string `json:"url_allowlist,omitempty"`
// Deprecated: see godoc.
DefaultSubmitArgs []string `json:"default_submit_args,omitempty"`
RateLimitPerMin int `json:"rate_limit_per_min"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
+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)
}
}
+16 -12
View File
@@ -29,8 +29,8 @@ func TestRun_FakeBinary(t *testing.T) {
defer cancel()
res, err := Run(ctx, SparkSubmitOpts{
Binary: "/bin/echo",
Args: []string{"--master", "yarn; rm -rf /tmp/this-should-not-exist"},
Binary: "/bin/echo",
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)
}
}
@@ -54,8 +54,8 @@ func TestRun_NonZeroExit(t *testing.T) {
t.Skip("uses unix binary")
}
res, err := Run(context.Background(), SparkSubmitOpts{
Binary: "/bin/sh",
Args: []string{"-c", "exit 7"},
Binary: "/bin/sh",
Args: []string{"-c", "exit 7"},
Timeout: 3 * time.Second,
})
if err == nil {
@@ -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)
}
}
}
+4 -1
View File
@@ -5,6 +5,7 @@ import (
"time"
"spark-mcp-go/internal/analyzer"
"spark-mcp-go/internal/audit"
"spark-mcp-go/internal/httpclient"
"spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
@@ -13,12 +14,14 @@ import (
// Deps bundles the dependencies shared by all MCP Tool handlers.
type Deps struct {
Logger *slog.Logger
AuditRepo *audit.Repo
ClusterRepo *storage.ClusterRepo
SubmissionRepo *storage.SubmissionRepo
SparkSubmitTimeout time.Duration
HTTPClient *httpclient.Client
MaxResponseBytes int64
DataDir string // upload_file writes to DataDir/uploads
UploadStore *uploads.Store
UploadStore uploads.Store
AnalyzerThresholds analyzer.Thresholds
}
+25 -7
View File
@@ -16,6 +16,7 @@ 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"
@@ -24,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) {
func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo, *storage.SubmissionRepo, *audit.Repo) {
t.Helper()
db, err := storage.Open(":memory:")
if err != nil {
@@ -43,11 +44,13 @@ func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo) {
Timeout: 5 * time.Second,
MaxResponseBytes: 1 << 20,
}),
AuditRepo: audit.NewRepo(db),
ClusterRepo: db.Clusters(),
SubmissionRepo: db.Submissions(),
MaxResponseBytes: 1 << 20,
DataDir: dataDir,
UploadStore: uploadStore,
}, db.Clusters()
}, db.Clusters(), db.Submissions(), audit.NewRepo(db)
}
// createCluster creates a cluster in the repository with the given fields.
@@ -98,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",
@@ -292,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",
@@ -321,7 +324,7 @@ func TestFetchURL_RedirectPreservesAuth(t *testing.T) {
}
func TestUploadFile(t *testing.T) {
deps, _ := testDepsWithDataDir(t)
deps, _, _, auditRepo := testDepsWithDataDir(t)
req := newToolRequest(UploadFileName, map[string]any{
"filename": "hello.txt",
@@ -367,10 +370,25 @@ func TestUploadFile(t *testing.T) {
if info.Mode().Perm() != 0o640 {
t.Errorf("mode=%o, want %o", info.Mode().Perm(), 0o640)
}
entries, err := auditRepo.List(context.Background(), 100)
if err != nil {
t.Fatalf("list audit: %v", err)
}
found := false
for _, e := range entries {
if e.Action == audit.ActionUploadCreate && e.Actor == "tool:upload_file" && e.ClusterID == fileID {
found = true
break
}
}
if !found {
t.Errorf("missing upload.create audit entry for file_id %s", fileID)
}
}
func TestUploadFile_PathTraversal(t *testing.T) {
deps, _ := testDepsWithDataDir(t)
deps, _, _, _ := testDepsWithDataDir(t)
cases := []string{
"../../../etc/passwd",
@@ -401,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",
+2 -6
View File
@@ -209,12 +209,8 @@ func hostAllowed(host string, allowed []string) bool {
if pattern == host {
return true
}
if strings.HasPrefix(pattern, "*.") {
pattern = pattern[2:]
}
if strings.HasPrefix(pattern, ".") {
pattern = pattern[1:]
}
pattern = strings.TrimPrefix(pattern, "*.")
pattern = strings.TrimPrefix(pattern, ".")
if host == pattern || strings.HasSuffix(host, "."+pattern) {
return true
}
+123 -39
View File
@@ -3,12 +3,15 @@ package tools
import (
"context"
"fmt"
"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"
@@ -32,7 +35,7 @@ func NewSparkSubmitTool() mcp.Tool {
),
mcp.WithString("script_path",
mcp.Required(),
mcp.Description("Absolute path to the script. Pass the path field returned by upload_file — do not construct your own path. The script is always the last argv element."),
mcp.Description("Absolute path to the script. Pass the `path` field returned by upload_file — do not construct your own path. The script is always the last argv element. Paths not minted by upload_file on this server are rejected."),
),
mcp.WithString("queue",
mcp.Required(),
@@ -63,11 +66,13 @@ func NewSparkSubmitTool() mcp.Tool {
// SparkSubmitHandler runs spark-submit against the requested cluster.
func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
args := req.GetArguments()
clusterID, err := req.RequireString("cluster_id")
if err != nil {
return errResult("spark_submit: " + err.Error()), nil
}
master, err := req.RequireString("master")
master, err := requireNonEmptyString(args, "master")
if err != nil {
return errResult("spark_submit: " + err.Error()), nil
}
@@ -79,16 +84,29 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
if err != nil {
return errResult("spark_submit: " + err.Error()), nil
}
queue, err := req.RequireString("queue")
queue, err := requireNonEmptyString(args, "queue")
if err != nil {
return errResult("spark_submit: " + err.Error()), nil
}
executorMemory, err := req.RequireString("executor_memory")
executorMemory, err := requireNonEmptyString(args, "executor_memory")
if err != nil {
return errResult("spark_submit: " + err.Error()), nil
}
args := req.GetArguments()
// Reject paths that didn't come through upload_file on this server.
// Cluster-local paths and arbitrary host paths are unreachable from the MCP
// server's spark-submit, so we fail fast with a clear message rather than
// 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, ...).
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()))
}
return errResult(fmt.Sprintf("spark_submit: script_path %q is not from upload_file on this server; upload the file first via upload_file and pass back the path field", scriptPath)), nil
}
// deploy_mode is required for forward compatibility but the builder does not
// emit --deploy-mode, matching the Python reference's actual behavior. If the
@@ -136,25 +154,16 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
}
binary := cl.SparkSubmitExecuteBin
// Build argv inline, mirroring the Python reference's output order exactly.
// DefaultSubmitArgs is no longer prepended; the field is currently unused at
// runtime but kept for backward compatibility with the admin API. Removal is
// a separate follow-up.
cmd := []string{binary}
cmd = append(cmd, "--master", master)
cmd = append(cmd, "--queue", queue)
cmd = append(cmd, "--executor-memory", executorMemory)
cmd = append(cmd, "--executor-cores", strconv.Itoa(executorCores))
cmd = append(cmd, "--num-executors", strconv.Itoa(numExecutors))
for _, k := range sortedStringKeys(sparkConf) {
cmd = append(cmd, "--conf", k+"="+sparkConf[k])
}
for _, k := range sortedStringKeys(extraArgs) {
cmd = append(cmd, "--"+k, extraArgs[k])
}
cmd = append(cmd, scriptPath)
cmd := buildSparkSubmitCommand(SparkSubmitCommandOpts{
Master: master,
Queue: queue,
ExecutorMemory: executorMemory,
ExecutorCores: executorCores,
NumExecutors: numExecutors,
SparkConf: sparkConf,
ExtraArgs: extraArgs,
ScriptPath: scriptPath,
})
if d.Logger != nil {
d.Logger.Debug("spark_submit.built", "cluster_id", clusterID, "argv", cmd)
@@ -167,31 +176,92 @@ 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{
"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
// spark-submit argv. The returned argv does NOT include the binary itself.
type SparkSubmitCommandOpts struct {
Master string
Queue string
ExecutorMemory string
ExecutorCores int
NumExecutors int
SparkConf map[string]string
ExtraArgs map[string]string
ScriptPath string
}
// buildSparkSubmitCommand builds the post-binary argv for spark-submit.
// DefaultSubmitArgs is no longer prepended; the field is currently unused at
// runtime but kept for backward compatibility with the admin API. Removal is
// a separate follow-up.
func buildSparkSubmitCommand(opts SparkSubmitCommandOpts) []string {
cmd := []string{"--master", opts.Master}
cmd = append(cmd, "--queue", opts.Queue)
cmd = append(cmd, "--executor-memory", opts.ExecutorMemory)
cmd = append(cmd, "--executor-cores", strconv.Itoa(opts.ExecutorCores))
cmd = append(cmd, "--num-executors", strconv.Itoa(opts.NumExecutors))
for _, k := range sortedStringKeys(opts.SparkConf) {
cmd = append(cmd, "--conf", k+"="+opts.SparkConf[k])
}
for _, k := range sortedStringKeys(opts.ExtraArgs) {
cmd = append(cmd, "--"+k, opts.ExtraArgs[k])
}
cmd = append(cmd, opts.ScriptPath)
return cmd
}
// requireNonNegativeInt extracts an integer parameter from the request and
@@ -248,3 +318,17 @@ func sortedStringKeys(m map[string]string) []string {
sort.Strings(keys)
return keys
}
// requireNonEmptyString extracts a required string parameter and rejects empty
// values. MCP's RequireString already enforces presence/type; this enforces
// that the field is not blank.
func requireNonEmptyString(args map[string]any, key string) (string, error) {
s, ok := args[key].(string)
if !ok {
return "", fmt.Errorf("%s is required", key)
}
if s == "" {
return "", fmt.Errorf("%s is required", key)
}
return s, nil
}
+190 -7
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)
}
@@ -66,7 +70,6 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
lines := strings.Split(strings.TrimSpace(string(got)), "\n")
want := []string{
echoScript,
"--master", "yarn",
"--queue", "default",
"--executor-memory", "2G",
@@ -80,6 +83,9 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
if len(lines) != len(want) {
t.Fatalf("lines=%v\nwant=%v", lines, want)
}
if lines[0] != "--master" {
t.Errorf("line[0]=%q, want \"--master\"", lines[0])
}
for i, l := range lines {
if l != want[i] {
t.Errorf("line[%d]=%q, want %q", i, l, want[i])
@@ -95,13 +101,117 @@ 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 {
t.Fatalf("save upload: %v", err)
}
req := newToolRequest(SparkSubmitName, map[string]any{
"cluster_id": "cluster-echo",
"deploy_mode": "cluster",
"script_path": "/tmp/script.py",
"script_path": mintedPath,
"queue": "default",
"executor_memory": "2G",
"executor_cores": 1,
@@ -124,13 +234,18 @@ 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 {
t.Fatalf("save upload: %v", err)
}
req := newToolRequest(SparkSubmitName, map[string]any{
"cluster_id": "cluster-echo",
"master": "yarn",
"deploy_mode": "cluster",
"script_path": "/tmp/script.py",
"script_path": mintedPath,
"queue": "default",
"executor_memory": "2G",
"executor_cores": 1,
@@ -152,3 +267,71 @@ func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
t.Errorf("error text=%q, want mention of spark_conf", text.Text)
}
}
func TestSparkSubmit_RejectsNonMintedPath(t *testing.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 {
t.Fatalf("write unminted file: %v", err)
}
req := newToolRequest(SparkSubmitName, map[string]any{
"cluster_id": "cluster-echo",
"master": "yarn",
"deploy_mode": "cluster",
"script_path": unminted,
"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("expected error result, got: %v", res.Content)
}
text, ok := mcp.AsTextContent(res.Content[0])
if !ok {
t.Fatalf("content is not text: %T", res.Content[0])
}
if !strings.Contains(text.Text, "not from upload_file") {
t.Errorf("error text=%q, want mention of not from upload_file", text.Text)
}
}
func TestSparkSubmit_EmptyMaster(t *testing.T) {
deps, _, _, _ := testDepsWithDataDir(t)
store := deps.UploadStore
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
if err != nil {
t.Fatalf("save upload: %v", err)
}
req := newToolRequest(SparkSubmitName, map[string]any{
"cluster_id": "cluster-echo",
"master": "",
"deploy_mode": "cluster",
"script_path": mintedPath,
"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("expected error result, got: %v", res.Content)
}
text, ok := mcp.AsTextContent(res.Content[0])
if !ok {
t.Fatalf("content is not text: %T", res.Content[0])
}
if !strings.Contains(text.Text, "master is required") && !strings.Contains(text.Text, "master") {
t.Errorf("error text=%q, want mention of master", text.Text)
}
}
+18 -5
View File
@@ -8,6 +8,8 @@ import (
"regexp"
"github.com/mark3labs/mcp-go/mcp"
"spark-mcp-go/internal/audit"
)
const UploadFileName = "upload_file"
@@ -18,7 +20,7 @@ var filenameRegex = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`)
// NewUploadFileTool returns the schema for the upload_file MCP Tool.
func NewUploadFileTool() mcp.Tool {
return mcp.NewTool(UploadFileName,
mcp.WithDescription("Upload a script or config file to the server's data/uploads/ directory. Returns {file_id, path, name, size, sha256}. To submit it, pass the returned `path` as one of the args in spark_submit (e.g. args=[<path>, '--class', 'Main']). The path is server-side; the agent does not need to construct one."),
mcp.WithDescription("Upload a script or config file to the server's data/uploads/ directory. Returns {file_id, path, name, size, sha256}. To submit it, pass the returned `path` as the `script_path` field in spark_submit, along with the structured fields (master, deploy_mode, queue, executor_memory, executor_cores, num_executors; optional spark_conf, extra_args)."),
mcp.WithString("filename",
mcp.Required(),
mcp.Description("Plain file name without path separators (1-128 chars, [a-zA-Z0-9._-])"),
@@ -59,10 +61,6 @@ func (d *Deps) UploadFileHandler(ctx context.Context, req mcp.CallToolRequest) (
return errResult("upload_file: " + err.Error()), nil
}
if d.UploadStore == nil {
return errResult("upload_file: upload store not configured"), nil
}
var data []byte
switch encoding {
case "base64":
@@ -80,6 +78,21 @@ func (d *Deps) UploadFileHandler(ctx context.Context, req mcp.CallToolRequest) (
return errResult("upload_file: save upload: " + err.Error()), nil
}
if d.AuditRepo != nil {
details, _ := audit.MarshalDetails(map[string]any{
"file_id": fileID,
"name": filename,
"size": size,
"sha256": sha256Hex,
})
_ = d.AuditRepo.Insert(ctx, &audit.Entry{
Actor: "tool:upload_file",
Action: audit.ActionUploadCreate,
ClusterID: fileID,
Details: details,
})
}
result := map[string]any{
"file_id": fileID,
"path": absPath,
+22
View File
@@ -38,6 +38,28 @@ CREATE TABLE IF NOT EXISTS audit_log (
details TEXT
);
CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(timestamp DESC);
CREATE TABLE IF NOT EXISTS upload_files (
file_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
size INTEGER NOT NULL,
sha256 TEXT NOT NULL,
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()
}
+149
View File
@@ -0,0 +1,149 @@
package storage
import (
"context"
"database/sql"
"errors"
"fmt"
"strings"
"time"
)
// UploadMeta is the database index record for an uploaded file.
//
// The .meta.json sidecar remains the source of truth; this struct is the
// queryable mirror used by the admin UI.
type UploadMeta struct {
FileID string `json:"file_id"`
Name string `json:"name"`
Size int64 `json:"size"`
SHA256 string `json:"sha256"`
UploadedAt time.Time `json:"uploaded_at"`
}
// UploadRepo provides CRUD operations for upload file index records.
type UploadRepo struct {
db *DB
}
// Uploads returns a repository bound to this DB handle.
func (d *DB) Uploads() *UploadRepo {
return &UploadRepo{db: d}
}
// Create inserts a new upload index record. Timestamps are stored as Unix
// nanoseconds for stable ORDER BY semantics.
func (r *UploadRepo) Create(ctx context.Context, fileID, name string, size int64, sha256Hex string, uploadedAt time.Time) error {
_, err := r.db.sqlDB.ExecContext(ctx, `
INSERT INTO upload_files (file_id, name, size, sha256, uploaded_at)
VALUES (?, ?, ?, ?, ?)`,
fileID, name, size, sha256Hex, uploadedAt.UnixNano(),
)
if err != nil {
if strings.Contains(err.Error(), "UNIQUE constraint failed") {
return fmt.Errorf("storage: create upload %s: primary key conflict: %w", fileID, err)
}
return fmt.Errorf("storage: create upload %s: %w", fileID, err)
}
return nil
}
// Get retrieves an upload index record by file_id. If the record does not
// exist, it returns ErrNotFound.
func (r *UploadRepo) Get(ctx context.Context, fileID string) (UploadMeta, error) {
row := r.db.sqlDB.QueryRowContext(ctx, `
SELECT file_id, name, size, sha256, uploaded_at
FROM upload_files
WHERE file_id = ?`, fileID)
m, err := scanUpload(row.Scan)
if errors.Is(err, sql.ErrNoRows) {
return UploadMeta{}, ErrNotFound
}
if err != nil {
return UploadMeta{}, fmt.Errorf("storage: get upload %s: %w", fileID, err)
}
return m, nil
}
// List returns upload index records ordered by uploaded_at descending.
//
// 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 {
limit = 100
}
if limit > 500 {
limit = 500
}
var rows *sql.Rows
var err error
if search != "" {
rows, err = r.db.sqlDB.QueryContext(ctx, `
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+"%", "%"+search+"%", limit,
)
} else {
rows, err = r.db.sqlDB.QueryContext(ctx, `
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 {
return nil, fmt.Errorf("storage: list uploads: %w", err)
}
defer rows.Close()
var out []UploadMeta
for rows.Next() {
m, err := scanUpload(rows.Scan)
if err != nil {
return nil, fmt.Errorf("storage: list uploads: %w", err)
}
out = append(out, m)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("storage: list uploads: %w", err)
}
if out == nil {
// Force empty slice (not null) in JSON.
out = []UploadMeta{}
}
return out, nil
}
// Delete removes an upload index record by file_id. If the record does not
// exist, it returns ErrNotFound.
func (r *UploadRepo) Delete(ctx context.Context, fileID string) error {
res, err := r.db.sqlDB.ExecContext(ctx, `DELETE FROM upload_files WHERE file_id = ?`, fileID)
if err != nil {
return fmt.Errorf("storage: delete upload %s: %w", fileID, err)
}
n, err := res.RowsAffected()
if err != nil {
return fmt.Errorf("storage: delete upload %s: rows affected: %w", fileID, err)
}
if n == 0 {
return ErrNotFound
}
return nil
}
func scanUpload(scan scanFunc) (UploadMeta, error) {
var m UploadMeta
var uploadedAt int64
if err := scan(&m.FileID, &m.Name, &m.Size, &m.SHA256, &uploadedAt); err != nil {
return UploadMeta{}, err
}
m.UploadedAt = time.Unix(0, uploadedAt)
return m, nil
}
+162
View File
@@ -0,0 +1,162 @@
package storage
import (
"context"
"errors"
"fmt"
"testing"
"time"
)
func newUploadRepo(t *testing.T) *UploadRepo {
t.Helper()
db, err := Open(":memory:")
if err != nil {
t.Fatalf("Open in-memory db: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
return db.Uploads()
}
func assertUploadEqual(t *testing.T, got, want UploadMeta) {
t.Helper()
if got.FileID != want.FileID {
t.Errorf("FileID: got %q, want %q", got.FileID, want.FileID)
}
if got.Name != want.Name {
t.Errorf("Name: got %q, want %q", got.Name, want.Name)
}
if got.Size != want.Size {
t.Errorf("Size: got %d, want %d", got.Size, want.Size)
}
if got.SHA256 != want.SHA256 {
t.Errorf("SHA256: got %q, want %q", got.SHA256, want.SHA256)
}
if !got.UploadedAt.Equal(want.UploadedAt) {
t.Errorf("UploadedAt: got %v, want %v", got.UploadedAt, want.UploadedAt)
}
}
func TestUploadRepo_CreateGetListDelete(t *testing.T) {
repo := newUploadRepo(t)
ctx := context.Background()
uploadedAt := time.Unix(0, time.Now().UnixNano())
want := UploadMeta{
FileID: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
Name: "report.csv",
Size: 42,
SHA256: "deadbeef",
UploadedAt: uploadedAt,
}
if err := repo.Create(ctx, want.FileID, want.Name, want.Size, want.SHA256, want.UploadedAt); err != nil {
t.Fatalf("Create: %v", err)
}
got, err := repo.Get(ctx, want.FileID)
if err != nil {
t.Fatalf("Get: %v", err)
}
assertUploadEqual(t, got, want)
if err := repo.Create(ctx, want.FileID, "other", 1, "abcd", time.Now()); err == nil {
t.Errorf("duplicate Create succeeded, want error")
}
list, err := repo.List(ctx, "", 100)
if err != nil {
t.Fatalf("List: %v", err)
}
if len(list) != 1 {
t.Fatalf("List len: got %d, want 1", len(list))
}
assertUploadEqual(t, list[0], want)
if err := repo.Delete(ctx, want.FileID); err != nil {
t.Fatalf("Delete: %v", err)
}
if _, err := repo.Get(ctx, want.FileID); !errors.Is(err, ErrNotFound) {
t.Fatalf("Get after Delete: got %v, want ErrNotFound", err)
}
if err := repo.Delete(ctx, want.FileID); !errors.Is(err, ErrNotFound) {
t.Fatalf("Delete again: got %v, want ErrNotFound", err)
}
}
func TestUploadRepo_ListOrderAndSearch(t *testing.T) {
repo := newUploadRepo(t)
ctx := context.Background()
base := time.Unix(0, time.Now().UnixNano())
records := []UploadMeta{
{FileID: "00000000000000000000000000000001", Name: "alpha.csv", Size: 1, SHA256: "a", UploadedAt: base.Add(-2 * time.Hour)},
{FileID: "00000000000000000000000000000002", Name: "beta.log", Size: 2, SHA256: "b", UploadedAt: base.Add(-1 * time.Hour)},
{FileID: "00000000000000000000000000000003", Name: "gamma.csv", Size: 3, SHA256: "c", UploadedAt: base},
}
for _, r := range records {
if err := repo.Create(ctx, r.FileID, r.Name, r.Size, r.SHA256, r.UploadedAt); err != nil {
t.Fatalf("Create %s: %v", r.FileID, err)
}
}
all, err := repo.List(ctx, "", 100)
if err != nil {
t.Fatalf("List: %v", err)
}
if len(all) != 3 {
t.Fatalf("List len: got %d, want 3", len(all))
}
wantOrder := []string{"00000000000000000000000000000003", "00000000000000000000000000000002", "00000000000000000000000000000001"}
for i, id := range wantOrder {
if all[i].FileID != id {
t.Errorf("List order[%d]: got %q, want %q", i, all[i].FileID, id)
}
}
csv, err := repo.List(ctx, "csv", 100)
if err != nil {
t.Fatalf("List search: %v", err)
}
if len(csv) != 2 {
t.Fatalf("search csv len: got %d, want 2", len(csv))
}
if csv[0].Name != "gamma.csv" || csv[1].Name != "alpha.csv" {
t.Errorf("search csv order: got %v", []string{csv[0].Name, csv[1].Name})
}
beta, err := repo.List(ctx, "beta", 100)
if err != nil {
t.Fatalf("List search beta: %v", err)
}
if len(beta) != 1 || beta[0].Name != "beta.log" {
t.Errorf("search beta: got %+v", beta)
}
}
func TestUploadRepo_ListLimitCap(t *testing.T) {
repo := newUploadRepo(t)
ctx := context.Background()
base := time.Unix(0, time.Now().UnixNano())
for i := 0; i < 10; i++ {
id := fmt.Sprintf("%032d", i)
if err := repo.Create(ctx, id, "x", int64(i), "h", base.Add(time.Duration(i)*time.Second)); err != nil {
t.Fatalf("Create %d: %v", i, err)
}
}
list, err := repo.List(ctx, "", 0)
if err != nil {
t.Fatalf("List default limit: %v", err)
}
// Default limit is 100 but only 10 rows exist.
if len(list) != 10 {
t.Errorf("default limit: got %d, want 10", len(list))
}
list, err = repo.List(ctx, "", 501)
if err != nil {
t.Fatalf("List cap: %v", err)
}
if len(list) != 10 {
t.Errorf("cap limit: got %d, want 10", len(list))
}
// Explicit small limit is respected.
list, err = repo.List(ctx, "", 3)
if err != nil {
t.Fatalf("List small limit: %v", err)
}
if len(list) != 3 {
t.Errorf("small limit: got %d, want 3", len(list))
}
}
+107 -15
View File
@@ -4,16 +4,21 @@
package uploads
import (
"context"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"os"
"path/filepath"
"regexp"
"strings"
"time"
"spark-mcp-go/internal/storage"
)
var fileIDRegex = regexp.MustCompile(`^[0-9a-f]{32}$`)
@@ -21,35 +26,49 @@ var fileIDRegex = regexp.MustCompile(`^[0-9a-f]{32}$`)
// Store is a local file store rooted at Root.
type Store struct {
Root string
repo UploadsDB
}
// UploadsDB is the minimal surface the Store needs from an upload index.
type UploadsDB interface {
Create(ctx context.Context, fileID, name string, size int64, sha256Hex string, uploadedAt time.Time) error
Get(ctx context.Context, fileID string) (storage.UploadMeta, error)
}
// SetRepo binds an upload index repository for dual-write. The DB is an
// index only; failures are logged but never fail the file write because
// .meta.json is the source of truth and startup backfill can recover.
func (s *Store) SetRepo(repo UploadsDB) {
s.repo = repo
}
// New creates a Store rooted at root. It creates root with mode 0o750 if it
// does not exist. The root must be an absolute path.
func New(root string) (*Store, error) {
func New(root string) (Store, error) {
if root == "" {
return nil, fmt.Errorf("uploads: root must not be empty")
return Store{}, fmt.Errorf("uploads: root must not be empty")
}
absRoot, err := filepath.Abs(root)
if err != nil {
return nil, fmt.Errorf("uploads: resolve root: %w", err)
return Store{}, fmt.Errorf("uploads: resolve root: %w", err)
}
if !filepath.IsAbs(absRoot) {
return nil, fmt.Errorf("uploads: root %q is not an absolute path", absRoot)
return Store{}, fmt.Errorf("uploads: root %q is not an absolute path", absRoot)
}
if err := os.MkdirAll(absRoot, 0o750); err != nil {
return nil, fmt.Errorf("uploads: create root: %w", err)
return Store{}, fmt.Errorf("uploads: create root: %w", err)
}
info, err := os.Stat(absRoot)
if err != nil {
return nil, fmt.Errorf("uploads: stat root: %w", err)
return Store{}, fmt.Errorf("uploads: stat root: %w", err)
}
if !info.IsDir() {
return nil, fmt.Errorf("uploads: root %q is not a directory", absRoot)
return Store{}, fmt.Errorf("uploads: root %q is not a directory", absRoot)
}
return &Store{Root: absRoot}, nil
return Store{Root: absRoot}, nil
}
type sidecar struct {
@@ -110,6 +129,13 @@ func (s *Store) Save(data []byte, originalName string) (fileID, name string, siz
return "", "", 0, "", "", fmt.Errorf("uploads: write sidecar: %w", err)
}
// Index the upload in the DB as a best-effort mirror of the sidecar.
if s.repo != nil {
if dbErr := s.repo.Create(context.Background(), fileID, originalName, meta.Size, sha256Hex, meta.UploadedAt); dbErr != nil {
slog.Default().Warn("uploads: failed to index upload in DB", "file_id", fileID, "err", dbErr)
}
}
return fileID, originalName, meta.Size, sha256Hex, absPath, nil
}
@@ -137,6 +163,10 @@ func (s *Store) Validate(absPath string) (string, error) {
return "", fmt.Errorf("uploads: path %q is not under Root", absPath)
}
if strings.HasSuffix(rel, ".meta.json") {
return "", fmt.Errorf("uploads: path %q is a sidecar, pass the data file path", absPath)
}
fileID := strings.TrimSuffix(rel, ".meta.json")
if !fileIDRegex.MatchString(fileID) {
return "", fmt.Errorf("uploads: path %q does not denote a valid upload", absPath)
@@ -156,7 +186,7 @@ func (s *Store) Validate(absPath string) (string, error) {
// Sweep deletes data files (and their sidecars) whose uploaded_at is older
// than ttl, as well as orphan sidecars that have no matching data file. For
// data files whose sidecar is missing, the file's mtime is used as a fallback.
func (s *Store) Sweep(ttl time.Duration) (deleted int, err error) {
func (s *Store) Sweep(ctx context.Context, ttl time.Duration) (deleted int, err error) {
entries, err := os.ReadDir(s.Root)
if err != nil {
return 0, fmt.Errorf("uploads: read root: %w", err)
@@ -193,18 +223,24 @@ func (s *Store) Sweep(ttl time.Duration) (deleted int, err error) {
cutoff := time.Now().Add(-ttl)
for fileID, it := range items {
if err := ctx.Err(); err != nil {
return deleted, err
}
dataPath := filepath.Join(s.Root, fileID)
metaPath := dataPath + ".meta.json"
if it.data {
expired := false
corruptSidecar := false
if it.meta {
uploadedAt, parseErr := readUploadedAt(metaPath)
if parseErr == nil && uploadedAt.Before(cutoff) {
expired = true
} else if parseErr != nil {
corruptSidecar = true
}
}
if !expired && !it.meta {
if !expired && (corruptSidecar || !it.meta) {
info, statErr := os.Stat(dataPath)
if statErr == nil && info.ModTime().Before(cutoff) {
expired = true
@@ -229,6 +265,54 @@ func (s *Store) Sweep(ttl time.Duration) (deleted int, err error) {
return deleted, nil
}
// Backfill walks Root and inserts a DB index row for every existing
// .meta.json sidecar that is not already indexed. It is the recovery path
// for sidecars created before the DB table existed.
func (s *Store) Backfill(ctx context.Context, repo UploadsDB) (inserted int, err error) {
entries, err := os.ReadDir(s.Root)
if err != nil {
return 0, fmt.Errorf("uploads: backfill read root: %w", err)
}
for _, e := range entries {
if err := ctx.Err(); err != nil {
return inserted, err
}
name := e.Name()
base := strings.TrimSuffix(name, ".meta.json")
if base == name || !fileIDRegex.MatchString(base) {
continue
}
metaPath := filepath.Join(s.Root, name)
sc, err := readSidecar(metaPath)
if err != nil {
// Skip corrupt sidecars; Sweep will reap them later.
continue
}
_, getErr := repo.Get(ctx, base)
if getErr == nil {
continue
}
if !errors.Is(getErr, storage.ErrNotFound) {
return inserted, fmt.Errorf("uploads: backfill lookup %s: %w", base, getErr)
}
createErr := repo.Create(ctx, base, sc.Name, sc.Size, sc.Sha256, sc.UploadedAt)
if createErr != nil {
if strings.Contains(createErr.Error(), "UNIQUE constraint failed") {
continue
}
return inserted, fmt.Errorf("uploads: backfill index %s: %w", base, createErr)
}
inserted++
}
return inserted, nil
}
func newFileID() (string, error) {
var b [16]byte
if _, err := rand.Read(b[:]); err != nil {
@@ -238,16 +322,24 @@ func newFileID() (string, error) {
}
func readUploadedAt(path string) (time.Time, error) {
data, err := os.ReadFile(path)
sc, err := readSidecar(path)
if err != nil {
return time.Time{}, err
}
var sc sidecar
if err := json.Unmarshal(data, &sc); err != nil {
return time.Time{}, err
}
if sc.UploadedAt.IsZero() {
return time.Time{}, fmt.Errorf("uploads: missing uploaded_at")
}
return sc.UploadedAt, nil
}
func readSidecar(path string) (sidecar, error) {
var sc sidecar
data, err := os.ReadFile(path)
if err != nil {
return sc, err
}
if err := json.Unmarshal(data, &sc); err != nil {
return sc, err
}
return sc, nil
}
+277 -2
View File
@@ -1,15 +1,19 @@
package uploads
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"regexp"
"testing"
"time"
"spark-mcp-go/internal/storage"
)
func TestStore_SaveAndRetrieve(t *testing.T) {
@@ -201,7 +205,7 @@ func TestStore_Sweep_DeletesExpiredKeepsRecent(t *testing.T) {
create(freshID, timePtr(now), true)
create(noMetaID, nil, true)
deleted, err := store.Sweep(1 * time.Hour)
deleted, err := store.Sweep(context.Background(), 1*time.Hour)
if err != nil {
t.Fatalf("sweep: %v", err)
}
@@ -241,7 +245,7 @@ func TestStore_Sweep_OrphanSidecarReaped(t *testing.T) {
t.Fatalf("create orphan sidecar: %v", err)
}
deleted, err := store.Sweep(1 * time.Hour)
deleted, err := store.Sweep(context.Background(), 1*time.Hour)
if err != nil {
t.Fatalf("sweep: %v", err)
}
@@ -256,3 +260,274 @@ func TestStore_Sweep_OrphanSidecarReaped(t *testing.T) {
func timePtr(t time.Time) *time.Time {
return &t
}
func TestStore_Validate_RejectsSidecarPath(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
fileID := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
if err := os.WriteFile(filepath.Join(store.Root, fileID), []byte("x"), 0o640); err != nil {
t.Fatalf("create data file: %v", err)
}
if err := os.WriteFile(filepath.Join(store.Root, fileID+".meta.json"), []byte(`{"uploaded_at":"`+time.Now().Format(time.RFC3339Nano)+`"}`), 0o600); err != nil {
t.Fatalf("create sidecar: %v", err)
}
_, err = store.Validate(filepath.Join(store.Root, fileID+".meta.json"))
if err == nil {
t.Errorf("Validate(sidecar) succeeded, want error")
}
}
func TestStore_Sweep_HonorsContextCancel(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
id := "00000000000000000000000000000006"
if err := os.WriteFile(filepath.Join(store.Root, id), []byte("x"), 0o640); err != nil {
t.Fatalf("create data file: %v", err)
}
sc := map[string]any{
"name": "x",
"size": 1,
"sha256": "abc",
"uploaded_at": time.Now().Add(-2 * time.Hour).Format(time.RFC3339Nano),
}
b, _ := json.Marshal(sc)
if err := os.WriteFile(filepath.Join(store.Root, id+".meta.json"), b, 0o600); err != nil {
t.Fatalf("create sidecar: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
deleted, err := store.Sweep(ctx, 1*time.Hour)
if err != ctx.Err() {
t.Fatalf("sweep err=%v, want %v", err, ctx.Err())
}
if deleted != 0 {
t.Errorf("deleted=%d, want 0", deleted)
}
if _, existErr := os.Stat(filepath.Join(store.Root, id)); os.IsNotExist(existErr) {
t.Logf("cancelled sweep managed to delete the file before checking ctx; this is timing-dependent and acceptable")
}
}
func TestStore_Sweep_DeletesDataWithCorruptSidecar(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
id := "00000000000000000000000000000005"
dataPath := filepath.Join(store.Root, id)
metaPath := dataPath + ".meta.json"
if err := os.WriteFile(dataPath, []byte("stale data"), 0o640); err != nil {
t.Fatalf("create data file: %v", err)
}
// Corrupt sidecar: not valid JSON.
if err := os.WriteFile(metaPath, []byte("not valid json"), 0o600); err != nil {
t.Fatalf("create sidecar: %v", err)
}
// Set mtime well in the past so the fallback triggers.
past := time.Now().Add(-2 * time.Hour)
if err := os.Chtimes(dataPath, past, past); err != nil {
t.Fatalf("set mtime: %v", err)
}
deleted, err := store.Sweep(context.Background(), 1*time.Hour)
if err != nil {
t.Fatalf("sweep: %v", err)
}
if deleted != 1 {
t.Errorf("deleted=%d, want 1", deleted)
}
if _, err := os.Stat(dataPath); !os.IsNotExist(err) {
t.Errorf("data file with corrupt sidecar still exists")
}
if _, err := os.Stat(metaPath); !os.IsNotExist(err) {
t.Errorf("corrupt sidecar still exists")
}
}
type fakeUploadRepo struct {
calls []storage.UploadMeta
err error
}
func (f *fakeUploadRepo) Create(ctx context.Context, fileID, name string, size int64, sha256Hex string, uploadedAt time.Time) error {
f.calls = append(f.calls, storage.UploadMeta{
FileID: fileID, Name: name, Size: size, SHA256: sha256Hex, UploadedAt: uploadedAt,
})
return f.err
}
func (f *fakeUploadRepo) Get(ctx context.Context, fileID string) (storage.UploadMeta, error) {
for _, m := range f.calls {
if m.FileID == fileID {
return m, nil
}
}
return storage.UploadMeta{}, storage.ErrNotFound
}
func (f *fakeUploadRepo) List(ctx context.Context, search string, limit int) ([]storage.UploadMeta, error) {
return f.calls, nil
}
func (f *fakeUploadRepo) Delete(ctx context.Context, fileID string) error {
return nil
}
func TestStore_Save_DualWrite(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{}
store.SetRepo(repo)
data := []byte("dual-write test")
fileID, _, size, sha256Hex, absPath, err := store.Save(data, "dual.txt")
if err != nil {
t.Fatalf("save: %v", err)
}
if absPath == "" {
t.Fatal("absPath empty")
}
if len(repo.calls) != 1 {
t.Fatalf("repo.Create calls: got %d, want 1", len(repo.calls))
}
call := repo.calls[0]
if call.FileID != fileID {
t.Errorf("FileID: got %q, want %q", call.FileID, fileID)
}
if call.Name != "dual.txt" {
t.Errorf("Name: got %q, want dual.txt", call.Name)
}
if call.Size != size {
t.Errorf("Size: got %d, want %d", call.Size, size)
}
if call.SHA256 != sha256Hex {
t.Errorf("SHA256: got %q, want %q", call.SHA256, sha256Hex)
}
if call.UploadedAt.IsZero() {
t.Errorf("UploadedAt is zero")
}
}
func TestStore_Save_RepoErrorDoesNotFailUpload(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{err: errors.New("db down")}
store.SetRepo(repo)
data := []byte("db error test")
fileID, _, _, _, absPath, err := store.Save(data, "error.txt")
if err != nil {
t.Fatalf("save should not fail when repo errors: %v", err)
}
if _, err := os.Stat(absPath); err != nil {
t.Errorf("data file missing: %v", err)
}
if _, err := os.Stat(absPath + ".meta.json"); err != nil {
t.Errorf("sidecar missing: %v", err)
}
if len(repo.calls) != 1 {
t.Errorf("repo.Create calls: got %d, want 1", len(repo.calls))
}
if repo.calls[0].FileID != fileID {
t.Errorf("repo call file_id: got %q, want %q", repo.calls[0].FileID, fileID)
}
}
func TestStore_Backfill(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{}
now := time.Now()
ids := []string{
"00000000000000000000000000000001",
"00000000000000000000000000000002",
"00000000000000000000000000000003",
}
for i, id := range ids {
sc := map[string]any{
"name": fmt.Sprintf("file%d.txt", i),
"size": i + 1,
"sha256": fmt.Sprintf("sha%d", i),
"uploaded_at": now.Add(time.Duration(i) * time.Second).Format(time.RFC3339Nano),
}
b, _ := json.Marshal(sc)
if err := os.WriteFile(filepath.Join(store.Root, id+".meta.json"), b, 0o600); err != nil {
t.Fatalf("create sidecar %s: %v", id, err)
}
}
inserted, err := store.Backfill(context.Background(), repo)
if err != nil {
t.Fatalf("backfill: %v", err)
}
if inserted != 3 {
t.Errorf("inserted=%d, want 3", inserted)
}
if len(repo.calls) != 3 {
t.Errorf("repo.Create calls: got %d, want 3", len(repo.calls))
}
// Second backfill must skip existing rows.
inserted, err = store.Backfill(context.Background(), repo)
if err != nil {
t.Fatalf("second backfill: %v", err)
}
if inserted != 0 {
t.Errorf("second inserted=%d, want 0", inserted)
}
if len(repo.calls) != 3 {
t.Errorf("repo.Create calls after second backfill: got %d, want 3", len(repo.calls))
}
}
func TestStore_Backfill_SkipsExisting(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{}
id := "00000000000000000000000000000004"
now := time.Now()
sc := map[string]any{
"name": "existing.txt",
"size": 5,
"sha256": "sha",
"uploaded_at": now.Format(time.RFC3339Nano),
}
b, _ := json.Marshal(sc)
if err := os.WriteFile(filepath.Join(store.Root, id+".meta.json"), b, 0o600); err != nil {
t.Fatalf("create sidecar: %v", err)
}
// Pre-seed the repo so the sidecar is already indexed.
if err := repo.Create(context.Background(), id, "existing.txt", 5, "sha", now); err != nil {
t.Fatalf("seed repo: %v", err)
}
inserted, err := store.Backfill(context.Background(), repo)
if err != nil {
t.Fatalf("backfill: %v", err)
}
if inserted != 0 {
t.Errorf("inserted=%d, want 0", inserted)
}
}
+24 -3
View File
@@ -30,6 +30,8 @@ import (
"spark-mcp-go/internal/uploads"
)
const startupSweepTimeout = 5 * time.Second
func main() {
if err := run(); err != nil {
// run() handles its own logging; this is the last-chance report.
@@ -71,13 +73,30 @@ func run() error {
if err != nil {
return err
}
n, err := uploadStore.Sweep(cfg.UploadTTL)
if err != nil {
sweepCtx, sweepCancel := context.WithTimeout(context.Background(), startupSweepTimeout)
n, err := uploadStore.Sweep(sweepCtx, cfg.UploadTTL)
sweepCancel()
if errors.Is(err, context.DeadlineExceeded) {
logger.Warn("uploads.sweep.timeout", "deleted", n, "err", err)
} else if err != nil {
logger.Error("uploads.sweep", "err", err)
} else {
logger.Info("uploads.sweep", "deleted", n)
}
backfillCtx, backfillCancel := context.WithTimeout(context.Background(), startupSweepTimeout)
n, err = uploadStore.Backfill(backfillCtx, db.Uploads())
backfillCancel()
if errors.Is(err, context.DeadlineExceeded) {
logger.Warn("uploads.backfill.timeout", "inserted", n, "err", err)
} else if err != nil {
logger.Error("uploads.backfill", "err", err)
} else {
logger.Info("uploads.backfill", "inserted", n)
}
uploadStore.SetRepo(db.Uploads())
gin.SetMode(cfg.GinMode)
r := gin.New()
r.Use(gin.Recovery())
@@ -86,10 +105,11 @@ func run() error {
c.JSON(http.StatusOK, gin.H{"ok": true, "version": "0.0.0"})
})
admin.Mount(r, db.Clusters(), audit.NewRepo(db), 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"),
AuditRepo: audit.NewRepo(db),
ClusterRepo: db.Clusters(),
SparkSubmitTimeout: cfg.SparkSubmitTimeout,
HTTPClient: httpclient.New(httpclient.Config{
@@ -99,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,