Files

T

tao.chenandClaude a4e2472716 Phase 0+1+2 续: 项目骨架 + 配置 + 日志 + 存储层

Phase 0 (项目骨架):
- main.go 改为最小骨架 (config + logging + gin + /healthz + 优雅退出)
- internal/{config,logging,storage,cluster,...}/ 目录占位

Phase 1 (配置 + 日志):
- internal/config: env 解析 + 必填校验 + token 隐藏的 String()
- internal/logging: slog multi-handler 双输出 (终端 text + 主文件 JSON)
  + StartToolCall per-tool 独立文件 (0600), tools 目录 0700

Phase 2 续 (存储层):
- internal/cluster: 16 字段 Cluster struct (AuthPassword json:"-")
- internal/storage: modernc.org/sqlite 接入, WAL 模式, schema 自动迁移
- ClusterRepo CRUD: Create/Get/List/Update/Delete + ErrNotFound
  + AuthPassword 空字符串 = 保留旧密码 (核心约定)
- 7 个单元测试全绿 (:memory: DB)
- created_at/updated_at 改纳秒精度, 消除 List 测试的 sleep 特殊 case

Co-Authored-By: Claude <noreply@anthropic.com>

2026-07-10 12:12:53 +08:00

35 KiB

Raw Blame History

Spark MCP Server 实现计划

创建于 2026-07-10 当前状态: 仓库 init 完毕,等待开干范围: 7 个阶段,预计总工作量 12-15 小时 v0.3 修订: 加入 spark_submit Tool + upload_file 改为本地文件存储

目标

构建一个 MCP (Model Context Protocol) Server for Apache Spark 运维,作为统一入口让 LLM Agent 通过:

YARN RM / SHS REST API 监控任务
本地 spark-submit / spark2-submit 提交任务
通用 HTTP fetch 查询任意 endpoint
本地文件存储上传脚本/JAR 给 spark-submit 引用

同时提供 Gin HTTP 管理后台配置集群。

非目标 (本次不做):

集群部署 / 节点管理
Spark UI 渲染 (LLM 自己解析 JSON)
多租户 / OAuth / mTLS (留接口)

已确认的架构决策

决策点	选择	备注
存储	SQLite (替换 go.mod 里的 MongoDB)	单一持久化层,消除 special case
Spark 能力	读 + 写 (本地提交)	YARN RM/SHS 读,Server 本地 spark-submit 写
MCP 传输	Streamable HTTP (MCP 2025-03+ 标准)	mark3labs/mcp-go 原生支持
Tool 表面	5 个 Tool	见下方 Tool 矩阵
HTTP 框架	Gin	Admin API + 挂载 MCP HTTP handler
SQLite 驱动	modernc.org/sqlite	纯 Go,无 CGO
spark-submit 路径	admin 后台按 cluster 配置	每个 cluster 配单一 `spark_submit_execute_bin` (v0.3.1 合并)
spark-submit 执行位置	Server 本地 (无 SSH)	Server 就是 gateway 机器

架构总览

Tool 矩阵 (v0.3 最终方案)

Tool	类型	用途	底层
`list_clusters`	发现入口 (强制)	返回所有 cluster 的完整元信息 (id, name, rm_url, shs_url, spark_submit_execute_bin, auth, ssl, is_active)。LLM 调它是任何后续操作的前提	SQLite `clusters` 表
`spark_submit`	业务高层	用 cluster 配置的 binary 提交任务	本地 `exec.Command(binary, args...)`
`list_applications`	业务高层 (RM)	列出 YARN 应用,支持 state/queue/limit 过滤	RM `GET /ws/v1/cluster/apps`
`get_application_status`	业务高层 (RM)	取单个 app 的 state + amContainerLogs 字段	RM `GET /ws/v1/cluster/apps/{app_id}`
`get_application_logs`	业务高层 (RM)	拉聚合 container 日志,404/501 降级到 amContainerLogs + NM	RM `GET /ws/v1/cluster/apps/{app_id}/aggregated-logs` + 3xx 跟随
`kill_application`	业务高层 (RM)	杀掉应用	RM `PUT /ws/v1/cluster/apps/{app_id}/state`
`fetch_spark_metrics`	业务高层 (SHS)	拿 SHS 应用结构化数据 (jobs/stages/executors)	SHS `/api/v1/applications/{app_id}/...` (内部 `fetch_url`)
`fetch_cluster_env`	业务高层 (RM)	拿 Yarn 集群资源状态 (nodes/metrics)	RM `/ws/v1/cluster/nodes` (内部 `fetch_url`)
`analyze_spark_log`	业务高层 (分析)	启发式检测 + 自动抓相关日志 (走 RM 链路) + LLM-Ready Prompt	规则引擎 + `get_application_logs` (404/501 降级)
`fetch_url`	通用 HTTP 原语	LLM 拼任意 URL 抓数据 (SHS 端点都走这条)	共享 HTTP 客户端 + SSRF + cluster allowlist
`upload_file`	本地文件存储	LLM 把脚本/JAR 存到 Server `./data/` 供 spark_submit 引用	本地文件系统

Tool 设计取舍:

原 5 个 Tool 方案,现在仍 5 个,但重组成 5 个真正不同的概念
fetch_url 是 LLM 唯一的"主动 HTTP 出口" — 业务 Tool 不重新发明轮子
upload_file 不是 HTTP 上传到远程,而是本地文件写入(LLM 调它存脚本,Server 返回本地路径)
spark_submit 用 exec.Command(name, args...) 切片形式传参,LLM 不接触 shell

