package tools import ( "context" "fmt" "github.com/mark3labs/mcp-go/mcp" "spark-mcp-go/internal/analyzer" "spark-mcp-go/internal/cluster" "spark-mcp-go/internal/rm" ) const AnalyzeSparkLogName = "analyze_spark_log" const defaultAnalyzeMaxLogBytes = 10240 // NewAnalyzeSparkLogTool returns the schema for the analyze_spark_log MCP Tool. func NewAnalyzeSparkLogTool() mcp.Tool { return mcp.NewTool(AnalyzeSparkLogName, mcp.WithDescription("Fetch YARN application logs and produce an LLM-ready analysis prompt with heuristic findings."), mcp.WithString("cluster_id", mcp.Required(), mcp.Description("ID of the configured cluster (from list_clusters)"), ), mcp.WithString("app_id", mcp.Required(), mcp.Description("YARN application ID, e.g. application_1234567890_0001"), ), mcp.WithString("container", mcp.Description("Container ID used for the aggregated-logs fallback"), ), mcp.WithNumber("max_log_bytes", mcp.Description("Maximum bytes of the log tail to include in the prompt"), mcp.DefaultNumber(10240), ), ) } // AnalyzeSparkLogHandler retrieves logs and builds an LLM-ready prompt. func (d *Deps) AnalyzeSparkLogHandler(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { clusterID, err := req.RequireString("cluster_id") if err != nil { return errResult("analyze_spark_log: " + err.Error()), nil } appID, err := req.RequireString("app_id") if err != nil { return errResult("analyze_spark_log: " + err.Error()), nil } args := req.GetArguments() container := "" if v, ok := args["container"].(string); ok { container = v } if container == "" { container = fmt.Sprintf("container_%s_01", appID) } maxLogBytes := defaultAnalyzeMaxLogBytes if v, ok := args["max_log_bytes"].(float64); ok { maxLogBytes = int(v) } if maxLogBytes <= 0 { maxLogBytes = defaultAnalyzeMaxLogBytes } callLog := startToolCall(ctx, d.Logger, AnalyzeSparkLogName, map[string]any{ "cluster_id": clusterID, "app_id": appID, "container": container, "max_log_bytes": maxLogBytes, }) defer callLog.End() cl, err := d.ClusterRepo.Get(ctx, clusterID) if err != nil { callLog.WithError(err) return errResult("analyze_spark_log: cluster " + clusterID + ": " + err.Error()), nil } rmc := rm.New(d.HTTPClient, cl) body, source, err := rmc.GetLogs(ctx, appID, container) if err != nil { callLog.WithError(err) return errResult("analyze_spark_log: " + err.Error()), nil } logTail := truncateTail(string(body), maxLogBytes) var findings []analyzer.Finding input, metricsErr := d.sparkMetricsInput(ctx, cl, appID) if metricsErr == nil { findings = analyzer.Analyze(input, d.AnalyzerThresholds) } prompt := buildSparkLogPrompt(appID, cl, source, len(body), maxLogBytes, logTail, findings) result := map[string]any{ "findings": findings, "log_source": source, "log_tail": logTail, "prompt": prompt, } callLog.WithResult(map[string]any{"source": source, "bytes": len(body), "findings": len(findings)}) return textResult(encodeJSON(result)), nil } func buildSparkLogPrompt(appID string, cl *cluster.Cluster, source string, totalBytes, maxBytes int, logTail string, findings []analyzer.Finding) string { findingsBlock := "no heuristic findings; rely on raw log + LLM analysis" if len(findings) > 0 { findingsBlock = encodeJSON(findings) } return fmt.Sprintf("# Spark Log Analysis for %s\n\n## Cluster\n%s (%s) — RM=%s, SHS=%s\n\n## Log Source\n%s (%d bytes, last %d shown below)\n\n## Log Tail\n"+"```"+"\n%s\n"+"```"+"\n\n## Heuristic Findings\n%s\n\n## Suggested LLM Analysis\n1. Check for ERROR/Exception stack traces\n2. Look for OOM/Timeout/Shuffle fetch failures\n3. Identify slow stages (compare to median)\n4. Suggest mitigations\n", appID, cl.Name, cl.ID, cl.RMURL, cl.SHSURL, source, totalBytes, maxBytes, logTail, findingsBlock) }