tools: 仿照 Python build_spark_submit_command 将 spark_submit 改为结构化参数

将 spark_submit 的自由 args[] 替换为与 Python 服务一致的 schema：cluster_id、 master、deploy_mode、script_path、queue、executor_memory、executor_cores、 num_executors、spark_conf、extra_args。命令行顺序严格对齐 Python 实现： binary → --master → --queue → --executor-memory → --executor-cores → --num-executors → 按 key 排序的 --conf key=value → 按 key 排序的 --flag value → script_path 永远在最后。 deploy_mode 是 Python 参考函数的入参，但该函数实际并不输出 --deploy-mode； Go 端保留这个字段并校验其值，但同样不发射它，以保持行为一致。如果后续需要 --deploy-mode，再在此处添加并记录差异。 cluster.DefaultSubmitArgs 不再被拼接到 argv 中；该字段仍保留用于 admin API 向后兼容，运行时移除它的工作是后续独立的 follow-up。删除了 translateArgs、pathLikeArg 以及 spark_submit.unminted_path 警告逻辑（本次提交取代 2570713），因为自由参数已不存在，警告无从触发。新增 parseStringMap、sortedStringKeys 和 requireNonNegativeInt 辅助函数； requireNonNegativeInt 同时接受 int 与 float64，以兼容测试直接构造的整数字面量和 JSON 解码后的 float64。测试更新： - TestSparkSubmit_StructuredCommand：验证完整 argv 顺序，并断言 script_path 是最后一项。 - TestSparkSubmit_MissingRequiredField：校验缺少 master 时返回错误。 - TestSparkSubmit_BadSparkConfValue：校验 spark_conf 的值不是字符串时报错。 Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-13 19:35:16 +08:00
parent 2570713fbf
commit 7920dc9597
2 changed files with 239 additions and 203 deletions
@@ -3,14 +3,12 @@ package tools
 import (
 	"context"
 	"fmt"
-	"log/slog"
-	"path/filepath"
-	"strings"
+	"sort"
+	"strconv"

 	"github.com/mark3labs/mcp-go/mcp"

 	"spark-mcp-go/internal/executor"
-	"spark-mcp-go/internal/uploads"
 )

 const SparkSubmitName = "spark_submit"
@@ -18,15 +16,47 @@ const SparkSubmitName = "spark_submit"
 // NewSparkSubmitTool returns the schema for the spark_submit MCP Tool.
 func NewSparkSubmitTool() mcp.Tool {
 	return mcp.NewTool(SparkSubmitName,
-		mcp.WithDescription("Submit a Spark application to a configured cluster using the cluster's local spark-submit binary. The `args` array is appended to the cluster's default_submit_args; pass one argv element per CLI flag. If the script to run was just uploaded via upload_file, put the `path` field from that response as one of the args — do NOT construct your own path. Cluster-local paths (e.g. /opt/spark/examples/pi.py) are accepted but will be logged as a warning. To avoid the warning, always use the `path` field returned by upload_file. Returns {app_id, exit_code, stdout_tail, stderr_tail, duration_ms}. Binary is invoked as a child process — no shell, no command injection."),
+		mcp.WithDescription("Submit a Spark application to a configured cluster. The command is built in this order: --master, --queue, --executor-memory, --executor-cores, --num-executors, then --conf per spark_conf entry, then --flag value per extra_args entry, then script_path last. For script_path, pass the `path` field returned by upload_file (do not construct your own path). Returns {app_id, exit_code, stdout_tail, stderr_tail, duration_ms}. Binary is invoked as a child process — no shell, no command injection."),
 		mcp.WithString("cluster_id",
 			mcp.Required(),
 			mcp.Description("ID of the configured cluster (from list_clusters)"),
 		),
-		// mcp-go v0.56.0 uses PropertyOption for array item schema.
-		mcp.WithArray("args",
-			mcp.Description("Spark-submit CLI args (after default_submit_args). Each element becomes one argv entry — no shell interpretation."),
-			mcp.Items(map[string]any{"type": "string"}),
+		mcp.WithString("master",
+			mcp.Required(),
+			mcp.Description("Spark master URL, e.g. yarn, k8s://https://..."),
+		),
+		mcp.WithString("deploy_mode",
+			mcp.Required(),
+			mcp.Description("Required for forward compatibility; not currently emitted in the command by the builder (matches Python reference)."),
+			mcp.Enum("client", "cluster"),
+		),
+		mcp.WithString("script_path",
+			mcp.Required(),
+			mcp.Description("Absolute path to the script. Pass the path field returned by upload_file — do not construct your own path. The script is always the last argv element."),
+		),
+		mcp.WithString("queue",
+			mcp.Required(),
+			mcp.Description("YARN queue name"),
+		),
+		mcp.WithString("executor_memory",
+			mcp.Required(),
+			mcp.Description("e.g. 4G"),
+		),
+		mcp.WithNumber("executor_cores",
+			mcp.Required(),
+			mcp.Description("Cores per executor (integer)"),
+		),
+		mcp.WithNumber("num_executors",
+			mcp.Required(),
+			mcp.Description("Static executor count (integer)"),
+		),
+		mcp.WithObject("spark_conf",
+			mcp.Description("Map of key=value pairs, each emitted as --conf key=value"),
+			mcp.AdditionalProperties(map[string]any{"type": "string"}),
+		),
+		mcp.WithObject("extra_args",
+			mcp.Description("Map of flag -> value, each emitted as --flag value"),
+			mcp.AdditionalProperties(map[string]any{"type": "string"}),
 		),
 	)
 }
@@ -37,26 +67,65 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
 	if err != nil {
 		return errResult("spark_submit: " + err.Error()), nil
 	}
+	master, err := req.RequireString("master")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}
+	deployMode, err := req.RequireString("deploy_mode")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}
+	scriptPath, err := req.RequireString("script_path")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}
+	queue, err := req.RequireString("queue")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}
+	executorMemory, err := req.RequireString("executor_memory")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}

 	args := req.GetArguments()
