diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..9829512 --- /dev/null +++ b/.env.example @@ -0,0 +1,37 @@ +# spark-mcp-go environment configuration +# Copy to .env and edit the *REQUIRED* values. + +# HTTP listen address (default: :8080) +LISTEN_ADDR=:8080 + +# Root directory for SQLite, uploads, and logs (default: ./data) +DATA_DIR=./data + +# *REQUIRED* Comma-separated admin bearer tokens for /admin/* endpoints +ADMIN_TOKENS=change-me-admin-token-1,change-me-admin-token-2 + +# *REQUIRED* Agent bearer token for /mcp endpoints +AGENT_TOKEN=change-me-agent-token + +# Timeout for fetch_url and RM tool HTTP calls (default: 30s) +HTTP_CLIENT_TIMEOUT=30s + +# Maximum HTTP response bytes before truncation (default: 1048576) +MAX_RESPONSE_BYTES=1048576 + +# Timeout for spark-submit process execution (default: 60s) +SPARK_SUBMIT_TIMEOUT=60s + +# Log directory (default: ./data/logs) +LOG_DIR=./data/logs + +# Log level: debug, info, warn, error (default: info) +LOG_LEVEL=info + +# Log format: text (terminal) or json (log file; default: text) +LOG_FORMAT=text + +# Analyzer heuristic thresholds (defaults shown) +ANALYZER_DATA_SKEW_RATIO=3.0 +ANALYZER_GC_PRESSURE_RATIO=0.1 +ANALYZER_BOTTLENECK_SHUFFLE_GB=50 diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 0000000..9eeb61d --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,63 @@ +# spark-mcp-go 架构说明 + +## 包依赖图 + +``` +main + ├─ config + ├─ logging + ├─ storage (SQLite) + ├─ admin + │ └─ storage / cluster / audit + ├─ mcp + │ ├─ tools + │ │ ├─ rm + │ │ ├─ httpclient + │ │ ├─ analyzer + │ │ └─ storage / cluster + │ └─ server (mark3labs/mcp-go) + └─ middleware +``` + +- `admin` 只依赖 `storage`、`cluster`、`audit`、`middleware`。 +- `tools` 依赖 `rm`(ResourceManager 客户端)、`httpclient`(通用 HTTP 原语)、`analyzer`(日志分析)。 +- `mcp` 聚合所有 Tools,注册到 `mark3labs/mcp-go` 的 Streamable HTTP server。 +- `main` 负责加载配置、打开数据库、挂载路由、启动 HTTP 服务。 + +## 数据流 + +1. 管理员调用 `/admin/clusters` POST,写入 cluster JSON。 +2. `storage` 把 cluster 持久化到 SQLite(后续会对敏感字段加密)。 +3. LLM Agent 调用 `/mcp` Tool(如 `list_clusters`)。 +4. Tool 从 storage 读取 cluster,携带 auth 信息,通过 `httpclient` 或本地 `spark-submit` 调用 YARN/Spark。 +5. 结果返回给 LLM;每次 Tool 调用同时写入独立审计日志。 + +## 11 Tools 分类 + +| 分类 | Tools | +|---|---| +| discovery | `list_clusters` | +| exec | `spark_submit` | +| RM | `list_applications`, `get_application_status`, `get_application_logs`, `kill_application`, `fetch_cluster_env` | +| SHS | `fetch_spark_metrics` | +| analyzer | `analyze_spark_log` | +| 原语 | `fetch_url`, `upload_file` | + +## 关键设计决策 + +- **slice-form `spark_submit`**:参数以字符串数组传入,避免 shell 拼接和注入。 +- **cluster `url_allowlist`**:所有外发 HTTP(`fetch_url`、RM、SHS)必须匹配 allowlist,防止 SSRF。 +- **日志降级链**:`get_application_logs` 依次尝试 `amContainerLogs`、aggregated-logs、通用 logs 端点。 +- **多 token admin**:`ADMIN_TOKENS` 支持逗号分隔多管理员 token,便于轮换。 +- **per-tool 独立审计日志**:每个 Tool 每次调用写入 `${LOG_DIR}/tools/.log`,文件权限 `0600`。 +- **`auth_password` 不可通过 JSON 创建**:字段带 `json:"-"` tag,避免接口泄露密码;生产通过专用端点或数据库初始化写入。 + +## 安全模型(三道防线) + +1. **Token 鉴权**:`ADMIN_TOKENS` 保护管理面,`AGENT_TOKEN` 保护 MCP 面。 +2. **SSRF 防护**:`fetch_url` 与所有集群请求都走 cluster allowlist,且内置 DNS rebinding 检查。 +3. **路径/参数白名单**:`spark_submit` 仅允许 jar、class、--conf 等安全参数;禁止 shell 元字符与重定向。 + +## 运行与部署 + +详见 [docs/runbook.md](docs/runbook.md)。 diff --git a/README.md b/README.md new file mode 100644 index 0000000..15ac266 --- /dev/null +++ b/README.md @@ -0,0 +1,139 @@ +# spark-mcp-go + +MCP (Model Context Protocol) server for Apache Spark on YARN. Lets an LLM agent +discover Spark/YARN endpoints, submit jobs, fetch logs, and analyze them via +11 typed Tools. Streamable HTTP transport, SQLite-backed configuration, and +`log/slog` structured logging with per-Tool call files. + +## What it does + +- Exposes 11 MCP Tools over Streamable HTTP at `/mcp` +- Admin API at `/admin/*` for cluster and audit configuration +- HTTP Basic and YARN SimpleAuth for Spark/YARN endpoints +- SSRF protection with cluster URL allowlist plus DNS rebinding guard +- Per-Tool independent audit log (file mode `0600`) +- `spark-submit` via local `exec.Command` (no shell, no injection) + +## Quick start (3 steps) + +1. Copy and edit the environment file: + ```bash + cp .env.example .env + # Edit ADMIN_TOKENS and AGENT_TOKEN + ``` + +2. Build and start the server: + ```bash + go build ./... + ./spark-mcp-go + # Or use the helper: + # ./scripts/dev.sh + ``` + +3. Initialize an MCP session, then call `list_clusters` to discover endpoints. + +## 11 Tools + +| Tool | Type | Purpose | +|---|---|---| +| `list_clusters` | discovery | Return all active clusters (LLM entry point) | +| `spark_submit` | exec | Local `spark-submit` process, extracts `app_id` | +| `list_applications` | RM | `GET /ws/v1/cluster/apps[?state=&user=]` | +| `get_application_status` | RM | `GET /ws/v1/cluster/apps/{id}` | +| `get_application_logs` | RM | Fallback chain: `amContainerLogs` -> aggregated-logs -> logs | +| `kill_application` | RM | `PUT` state `KILLED` | +| `fetch_spark_metrics` | SHS | Executor metrics + summary mode triggers analyzer | +| `fetch_cluster_env` | RM | Aggregate `/cluster/info` and `/cluster/metrics` | +| `analyze_spark_log` | analyzer | RM logs + heuristic rules + LLM-ready prompt | +| `fetch_url` | HTTP | Generic primitive, uses cluster allowlist and auth | +| `upload_file` | FS | Write to `./data/uploads/`, reference from `spark_submit` | + +## Environment variables + +| Variable | Default | Required | Description | +|---|---|---|---| +| `LISTEN_ADDR` | `:8080` | no | HTTP listen address | +| `DATA_DIR` | `./data` | no | SQLite, uploads, and log root | +| `ADMIN_TOKENS` | — | **yes** | Comma-separated tokens for `/admin/*` | +| `AGENT_TOKEN` | — | **yes** | Token for `/mcp` requests | +| `HTTP_CLIENT_TIMEOUT` | `30s` | no | Timeout for `fetch_url`/RM Tool HTTP calls | +| `MAX_RESPONSE_BYTES` | `1048576` (1 MiB) | no | HTTP response truncation limit | +| `SPARK_SUBMIT_TIMEOUT` | `60s` | no | `spark-submit` process timeout | +| `LOG_DIR` | `./data/logs` | no | Log root directory | +| `LOG_LEVEL` | `info` | no | `debug`, `info`, `warn`, or `error` | +| `LOG_FORMAT` | `text` | no | `text` for terminal, `json` for log files | +| `ANALYZER_DATA_SKEW_RATIO` | `3.0` | no | Data-skew rule threshold (max/min ratio) | +| `ANALYZER_GC_PRESSURE_RATIO` | `0.1` | no | GC-pressure rule threshold (GC/CPU ratio) | +| `ANALYZER_BOTTLENECK_SHUFFLE_GB` | `50` | no | Bottleneck rule threshold (shuffle GB) | + +## Admin API + +All endpoints require `Authorization: Bearer `. + +| Method | Path | Description | +|---|---|---| +| `GET` | `/admin/clusters` | List all clusters | +| `POST` | `/admin/clusters` | Create a cluster | +| `GET` | `/admin/clusters/:id` | Get one cluster | +| `PUT` | `/admin/clusters/:id` | Update a cluster | +| `DELETE` | `/admin/clusters/:id` | Delete a cluster | +| `GET` | `/admin/audit?limit=100` | Query audit log | + +Create a cluster: + +```bash +curl -sS -X POST \\ + -H "Authorization: Bearer ${ADMIN_TOKEN}" \\ + -H "Content-Type: application/json" \\ + -d '{ + "id": "prod", + "name": "prod", + "rm_url": "http://rm.example.com:8088", + "shs_url": "http://shs.example.com:18080", + "spark_submit_execute_bin": "/opt/spark/bin/spark-submit", + "is_active": true, + "auth_type": "simple", + "auth_username": "yarn", + "rate_limit_per_min": 10, + "url_allowlist": ["rm.example.com:8088", "shs.example.com:18080"] + }' \\ + "http://127.0.0.1:${LISTEN_ADDR:-:8080}/admin/clusters" +``` + +Note: `auth_password` is not accepted via JSON (`json:"-"`). Set it through a +dedicated password endpoint or seed the database directly. + +## Security model + +Three lines of defense: + +1. **Token auth** - `ADMIN_TOKENS` protects `/admin/*`; `AGENT_TOKEN` protects `/mcp`. +2. **SSRF guard** - Every outbound URL must match the cluster's `url_allowlist`, + with private-IP CIDR blacklist (incl. `169.254.0.0/16` AWS/GCP metadata). +3. **Path allowlist** - `upload_file` writes only to `./data/uploads/`, rejects + absolute paths, `..`, and non-`[a-zA-Z0-9._-]` filenames. + +Additional guarantees: + +- `spark_submit` runs the cluster's binary via `exec.Command(name, args...)` — + slice form, never `sh -c`. No shell metacharacter interpretation. +- `auth_password` is tagged `json:"-"`; never serialized in responses, never + accepted from JSON request bodies. +- Cross-host redirects (RM → NM 307) preserve the `Authorization` header + through `DoWithRedirect` only when the destination host is in + `url_allowlist`. +- Per-Tool log files are created with mode `0600` and live under + `./data/logs/tools/`. + +## Development + +```bash +go build ./... +go test ./... +go vet ./... +``` + +## More docs + +- [ARCHITECTURE.md](ARCHITECTURE.md) - Design and package layout +- [docs/runbook.md](docs/runbook.md) - Deployment, upgrade, and troubleshooting diff --git a/docs/runbook.md b/docs/runbook.md new file mode 100644 index 0000000..70e4f17 --- /dev/null +++ b/docs/runbook.md @@ -0,0 +1,80 @@ +# spark-mcp-go 部署与排障手册 + +## 部署步骤 + +### 1. 构建 + +```bash +cd /opt/spark-mcp-go +go build -o spark-mcp-go . +``` + +### 2. 配置 + +```bash +cp .env.example .env +# 编辑 .env:ADMIN_TOKENS、AGENT_TOKEN、LISTEN_ADDR、DATA_DIR 等 +``` + +### 3. 使用 systemd 运行 + +参见 [docs/systemd/spark-mcp-go.service](systemd/spark-mcp-go.service)。 + +### 4. 初始化示例 cluster + +```bash +ADMIN_HOST=http://127.0.0.1:8080 ./scripts/seed.sh prod http://rm.example.com:8088 http://shs.example.com:18080 /opt/spark/bin/spark-submit +``` + +### 5. 升级流程 + +1. 停止服务:`systemctl stop spark-mcp-go` +2. 备份数据库:`cp ./data/spark-mcp.db ./data/spark-mcp.db.bak` +3. 替换二进制:`cp spark-mcp-go /usr/local/bin/` +4. 启动服务:`systemctl start spark-mcp-go` +5. 验证:`curl -sS http://127.0.0.1:${LISTEN_ADDR:-:8080}/healthz` + +## 故障排查 + +### 401 Unauthorized + +- 检查请求是否带 `Authorization: Bearer `。 +- 区分 `ADMIN_TOKENS`(管理面)和 `AGENT_TOKEN`(MCP 面)。 +- `.env` 中 token 不能含前导/尾随空格。 + +### 404 "Invalid session ID" + +- Streamable HTTP 需要先走 MCP `initialize` 握手,拿到 `Mcp-Session-Id` 后再调用 Tool。 +- 确保客户端保存并回传 response header 中的 session ID。 + +### 422 cluster not found + +- `cluster_id` 拼写错误,或 cluster 未激活。 +- 先用 `list_clusters` 确认可用的 cluster ID。 + +### "host not in allowlist" + +- 对应 cluster 的 `url_allowlist` 未包含目标主机。 +- 使用 `admin` 更新 cluster,把 RM/SHS 主机加入 allowlist。 + +### `spark_submit` 返回 exit_code 7 + +- `spark_submit_execute_bin` 路径错误,或 YARN 拒绝提交。 +- 查看返回的 `stderr_tail`;在服务器上直接运行相同命令验证。 + +### `analyze_spark_log` 的 `findings` 为空 + +- SHS metrics 端点 404 或返回格式不匹配。 +- 可 fallback 到 raw log + LLM 自行分析。 + +### SQLite 锁冲突 + +- `storage` 已开启 WAL 模式,但 SQLite 仍是 single-writer。 +- 多个 server 实例共用一个 `spark-mcp.db` 会产生锁竞争;请保持一实例一数据库。 + +## 监控建议 + +- **存活探测**:HTTP `GET /healthz` +- **日志告警**:监控 `data/logs/spark-mcp.log` 中的 `ERROR` 级别行 +- **审计日志告警**:`data/logs/tools/` 下文件过大或增长过快时检查异常调用 +- **磁盘告警**:`data/` 目录包含 SQLite、上传文件、日志,需预留空间 diff --git a/docs/systemd/spark-mcp-go.service b/docs/systemd/spark-mcp-go.service new file mode 100644 index 0000000..0d70598 --- /dev/null +++ b/docs/systemd/spark-mcp-go.service @@ -0,0 +1,22 @@ +[Unit] +Description=Spark MCP Go Server +After=network.target + +[Service] +Type=simple +User=spark-mcp +Group=spark-mcp +WorkingDirectory=/opt/spark-mcp-go +EnvironmentFile=/opt/spark-mcp-go/.env +ExecStart=/opt/spark-mcp-go/spark-mcp-go +Restart=on-failure +RestartSec=5 + +# Security hardening +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=true +ReadWritePaths=/opt/spark-mcp-go/data + +[Install] +WantedBy=multi-user.target diff --git a/scripts/dev.sh b/scripts/dev.sh new file mode 100755 index 0000000..eb455ff --- /dev/null +++ b/scripts/dev.sh @@ -0,0 +1,45 @@ +#!/usr/bin/env bash +# One-command build + start + health check for spark-mcp-go. +set -euo pipefail + +cd "$(dirname "$0")/.." + +# Load .env if present +if [ -f .env ]; then + set -a + # shellcheck source=/dev/null + . ./.env + set +a +fi + +# Required tokens +: "${ADMIN_TOKENS:?ADMIN_TOKENS is required}" +: "${AGENT_TOKEN:?AGENT_TOKEN is required}" + +LISTEN_ADDR="${LISTEN_ADDR:-:8080}" +PORT="${LISTEN_ADDR##*:}" +PID_FILE="./data/spark-mcp.pid" + +echo ">> go build" +go build -o ./spark-mcp-go . + +mkdir -p ./data/logs/tools + +echo ">> starting server on port ${PORT}" +nohup ./spark-mcp-go > ./data/logs/spark-mcp.log 2>&1 & +PID=$! +echo "${PID}" > "${PID_FILE}" + +sleep 2 + +echo ">> health check" +if curl -sS "http://127.0.0.1:${PORT}/healthz"; then + echo + echo ">> running, PID=${PID}, log=./data/logs/spark-mcp.log" +else + echo + echo ">> health check failed; see ./data/logs/spark-mcp.log" + kill "${PID}" 2>/dev/null || true + rm -f "${PID_FILE}" + exit 1 +fi diff --git a/scripts/seed.sh b/scripts/seed.sh new file mode 100755 index 0000000..906050a --- /dev/null +++ b/scripts/seed.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash +# Admin POST to create a sample cluster. +# Usage: ./scripts/seed.sh [cluster_id] [rm_url] [shs_url] [spark_submit_bin] +set -euo pipefail + +cd "$(dirname "$0")/.." + +# Load .env if present +if [ -f .env ]; then + set -a + # shellcheck source=/dev/null + . ./.env + set +a +fi + +: "${ADMIN_TOKENS:?ADMIN_TOKENS is required}" + +HOST="${ADMIN_HOST:-http://127.0.0.1:8080}" +TOKEN="${ADMIN_TOKENS%%,*}" +AUTH="Authorization: Bearer ${TOKEN}" + +CLUSTER_ID="${1:-prod}" +RM_URL="${2:-http://rm.example.com:8088}" +SHS_URL="${3:-http://shs.example.com:18080}" +SPARK_BIN="${4:-/opt/spark/bin/spark-submit}" + +echo ">> creating cluster '${CLUSTER_ID}' via ${HOST}/admin/clusters" + +curl -sS -X POST -H "${AUTH}" -H "Content-Type: application/json" \ + -d "{\"id\":\"${CLUSTER_ID}\",\"name\":\"${CLUSTER_ID}\",\"rm_url\":\"${RM_URL}\",\"shs_url\":\"${SHS_URL}\",\"spark_submit_execute_bin\":\"${SPARK_BIN}\",\"is_active\":true,\"auth_type\":\"simple\",\"auth_username\":\"yarn\",\"rate_limit_per_min\":10,\"url_allowlist\":[\"${RM_URL#*://}\",\"${SHS_URL#*://}\"]}" \ + "${HOST}/admin/clusters" + +echo