tao.chen/spark-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity
Files
main
spark-mcp/task_plan.md
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

701 lines
35 KiB
Markdown
Raw Permalink 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**:
```go
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):
```sql
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 写操作留痕):
```sql
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`:
```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:
```go
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:
```go
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:
```go
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):
```go
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. 返回:
```go
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:
```go
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:
```go
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 函数入口:
```go
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) | 中 | 中 | 规则是 first-pass filter,不是真理。LLM 拿到 findings 后仍需自己判断。Prompt 明确说"请基于以上信息生成根因分析报告" |
| **SHS API 字段变动** (analyze_spark_log) | 中 | 中 | metrics JSON 解析失败时,跳过对应规则不出 finding,不让一个 Tool 崩整个调用 |
| **Hadoop 2.7 兼容性** (v0.3.5 → v0.3.6 已验证) | 低 | 低 | **用户已验证 endpoint 在 2.7 集群上工作**。RM REST API 4 个端点 + SHS 端点都测过 |
| **manual redirect 误用** (v0.3.5) | 中 | 高 | 自实现 `CheckRedirect` 必须严格保留 Authorization,否则 RM→NM 降级会 401/403。代码 review 必查 |
---
## 决策日志
| 日期 | 决策 | 理由 |
|---|---|---|
| 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 必查)
---
## 进度
- [x] 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 - 端到端验证
Reference in New Issue View Git Blame Copy Permalink