-	var userArgs []string
-	if rawArgs, ok := args["args"]; ok && rawArgs != nil {
-		arr, ok := rawArgs.([]any)
-		if !ok {
-			return errResult("spark_submit: args is not an array"), nil
-		}
-		for i, item := range arr {
-			s, ok := item.(string)
-			if !ok {
-				return errResult(fmt.Sprintf("spark_submit: args[%d] is not a string", i)), nil
-			}
-			userArgs = append(userArgs, s)
-		}
+
+	// deploy_mode is required for forward compatibility but the builder does not
+	// emit --deploy-mode, matching the Python reference's actual behavior. If the
+	// Go side needs --deploy-mode later, add it here and document the divergence.
+	if deployMode != "client" && deployMode != "cluster" {
+		return errResult("spark_submit: deploy_mode must be client or cluster"), nil
+	}
+
+	sparkConf, err := parseStringMap(args, "spark_conf")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}
+	extraArgs, err := parseStringMap(args, "extra_args")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}
+
+	executorCores, err := requireNonNegativeInt(args, "executor_cores")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
+	}
+	numExecutors, err := requireNonNegativeInt(args, "num_executors")
+	if err != nil {
+		return errResult("spark_submit: " + err.Error()), nil
 	}

 	callLog := startToolCall(ctx, d.Logger, SparkSubmitName, map[string]any{
-		"cluster_id": clusterID,
-		"args":       userArgs,
+		"cluster_id":      clusterID,
+		"master":          master,
+		"deploy_mode":     deployMode,
+		"script_path":     scriptPath,
+		"queue":           queue,
+		"executor_memory": executorMemory,
+		"executor_cores":  executorCores,
+		"num_executors":   numExecutors,
+		"spark_conf":      sparkConf,
+		"extra_args":      extraArgs,
 	})
 	defer callLog.End()

@@ -66,12 +135,34 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
 		return errResult("spark_submit: cluster " + clusterID + ": " + err.Error()), nil
 	}

