spark-mcp/internal/mcp/tools/analyze_spark_log.go

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)
}