tao.chenandClaude
e6411b0dc4
- internal/analyzer: 纯函数规则引擎
- Finding / Severity / Thresholds / Input / StageMetric / ExecutorMetric
- 3 规则: data_skew (max/min ratio), gc_pressure (GC/CPU ratio),
bottleneck (shuffle read+write GB)
- 严重度阶梯: warning / critical (阈值 ×2)
- Analyze 入口数据驱动 for ... range
- 11 个单测 (3 规则 × normal/warning/critical + 零除防御)
- fetch_spark_metrics: SHS 业务封装, format=summary 触发 analyzer
- fetch_cluster_env: RM /cluster/info + /cluster/metrics 聚合
- analyze_spark_log: RM 日志降级链 + LLM-Ready Prompt 拼装
- prompt 结构: cluster 元信息 + log source + log tail +
heuristic findings + suggested LLM analysis
- findings 是 first-pass filter, LLM 做最终判断
- deps.go: +AnalyzerThresholds
- main.go: 注入 cfg 的 3 个阈值
- 端到端实测: analyze_spark_log 完整 prompt 渲染
完成 11 个 Tool 表面 (list_clusters / spark_submit / 4 RM /
fetch_spark_metrics / fetch_cluster_env / analyze_spark_log /
fetch_url / upload_file)
Co-Authored-By: Claude <noreply@anthropic.com>
67 lines
1.9 KiB
Go
67 lines
1.9 KiB
Go
package tools
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/mark3labs/mcp-go/mcp"
|
|
)
|
|
|
|
const FetchClusterEnvName = "fetch_cluster_env"
|
|
|
|
// NewFetchClusterEnvTool returns the schema for the fetch_cluster_env MCP Tool.
|
|
func NewFetchClusterEnvTool() mcp.Tool {
|
|
return mcp.NewTool(FetchClusterEnvName,
|
|
mcp.WithDescription("Fetch YARN cluster environment information and metrics from the ResourceManager."),
|
|
mcp.WithString("cluster_id",
|
|
mcp.Required(),
|
|
mcp.Description("ID of the configured cluster (from list_clusters)"),
|
|
),
|
|
)
|
|
}
|
|
|
|
// FetchClusterEnvHandler fetches RM cluster info and metrics.
|
|
func (d *Deps) FetchClusterEnvHandler(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
clusterID, err := req.RequireString("cluster_id")
|
|
if err != nil {
|
|
return errResult("fetch_cluster_env: " + err.Error()), nil
|
|
}
|
|
|
|
callLog := startToolCall(ctx, d.Logger, FetchClusterEnvName, map[string]any{
|
|
"cluster_id": clusterID,
|
|
})
|
|
defer callLog.End()
|
|
|
|
cl, err := d.ClusterRepo.Get(ctx, clusterID)
|
|
if err != nil {
|
|
callLog.WithError(err)
|
|
return errResult("fetch_cluster_env: cluster " + clusterID + ": " + err.Error()), nil
|
|
}
|
|
|
|
base := strings.TrimRight(cl.RMURL, "/")
|
|
infoBody, err := d.fetchInternal(ctx, cl, base+"/ws/v1/cluster/info")
|
|
if err != nil {
|
|
callLog.WithError(err)
|
|
return errResult("fetch_cluster_env: info: " + err.Error()), nil
|
|
}
|
|
metricsBody, err := d.fetchInternal(ctx, cl, base+"/ws/v1/cluster/metrics")
|
|
if err != nil {
|
|
callLog.WithError(err)
|
|
return errResult("fetch_cluster_env: metrics: " + err.Error()), nil
|
|
}
|
|
|
|
var info, metrics any
|
|
_ = json.Unmarshal(infoBody, &info)
|
|
_ = json.Unmarshal(metricsBody, &metrics)
|
|
|
|
result := map[string]any{
|
|
"cluster_info": info,
|
|
"metrics": metrics,
|
|
"fetched_at": time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
callLog.WithResult(map[string]any{"info_bytes": len(infoBody), "metrics_bytes": len(metricsBody)})
|
|
return textResult(encodeJSON(result)), nil
|
|
}
|