-	fullArgs := append([]string{}, cl.DefaultSubmitArgs...)
-	fullArgs = append(fullArgs, translateArgs(userArgs, d.UploadStore, d.Logger)...)
+	binary := cl.SparkSubmitExecuteBin
+
+	// Build argv inline, mirroring the Python reference's output order exactly.
+	// DefaultSubmitArgs is no longer prepended; the field is currently unused at
+	// runtime but kept for backward compatibility with the admin API. Removal is
+	// a separate follow-up.
+	cmd := []string{binary}
+	cmd = append(cmd, "--master", master)
+	cmd = append(cmd, "--queue", queue)
+	cmd = append(cmd, "--executor-memory", executorMemory)
+	cmd = append(cmd, "--executor-cores", strconv.Itoa(executorCores))
+	cmd = append(cmd, "--num-executors", strconv.Itoa(numExecutors))
+
+	for _, k := range sortedStringKeys(sparkConf) {
+		cmd = append(cmd, "--conf", k+"="+sparkConf[k])
+	}
+	for _, k := range sortedStringKeys(extraArgs) {
+		cmd = append(cmd, "--"+k, extraArgs[k])
+	}
+	cmd = append(cmd, scriptPath)
+
+	if d.Logger != nil {
+		d.Logger.Debug("spark_submit.built", "cluster_id", clusterID, "argv", cmd)
+	}

 	result, err := executor.Run(ctx, executor.SparkSubmitOpts{
-		Binary:  cl.SparkSubmitExecuteBin,
-		Args:    fullArgs,
+		Binary:  binary,
+		Args:    cmd,
 		Timeout: d.SparkSubmitTimeout,
 	})
 	if err != nil {
@@ -103,52 +194,57 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
 	return textResult(encodeJSON(result)), nil
 }

-// translateArgs canonicalizes server-minted upload paths and leaves
-// cluster-local paths untouched. It is a code-level guardrail, not a
-// validator: non-minted absolute paths are intentionally accepted.
-func translateArgs(userArgs []string, store *uploads.Store, logger *slog.Logger) []string {
-	if store == nil {
-		return userArgs
+// requireNonNegativeInt extracts an integer parameter from the request and
+// rejects negative or non-integer values.
+func requireNonNegativeInt(args map[string]any, key string) (int, error) {
+	v, ok := args[key]
+	if !ok || v == nil {
+		return 0, fmt.Errorf("%s is required", key)
 	}
-	out := make([]string, len(userArgs))
-	for i, arg := range userArgs {
-		pathPart, isPath := pathLikeArg(arg)
-		if isPath {
-			if fileID, err := store.Validate(pathPart); err == nil {
-				if canonical, err := store.AbsPath(fileID); err == nil {
-					out[i] = canonical
-					continue
-				}
-			} else if logger != nil && filepath.IsAbs(pathPart) {
-				// Absolute path that this server did not mint. Still allowed,
-				// but warn so operators can spot LLM-invented remote paths.
-				logger.Warn("spark_submit.unminted_path",
-					"path", pathPart,
-					"note", "not produced by upload_file on this server; cluster-local paths are valid, but if this came from a remote agent the path is unreachable")
-			}
+	switch n := v.(type) {
+	case int:
+		if n < 0 {
+			return 0, fmt.Errorf("%s must be a non-negative integer", key)
 		}
-		out[i] = arg
+		return n, nil
+	case float64:
+		ni := int(n)
+		if n < 0 || float64(ni) != n {
+			return 0, fmt.Errorf("%s must be a non-negative integer", key)
+		}
+		return ni, nil
+	default:
+		return 0, fmt.Errorf("%s must be a number", key)
 	}
-	return out
 }

-// pathLikeArg extracts an absolute path from a spark-submit argv element.
-// It handles plain absolute paths ("/opt/spark/pi.py") and flag-style values
-// ("--conf=/x/y"). Non-path args like "--class" or "Main" return ok=false.
-func pathLikeArg(arg string) (string, bool) {
-	if arg == "" {
-		return "", false
+// parseStringMap extracts an optional object whose values are strings.
+func parseStringMap(args map[string]any, key string) (map[string]string, error) {
+	out := make(map[string]string)
+	v, ok := args[key]
+	if !ok || v == nil {
+		return out, nil
 	}
-	if strings.HasPrefix(arg, "/") {
-		return arg, true
+	m, ok := v.(map[string]any)
+	if !ok {
+		return nil, fmt.Errorf("%s must be an object", key)
 	}
-	if strings.HasPrefix(arg, "-") {
-		if eq := strings.Index(arg, "="); eq > 0 && eq < len(arg)-1 {
-			val := arg[eq+1:]
-			if strings.HasPrefix(val, "/") {
-				return val, true
-			}
+	for k, val := range m {
+		s, ok := val.(string)
+		if !ok {
+			return nil, fmt.Errorf("%s[%q] must be a string", key, k)
 		}
+		out[k] = s
 	}
-	return "", false
+	return out, nil
+}
+
+// sortedStringKeys returns the keys of m in deterministic order.
+func sortedStringKeys(m map[string]string) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	return keys
 }