典型工作流 (LLM 视角)

1. list_clusters           → 看到 cluster "prod" 配置
2. upload_file             → 把 Python 脚本存到 Server 的 ./data/scripts/etl.py
3. spark_submit            → 提交任务,args: ["--master", "yarn", "--name", "etl",
                                          "--deploy-mode", "cluster",
                                          "./data/scripts/etl.py"]
                            → 返回 app_id = "application_12345_0001"
4. yarn_job_manage         → 轮询 status,直到 FINISHED
5. fetch_url               → 抓 SHS 的 /api/v1/applications/application_12345_0001/jobs
                            → 拿结构化 metrics
6. fetch_url (SHS logs)    → 抓失败 executor 的 stderr
7. (LLM 自己做根因分析)

路由设计

:8080
├── /healthz              (公开, 探活)
├── /admin/*              (Gin, 需 ADMIN_TOKEN)
│   ├── GET    /admin/clusters
│   ├── POST   /admin/clusters
│   ├── PUT    /admin/clusters/:id
│   └── DELETE /admin/clusters/:id
└── /mcp                  (Streamable HTTP, 需 AGENT_TOKEN)
    └── POST /mcp

项目目录结构

spark-mcp-go/
├── main.go                       # 入口
├── go.mod / go.sum
├── internal/
│   ├── config/                   # 环境变量加载
│   │   └── config.go
│   ├── storage/                  # SQLite 存储
│   │   ├── sqlite.go             # DB 连接、迁移
│   │   └── cluster_repo.go       # Cluster CRUD
│   ├── cluster/                  # Cluster 业务
│   │   ├── types.go              # Cluster 结构体
│   │   └── handler.go            # /admin/clusters handlers
│   ├── httpclient/               # 共享 HTTP 客户端 (供 fetch_url / yarn_job_manage 用)
│   │   ├── client.go             # 带 timeout / size limit
│   │   └── ssrf.go               # SSRF 防护
│   ├── tools/                    # MCP Tool 实现 (与协议解耦)
│   │   ├── list_clusters.go
│   │   ├── yarn.go
│   │   ├── fetch_url.go
│   │   ├── upload_file.go        # 本地文件存储
│   │   ├── spark_submit.go
│   │   ├── fetch_spark_metrics.go # 高层 SHS 封装
│   │   ├── fetch_cluster_env.go   # 高层 RM 封装
│   │   └── analyze_spark_log.go   # 规则引擎驱动
│   ├── analyzer/                 # 日志/性能分析 (analyze_spark_log 内部)
│   │   ├── rules.go              # 启发式规则定义 (data_skew / gc_pressure / bottleneck)
│   │   └── engine.go             # 规则引擎:metrics → findings
│   ├── executor/                 # 进程执行封装
│   │   └── exec.go               # 安全的 exec.Command 包装
│   ├── mcp/                      # MCP Server 装配
│   │   ├── server.go
│   │   └── register.go           # 注册 5 个 Tool
│   └── middleware/
│       └── auth.go               # Bearer token 鉴权

阶段划分

Phase 0: 清理与依赖调整 (0.5h)

目标: 把 go.mod 调整到位,清空模板代码,建立目录骨架。

T0.1 从 go.mod 移除 go.mongodb.org/mongo-driver/v2
T0.2 添加 modernc.org/sqlite
T0.3 go mod tidy + go build ./... 通过
T0.4 删除 main.go GoLand 模板
T0.5 建立 internal/{config,storage,cluster,httpclient,tools,executor,mcp,middleware}/ 目录,各放占位文件
T0.6 重新写 main.go 空壳
验收: go build ./... 仍能成功,目录结构建立

Executor: Codex

Phase 1: 配置层 (0.5h)

目标: 从环境变量加载配置。

T1.1 internal/config/config.go 定义 Config (v0.3.7 + v0.3.8 加日志配置):
- 字段: ListenAddr (:8080), DataDir (./data), SQLitePath (派生), AdminTokens []string (v0.3.7,逗号分隔 env,必填 ≥1), AgentToken (必填), HTTPClientTimeout (30s), MaxResponseBytes (1MB), SparkSubmitTimeout (60s)
- v0.3.8 日志配置:
  - LogDir (./data/logs)
  - LogLevel (info,可选 debug / warn / error)
  - LogFormat (text 终端 / json 主文件)
- v0.3.7 analyzer 阈值改为 env:
  - ANALYZER_DATA_SKEW_RATIO (3.0)
  - ANALYZER_GC_PRESSURE_RATIO (0.1)
  - ANALYZER_BOTTLENECK_SHUFFLE_GB (50)
T1.2 main.go 调用 config.Load(),失败即退出
T1.3 启动时打印配置(隐藏 token)
T1.4 启动时 os.MkdirAll(cfg.DataDir, 0755) 确保 data 目录存在
T1.5 Logging 基础设施 (v0.3.8 新,internal/logging/ 包):
- 使用 log/slog (Go 1.21+ stdlib)
- 双输出 multi-handler:
  - 终端: slog.NewTextHandler(os.Stdout, ...)(可读,人类阅读)
  - 主文件: slog.NewJSONHandler(file, ...)(结构化,机器解析)
  - 自实现 MultiHandler fan-out 到两个 handler
- 主日志文件: data/logs/spark-mcp.log (JSON,所有事件,rotate 由运维负责)
- Per-Tool 日志文件 (核心需求): data/logs/tools/{ts}_{tool}_{short_id}.log
  - 每 Tool 调用独立文件
  - 头部: tool 名 + 参数 (JSON) + 开始时间
  - 尾部: 结果 (JSON,截断) + 耗时 + 错误
  - 文件权限 0600 (owner only,含敏感参数)
- API:
```
package logging

func New(cfg LogConfig) (*slog.Logger, func(), error)
func StartToolCall(ctx context.Context, base *slog.Logger, toolName string, params any) (ToolCallLogger, error)

type ToolCallLogger interface {
    WithResult(result any)
    WithError(err error)
    End()
}
```
- 启动时 os.MkdirAll(LogDir/tools, 0700) 确保目录存在(权限收紧)
验收: go build 通过,启动时输出配置,data 目录自动创建,终端看 text 日志 + 主文件 JSON 日志 + 调一个 Tool 后看 per-tool 文件存在

Executor: Codex

Phase 2: 存储层 (SQLite) (1.5h)

目标: SQLite 存储层,支持 clusters 表的 CRUD。Cluster 表新增 spark-submit 路径字段。

T2.1 internal/storage/sqlite.go

Open(path string) (*sql.DB, error)
自动应用 schema migration

clusters 表 (v0.3 + v0.3.2 + v0.3.5 + v0.3.6 + v0.3.7):

CREATE TABLE clusters (
  id                        TEXT PRIMARY KEY,
  name                      TEXT NOT NULL,
  rm_url                    TEXT NOT NULL,
  shs_url                   TEXT NOT NULL,
  spark_submit_execute_bin  TEXT NOT NULL,
  is_active                 INTEGER NOT NULL DEFAULT 1,
  -- v0.3.6 鉴权 (去掉 kerberos,用户环境用 simple)
  auth_type                 TEXT NOT NULL DEFAULT 'none',  -- 'none' | 'simple' | 'basic'
  auth_username             TEXT,                          -- simple (user.name) / basic
  auth_password_enc         BLOB,                          -- basic,加密存储 (AES-GCM);PUT 留空=保留
  -- v0.3.5 SSL
  ssl_verify                INTEGER NOT NULL DEFAULT 1,
  ssl_ca_bundle             TEXT,
  -- v0.3.5 URL allowlist (SHS fetch_url 用)
  url_allowlist             TEXT,                          -- JSON 数组,glob 模式
  -- v0.3.7 新增
  default_submit_args       TEXT,                          -- JSON 数组,spark_submit 时自动 prepend
  rate_limit_per_min        INTEGER NOT NULL DEFAULT 10,   -- 0=不限,默认 10 次/分钟
  created_at                INTEGER NOT NULL,
  updated_at                INTEGER NOT NULL
);

audit_log 表 (v0.3.7 新增,所有 admin 写操作留痕):

CREATE TABLE audit_log (
  id          INTEGER PRIMARY KEY AUTOINCREMENT,
  timestamp   INTEGER NOT NULL,
  actor       TEXT NOT NULL,                -- "admin:<token_name>"(多 token 支持)
  action      TEXT NOT NULL,                -- "cluster.create" | "cluster.update" | "cluster.delete"
  cluster_id  TEXT,                         -- NULL 表示不针对 cluster
  details     TEXT                          -- JSON,操作前后关键字段 diff
);
CREATE INDEX idx_audit_ts ON audit_log(timestamp DESC);

T2.2 internal/cluster/types.go:

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"`
    // v0.3.5 鉴权 (v0.3.6 简化:删 kerberos 字段)
    AuthType              string   `json:"auth_type"`              // "none" | "simple" | "basic"
    AuthUsername          string   `json:"auth_username,omitempty"`  // simple (user.name) / basic
    AuthPassword          string   `json:"-"`                        // basic only, 永不出网
    // (Kerberos 字段 v2 才有: AuthPrincipal / AuthKeytabPath)
    // v0.3.5 SSL
    SSLVerify             bool     `json:"ssl_verify"`
    SSLCABundle           string   `json:"ssl_ca_bundle,omitempty"`
    // v0.3.5 URL allowlist
    URLAllowlist          []string `json:"url_allowlist,omitempty"`
    // v0.3.7 新增
    DefaultSubmitArgs     []string `json:"default_submit_args,omitempty"` // spark_submit 自动 prepend
    RateLimitPerMin       int      `json:"rate_limit_per_min"`            // 0=不限,默认 10
    CreatedAt             int64    `json:"created_at"`
    UpdatedAt             int64    `json:"updated_at"`
}

LLM 通过 list_clusters 拿到完整字段 — RM/SHS endpoint + 鉴权 + allowlist + 默认 args 都是发现内容
AuthPassword 字段 JSON tag 是 "-" 永不出网,DB 里存加密 blob
PUT 留空 = 保留旧密码(避免 admin 端不知道现有密码时清空)

T2.3 internal/storage/cluster_repo.go: 完整 CRUD + ErrNotFound 哨兵
T2.4 单元测试 cluster_repo_test.go 用 :memory: DB 测全路径
验收: go test ./internal/storage/... 全绿

Executor: Codex

Phase 3: 共享 HTTP 客户端 + SSRF 防护 (1.5h)

目标: 一个共享的 HTTP 客户端,供 fetch_url 和 yarn_job_manage 用。

T3.1 internal/httpclient/client.go
- 基于 net/http 包装
- Timeout (默认 30s), MaxResponseBytes (默认 1MB)
- Do(ctx, req) (*http.Response, error),body 已被 truncate
T3.2 internal/httpclient/ssrf.go
- CheckURL(urlStr, allowedHosts) error
- 私网 IP 黑名单 (IPv4 + IPv6)
- DNS rebinding 防护: 解析时锁定 IP,后续 dial 用 IP
- allowlist 机制: cluster URL 的 host 自动加入 allowlist (解决 fetch_url 抓 SHS/RM 的需求)
T3.3 单元测试: 私网拦截、allowlist 豁免、scheme 校验、超大 body 截断
T3.4 手动 redirect following (v0.3.5 新):
- Go net/http 默认跨主机 redirect 会丢 Authorization header
- 自实现 CheckRedirect 把原始 Authorization 重新塞到 redirect 后的 request
- 最多 3 次 redirect 后停止
- 关键场景: get_application_logs 降级路径中,RM→NM 跨主机 307 redirect 必须保留 auth
T3.5 鉴权层 (v0.3.5 新, v0.3.6 简化):
- internal/httpclient/auth.go 定义 AuthConfig 接口
- none: 不加任何 header
- simple: YARN SimpleAuth(用户环境采用) — 加 user.name query param(默认 "yarn",可配)
- basic: req.SetBasicAuth(user, pass)
- ~~kerberos~~: v1 不实现(用户环境不需要,集成 gokrb5 是 v2)
- applyAuth(req *http.Request) error 统一入口
T3.6 SSL 配置 (v0.3.5 新):
- internal/httpclient/ssl.go 构造 *http.Transport
- SSLVerify=false → InsecureSkipVerify=true (DANGER,仅测试)
- SSLCABundle 非空 → 加载 CA cert 到 RootCAs
验收: go test ./internal/httpclient/... 全绿

Executor: Codex

Phase 4: Cluster 管理 (Admin API) (2h)

目标: /admin/clusters 完整 CRUD,新增 spark-submit 路径字段校验。

T4.1 internal/middleware/auth.go (v0.3.7 多 token)
- RequireAnyToken(tokens []string) gin.HandlerFunc — 任一 token 匹配即通过,否则 401
- 记录使用的 token 标识 (hash 前 8 位) 到 c.Set("admin_token_id", ...) 供 audit_log 用
T4.2 internal/cluster/handler.go
- GET/POST/PUT/DELETE /admin/clusters[/:id]
- 校验:
  - rm_url / shs_url 必须 http:// 或 https:// 开头
  - spark_submit_execute_bin 必须绝对路径(filepath.IsAbs)
  - 文件存在(os.Stat 通过)
  - 文件可执行(mode&0111 != 0)
  - id 非空
- 校验失败 → 400 + 错误详情
T4.3 main.go 注册 admin route group + audit 拦截器
T4.4 audit log 中间件 (v0.3.7): 任何 POST/PUT/DELETE 写一条 audit_log,包含 actor + action + cluster_id + details
T4.5 速率限制中间件 (v0.3.7): per-cluster,基于 golang.org/x/time/rate,spark_submit Tool 调用前检查 cluster.RateLimitPerMin
T4.6 集成测试: 完整 CRUD + 鉴权 + 路径校验失败 + audit_log 写入 + 速率限制触发
验收:
- 缺 Authorization → 401
- spark_submit_execute_bin 是不存在的文件 → 400
- 正常 cluster 配置 → 200
- admin 操作后 audit_log 表有对应记录
- spark_submit 超过 10/min → 429

Executor: Codex

Phase 5: MCP Tools 实现 (10-12h, v0.3.5 加 4 个 RM Tool + 鉴权层)

目标: 11 个 Tool 的业务逻辑实现,与 MCP 协议解耦。共享鉴权层(由 T3.5 实现)。

T5.1 `list_clusters`

internal/tools/list_clusters.go
签名: ListClusters(ctx, repo) ([]Cluster, error)
直接调 repo.List

T5.2 `list_applications` (RM, v0.3.5 新) ⭐

internal/tools/list_applications.go
端点: GET {rm_url}/ws/v1/cluster/apps

Params:

type ListApplicationsParams struct {
    ClusterID string `json:"cluster_id"`
    State     string `json:"state,omitempty"` // "RUNNING" | "FINISHED" | "FAILED" | "KILLED" | "ACCEPTED" 等
    Queue     string `json:"queue,omitempty"`
    Limit     int    `json:"limit,omitempty"` // 默认 100
}

流程: 查 cluster → 应用鉴权 → 拼 URL with query string → 调 doAuthenticatedRequest
单元测试: 各种 state / queue / limit 组合;cluster inactive → 错误

T5.3 `get_application_status` (RM, v0.3.5 新) ⭐

internal/tools/get_application_status.go
端点: GET {rm_url}/ws/v1/cluster/apps/{app_id}
Params: ClusterID, AppID
返回: 完整 app JSON(含 state, finalStatus, amContainerLogs URL 字段)
关键: amContainerLogs 字段是 T5.4 降级路径的入口

T5.4 `get_application_logs` (RM, v0.3.5 新,核心) ⭐

internal/tools/get_application_logs.go
降级链 (用户参考实现关键,3 步):
1. 首选: GET {rm_url}/ws/v1/cluster/apps/{app_id}/aggregated-logs
2. 降级 1: 如果返回 404/501,改调 get_application_status 拿 amContainerLogs 字段
3. 降级 2: amContainerLogs 指向 NM (NodeManager),手动跟随 3xx Location(最多 3 次)
4. 降级 3: 最终拿到 AM driver 的 stdout

Params:

type GetApplicationLogsParams struct {
    ClusterID string `json:"cluster_id"`
    AppID     string `json:"app_id"`
    TailLines int    `json:"tail_lines,omitempty"` // 默认 2000
    Filter    string `json:"filter,omitempty"`     // "ERROR" | "WARN" | "Exception" | "all"(默认)
}

关键鉴权: 手动跟随 redirect 时必须保留 Authorization header。Go net/http 默认跨主机 redirect 会丢 header,需自实现 CheckRedirect 把 Authorization 重新塞到 redirect 后的 request 上

T5.5 `kill_application` (RM, v0.3.5 新)

internal/tools/kill_application.go
端点: PUT {rm_url}/ws/v1/cluster/apps/{app_id}/state
Body: {"state": "KILLED"}
Params: ClusterID, AppID
单元测试: 用 httptest mock RM,验证 PUT 请求体和 URL 正确

T5.6 `fetch_url` (SSRF + 鉴权,关键路径)

internal/tools/fetch_url.go
Params: URL (必填), Method (默认 GET), Headers (可选), MaxBytes (可选)
allowed_hosts 由调用方(LLM via list_clusters 取到的 host)注入
流程: SSRF 检查 → 调 client → 截断 body → 返回 {status, headers, body, truncated}

T5.7 `upload_file` (本地文件存储)

internal/tools/upload_file.go

Params:

type UploadFileParams struct {
    Path     string `json:"path"`               // 相对 data dir,如 "scripts/main.py"
    Content  string `json:"content"`            // 文本或 base64
    Encoding string `json:"encoding,omitempty"` // "utf-8" (默认) | "base64"
}

流程:
1. 路径白名单校验: filepath.Clean(params.Path) 不能 .. 跳出 data_dir
2. 创建父目录 os.MkdirAll
3. 解码 content (utf-8 直接写,base64 先 decode)
4. 写文件 (mode 0644)
5. 返回 {path: <绝对路径>, size}
单元测试: 路径跳出 (../etc/passwd) 应被拒;二进制 base64 解码正确;大文件 (>1MB) 写成功

T5.8 `spark_submit` (核心) ⭐

internal/executor/exec.go
- 封装 exec.CommandContext + 输出捕获
- 超时控制、退出码、stdout/stderr 截断(默认 10MB 截断防内存爆)
- 永不调用 bash -c / sh -c
internal/tools/spark_submit.go

Params (v0.3.1 简化: 移除 Submitter):

type SparkSubmitParams struct {
    ClusterID      string   `json:"cluster_id"`
    Args           []string `json:"args"`                // binary 参数列表
    TimeoutSeconds int      `json:"timeout_seconds,omitempty"` // 默认 60
}

流程:
1. 查 cluster → 校验存在 + IsActive
2. 取 cluster.SparkSubmitExecuteBin 作为 binary (cluster 配置决定,LLM 不能改)
3. 校验 path 存在 + 可执行 (T4.2 已做,但 runtime 再 check 一次)
4. 关键: exec.CommandContext(ctx, binary, params.Args...) — 切片传参,绝不走 shell
5. 捕获 stdout/stderr
6. 解析 app_id — YARN 提交成功输出 "Submitted application application_xxx_yyy",用 regex application_\d+_\d+ 提取
7. 返回:
```
type SparkSubmitResult struct {
    Command  string `json:"command"`   // 完整命令 (for audit,不含二进制路径)
    Stdout   string `json:"stdout"`    // 截断
    Stderr   string `json:"stderr"`
    ExitCode int    `json:"exit_code"`
    AppID    string `json:"app_id,omitempty"`
}
```
单元测试:
- 用 fake binary (Go test helper) 验证 args 切片正确传递
- 超时场景
- app_id 解析 regex
- 失败: cluster 不存在、cluster inactive、binary 不存在、binary 不可执行

T5.9 `fetch_spark_metrics` (高层 SHS 封装) ⭐

internal/tools/fetch_spark_metrics.go
本质: 内部调用 fetch_url,把 SHS URL 拼好

Params:

type FetchSparkMetricsParams struct {
    ClusterID string `json:"cluster_id"`
    AppID     string `json:"app_id"`
    Path      string `json:"path,omitempty"` // 默认 "jobs",可选 "stages"/"executors"/"environment"
}

流程:
1. 查 cluster → 校验存在 + IsActive,拿 shs_url
2. 拼 URL: {shs_url}/api/v1/applications/{app_id}/{path}
3. 内部调用 FetchURL (T5.3 实现) — 不重新走 SSRF (cluster URL 已经在 allowlist)
4. 返回 {url, status, body, truncated}
单元测试:
- 用 httptest mock SHS,验证 URL 拼接正确
- cluster 不存在 → 错误
- cluster inactive → 错误
- 默认 path = "jobs"

T5.10 `fetch_cluster_env` (高层 RM 封装) ⭐

internal/tools/fetch_cluster_env.go
本质: 内部调用 fetch_url,把 RM URL 拼好

Params:

type FetchClusterEnvParams struct {
    ClusterID string `json:"cluster_id"`
    // 可选: "nodes" (默认) | "metrics" | "apps"
}

流程:
1. 查 cluster → 校验存在 + IsActive,拿 rm_url
2. 拼 URL: {rm_url}/ws/v1/cluster/{section} (默认 nodes)
3. 内部调用 FetchURL (T5.3 实现)
4. 返回 {url, status, body, truncated}
单元测试:
- 用 httptest mock RM,验证 URL 拼接正确
- 默认 section = "nodes"
- cluster 不存在 / inactive → 错误

T5.11 `analyze_spark_log` (规则引擎 + RM 日志降级) ⭐

internal/analyzer/rules.go + engine.go
internal/tools/analyze_spark_log.go
指标部分: 内部调 fetch_spark_metrics (T5.9) 拿 stages / executors / jobs
规则部分: data_skew / gc_pressure / bottleneck (硬编码阈值,接口预留可配)
日志部分 (v0.3.5 修正): 不走 SHS logs,改调 get_application_logs (T5.4) 走 RM 降级链
- 仅当 findings 非空时拉取,避免无谓请求
Prompt 部分: 拼 LLM-Ready 文本
单元测试: 各种 findings 场景 × 降级链组合

T5.12 综合单元测试

每个 Tool 至少 1 正向 + 1 边界用例
鉴权层单测 (none / basic / kerberos stub)
降级链单测 (aggregated-logs 404 → amContainerLogs → NM)
per-tool 日志单测 (v0.3.8): mock Tool 调用,验证 per-tool 文件生成 + 内容包含 params/result
go test ./internal/tools/... 全绿

T5.13 每个 Tool 集成 `logging.StartToolCall` (v0.3.8) ⭐

11 个 Tool 全部接入

Tool 函数入口:

callLog, _ := logging.StartToolCall(ctx, logger, "list_clusters", params)
defer callLog.End()

成功路径: callLog.WithResult(result)
失败路径: callLog.WithError(err)
per-tool 文件记录原始 params,便于 LLM 行为调试

Executor: Codex

Phase 6: MCP Server 装配 (2h)

目标: 把 5 个 Tool 注册到 mcp-go Server,挂上 Streamable HTTP handler。

T6.1 查 mark3labs/mcp-go v0.56.0 真实 API
- server.NewMCPServer(name, version, opts...)
- mcp.NewTool(name, opts...)
- mcp.WithString / mcp.WithNumber / mcp.WithArray / mcp.WithObject
- server.NewStreamableHTTPServer(s, opts...)
- 先查源码再写
T6.2 internal/mcp/server.go: NewServer() *server.MCPServer
T6.3 internal/mcp/register.go: RegisterTools(s, deps)
- 注册 5 个 Tool,handler 调 Phase 5 业务函数
- Tool input schema 严格声明 (LLM 调错就报错)
T6.4 internal/mcp/streamable.go: Streamable HTTP handler
- main.go 用 gin.WrapH 挂到 /mcp,加 auth 中间件
T6.5 集成测试: httptest + JSON-RPC 构造 tools/call 请求
验收:
- go build ./... 通过
- curl -X POST http://localhost:8080/mcp -d '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"list_clusters",...}}' 返回 MCP 响应

Executor: Codex

Phase 7: 端到端验证 (1h)

目标: 启动 server,跑通完整流程。

T7.1 go build ./... + go vet ./... 无 warning
T7.2 启动 server,确认 /healthz 200
T7.3 curl 跑 admin CRUD(创建/列出/更新/删除 cluster)
T7.4 用 curl 发 MCP tools/call:
- list_clusters → 返回 cluster 列表
- upload_file 上传 main.py → 返回路径
- spark_submit (用 fake spark-submit binary,test 模式) → 返回 exit_code + (mock) app_id
- yarn_job_manage 用 mock rm_url → 验证 status / kill 路径
- fetch_url 抓公网 URL → 200
- fetch_url 抓 127.0.0.1 → 被 SSRF 拦截
- upload_file path = "../etc/passwd" → 路径校验拦截
T7.5 go test ./... 全绿
验收: 全部通过

Executor: CC 验证

风险与依赖

外部依赖

mark3labs/mcp-go v0.56.0 API 稳定性: Phase 6.1 需要查真实 API
modernc.org/sqlite: 纯 Go,无外部依赖

数据结构风险

applications 表当前只为后续阶段预留,如果不需要缓存可删除
Cluster 配置变更频率极低,SQLite 完全够用

安全风险 (必须处理)

风险	严重度	缓解
命令注入 (spark_submit)	高	`exec.Command(name, args...)` 切片形式,绝不走 shell。binary 由 cluster 配置固定 (v0.3.1),LLM 不能在 Tool 调用中改 binary 路径。LLM 提供的 args 直接作为 binary 的 argv,无字符串拼接
SSRF (fetch_url)	高	私网 IP 黑名单 + cluster URL allowlist + DNS rebinding 锁定
任意文件读/写 (upload_file)	中	路径白名单 `./data/`,拒绝 `..`、绝对路径、符号链接逃逸
Token 泄露	中	必须从 env 读,启动日志隐藏 token 字段
内存爆炸 (大响应/大日志)	中	HTTP body 1MB 截断,spark-submit stdout/stderr 10MB 截断,analyze_spark_log 日志 tail 2000 行
慢请求 DoS	中	HTTP client 30s timeout,spark-submit 默认 60s timeout,analyze_spark_log 并行 3 个 SHS fetch (受同一 timeout 约束)
spark-submit binary 被替换	低	启动时 + runtime 都校验 binary 存在 + 可执行 + 绝对路径
集群配置被改	低	audit log 记录 cluster CRUD 操作 (可选,本阶段不强制)
启发式规则误报 (analyze_spark_log)	中	中
SHS API 字段变动 (analyze_spark_log)	中	中
Hadoop 2.7 兼容性 (v0.3.5 → v0.3.6 已验证)	低	低
manual redirect 误用 (v0.3.5)	中	高

决策日志

日期	决策	理由
2026-07-10	存储用 SQLite 而非 MongoDB	单持久化层,消除 special case
2026-07-10	暴露 `fetch_url` (通用 HTTP 原语)	LLM 需要通用 HTTP 能力
2026-07-10	MCP 传输用 Streamable HTTP	现代 MCP 标准
2026-07-10	HTTP 客户端独立包,所有 Tool 复用	SSRF 防护集中
2026-07-10	SQLite 驱动选 `modernc.org/sqlite`	纯 Go,无 CGO
2026-07-10 (v0.2→v0.3)	加 `spark_submit` Tool	用户从"纯读"改为"读+本地提交",业务上需要
2026-07-10 (v0.2→v0.3)	`upload_file` 改为本地文件存储 (非 HTTP 上传)	用途明确为 spark-submit 脚本上传,本地存储更简单安全
2026-07-10 (v0.2→v0.3)	spark-submit 在 Server 本地执行 (无 SSH)	用户明示"使用 spark-submit shell命令直接提交"
2026-07-10 (v0.2→v0.3)	Cluster 配置加单一 `spark_submit_execute_bin` (v0.3.1 合并 spark_submit_path + spark2_submit_path)	用户明示两个 binary 都要可配;后简化合并为单字段,LLM 不必选 binary
2026-07-10 (v0.2→v0.3)	`spark_submit` 用 `exec.Command` 切片传参	绝不走 shell,消除命令注入面
2026-07-10 (v0.3.1→v0.3.2)	`list_clusters` 标为强制发现入口,Cluster 加 `is_active` 字段	LLM 必须能拿到 rm_url/shs_url/spark_submit_execute_bin 才能用其他 Tool。`is_active` 允许禁用 cluster 而不删除
2026-07-10 (v0.3.2→v0.3.3)	加回 `fetch_spark_metrics` + `fetch_cluster_env` 作为高层 Tool,内部调用 `fetch_url`	业务常用端点封装,LLM 不必拼 SHS/RM URL,但仍保留 `fetch_url` 原语处理长尾查询
2026-07-10 (v0.3.3→v0.3.4)	加回 `analyze_spark_log` + 服务端启发式过滤 (data_skew / gc_pressure / bottleneck) + LLM-Ready Prompt	用户原 spec 明确要求 [Log Analyzer + Regex Rules + LLM-Ready Prompt] 组件。规则阈值硬编码,接口预留可配
2026-07-10 (v0.3.4→v0.3.5)	拆 `yarn_job_manage` 成 4 个 RM Tool:list_applications / get_application_status / get_application_logs / kill_application	用户给出参考实现(spark_executor 4 个独立 Python 入口),MCP 端按 1 Tool = 1 端点拆,更符合 MCP 习惯
2026-07-10 (v0.3.4→v0.3.5)	加鉴权层 (none / basic / kerberos) + SSL 配置 + manual redirect following	集群环境 Spark on YARN + Hadoop 2.7,生产环境需要 SPNEGO。`get_application_logs` RM→NM 降级路径必须保留 Authorization 跨主机 redirect
2026-07-10 (v0.3.4→v0.3.5)	`analyze_spark_log` 日志部分改走 RM `get_application_logs` 降级链 (不是 SHS)	SHS logs 端点不稳定,RM aggregated-logs + amContainerLogs + NM 是生产链路
2026-07-10 (v0.3.5→v0.3.6)	鉴权层简化:`AuthType` 改为 `none / simple / basic`,Kerberos 推到 v2	用户告知集群用 YARN SimpleAuth(`user.name` query param),不需 SPNEGO。v1 简化够用,生产环境部署可上
2026-07-10 (v0.3.5→v0.3.6)	Hadoop 2.7 兼容性"已验证",从风险表移除	用户实测 4 个 RM 端点 + SHS 端点都通,无需 Phase 7 集群验证
2026-07-10 (v0.3.6→v0.3.7)	加 per-cluster `default_submit_args` + `rate_limit_per_min`	减少 LLM 拼 args 出错,防误调刷爆集群
2026-07-10 (v0.3.6→v0.3.7)	删 `applications` 表	纯 RM/SHS 读模式不缓存 app 状态
2026-07-10 (v0.3.6→v0.3.7)	`auth_password` PUT 留空 = 保留	admin 端不需要知道现有密码
2026-07-10 (v0.3.6→v0.3.7)	加 `audit_log` 表 + 路由	留痕 cluster CRUD 操作
2026-07-10 (v0.3.6→v0.3.7)	多 `AdminTokens` 支持 (env 逗号分隔)	多人共用一个 token 不安全,改用多 token 配任意一个通过
2026-07-10 (v0.3.6→v0.3.7)	analyzer 阈值改为 env 可配 (`ANALYZER_*`)	不同集群阈值可能不同,硬编码不灵活
2026-07-10 (v0.3.7→v0.3.8)	用 `log/slog` 替换默认 log (Go 1.21+ stdlib)	用户要求结构化日志
2026-07-10 (v0.3.7→v0.3.8)	双输出:终端 text + 主文件 JSON	用户要求终端可读 + 文件持久化
2026-07-10 (v0.3.7→v0.3.8)	Per-Tool 独立日志文件 (`data/logs/tools/{ts}_{tool}_{id}.log`)	用户要求每次工具调用单独记录 tool 名 + 参数 + 结果 + 耗时,便于调试 LLM 行为

待确认项 (执行中可继续澄清)

(已清空 — v0.3.7 用户接受所有 6 项建议)

/mcp 路由前缀: /mcp 还是 /mcp/*?
Token 配置: ADMIN_TOKEN 和 AGENT_TOKEN 是否必须分两个?还是合并?
applications 表 (v0.3.7 已删): 纯 RM/SHS 读模式不需要缓存 app 状态
spark-submit 路径校验强度: T4.2 已校验存在+可执行,是否还要校验 owner / checksum 防篡改?(本阶段不做)
app_id 解析 regex: 不同 YARN 版本输出格式可能不同,需保留 regex 可配置
Metrics / Tracing: 是否需要集成 Prometheus?当前不包含

执行约定

每个 Phase 结束: go build ./... + go test ./... 全绿才能进入下一 Phase
代码生成: 全部走 Codex MCP,使用 model: "kimi/kimi-k2.7-code", sandbox: "danger-full-access", approval-policy: "on-failure"
API 变更: 任何 import 路径或公开签名变更,先在 task_plan.md 标注再改
测试覆盖: 每个 Tool 至少 1 正向 + 1 边界;auth 中间件必须有 401 测试;exec.Command 必须切片传参(代码 review 必查)

进度

Phase 0 - 清理与依赖调整
Phase 1 - 配置层
Phase 2 - 存储层
Phase 3 - 共享 HTTP 客户端 + SSRF 防护
Phase 4 - Cluster 管理 (Admin API)
Phase 5 - MCP Tools 实现 (5 个 Tool)
Phase 6 - MCP Server 装配
Phase 7 - 端到端验证

35 KiB Raw Blame History

Spark MCP Server 实现计划

目标

已确认的架构决策

架构总览

Tool 矩阵 (v0.3 最终方案)

典型工作流 (LLM 视角)

路由设计

项目目录结构

阶段划分

Phase 0: 清理与依赖调整 (0.5h)

Phase 1: 配置层 (0.5h)

Phase 2: 存储层 (SQLite) (1.5h)

Phase 3: 共享 HTTP 客户端 + SSRF 防护 (1.5h)

Phase 4: Cluster 管理 (Admin API) (2h)

Phase 5: MCP Tools 实现 (10-12h, v0.3.5 加 4 个 RM Tool + 鉴权层)

T5.1 list_clusters

T5.2 list_applications (RM, v0.3.5 新) ⭐

T5.3 get_application_status (RM, v0.3.5 新) ⭐

T5.4 get_application_logs (RM, v0.3.5 新,核心) ⭐

T5.5 kill_application (RM, v0.3.5 新)

T5.6 fetch_url (SSRF + 鉴权,关键路径)

T5.7 upload_file (本地文件存储)

T5.8 spark_submit (核心) ⭐

T5.9 fetch_spark_metrics (高层 SHS 封装) ⭐

T5.10 fetch_cluster_env (高层 RM 封装) ⭐

T5.11 analyze_spark_log (规则引擎 + RM 日志降级) ⭐

T5.12 综合单元测试

T5.13 每个 Tool 集成 logging.StartToolCall (v0.3.8) ⭐

Phase 6: MCP Server 装配 (2h)

Phase 7: 端到端验证 (1h)

风险与依赖

外部依赖

数据结构风险

安全风险 (必须处理)

决策日志

待确认项 (执行中可继续澄清)

执行约定

进度

35 KiB

Raw Blame History

T5.1 `list_clusters`

T5.2 `list_applications` (RM, v0.3.5 新) ⭐

T5.3 `get_application_status` (RM, v0.3.5 新) ⭐

T5.4 `get_application_logs` (RM, v0.3.5 新,核心) ⭐

T5.5 `kill_application` (RM, v0.3.5 新)

T5.6 `fetch_url` (SSRF + 鉴权,关键路径)

T5.7 `upload_file` (本地文件存储)

T5.8 `spark_submit` (核心) ⭐

T5.9 `fetch_spark_metrics` (高层 SHS 封装) ⭐

T5.10 `fetch_cluster_env` (高层 RM 封装) ⭐

T5.11 `analyze_spark_log` (规则引擎 + RM 日志降级) ⭐

T5.13 每个 Tool 集成 `logging.StartToolCall` (v0.3.8) ⭐