gitignore: 跨平台 build 产物 spark-mcp-*-*

覆盖 GOOS-GOARCH 命名约定的所有二进制变体 (spark-mcp-linux-amd64, spark-mcp-darwin-arm64 等). 之前每次交叉编译都会留一个未跟踪的二进制在仓库根. Co-Authored-By: Claude <noreply@anthropic.com>
logging: tool.call.error/end 接入主 logger (双写终端 + spark-mcp.log)
2026-07-13 15:28:01 +08:00 · 2026-07-13 15:27:28 +08:00 · 2026-07-13 15:15:21 +08:00 · 2026-07-13 14:16:49 +08:00 · 2026-07-13 14:08:16 +08:00
11 changed files with 537 additions and 28 deletions
@@ -23,3 +23,6 @@ htmlcov/

 # macOS
 .DS_Store
+
+# Build artifacts (cross-compiled binaries)
+spark-mcp-*-*
@@ -19,9 +19,13 @@ import (
 // Mount attaches the /admin sub-router to r, protecting every route with
 // bearer-token admin authentication.
 func Mount(r *gin.Engine, repo *storage.ClusterRepo, auditRepo *audit.Repo, adminTokens []string) {
+	// HTML page is public — its own modal prompts for the token.
+	// All other /admin/* endpoints (API + OpenAPI docs) still require auth.
+	gPublic := r.Group("/admin")
+	gPublic.GET("", webHandler)
+	gPublic.GET("/", webHandler)
+
 	g := r.Group("/admin", middleware.AdminAuth(adminTokens))
-	g.GET("", webHandler)
-	g.GET("/", webHandler)
 	g.GET("/clusters", listClusters(repo))
 	g.POST("/clusters", createCluster(repo, auditRepo))
 	g.GET("/clusters/:id", getCluster(repo))
@@ -96,6 +96,45 @@ func TestMount_RequiresAuth(t *testing.T) {
 	}
 }

+func TestAdminWeb_NoTokenReturnsHTML(t *testing.T) {
+	r, _, _ := newTestAdmin(t)
+
+	// 1) /admin without token: 200 + HTML
+	w := doReq(t, r, "GET", "/admin", "", nil)
+	if w.Code != http.StatusOK {
+		t.Fatalf("GET /admin without token: got %d, want 200", w.Code)
+	}
+	ct := w.Header().Get("Content-Type")
+	if !strings.Contains(ct, "text/html") {
+		t.Errorf("Content-Type = %q, want text/html", ct)
+	}
+	body := w.Body.String()
+	if !strings.Contains(body, "token-modal") {
+		t.Errorf("body missing token-modal element; body excerpt: %.200s", body)
+	}
+	if !strings.Contains(body, "Admin Token Required") {
+		t.Errorf("body missing modal title")
+	}
+
+	// 2) /admin/ (trailing slash) same behavior
+	w2 := doReq(t, r, "GET", "/admin/", "", nil)
+	if w2.Code != http.StatusOK {
+		t.Fatalf("GET /admin/ without token: got %d, want 200", w2.Code)
+	}
+
+	// 3) API routes still require auth (no token = 401)
+	w3 := doReq(t, r, "GET", "/admin/clusters", "", nil)
+	if w3.Code != http.StatusUnauthorized {
+		t.Errorf("GET /admin/clusters without token: got %d, want 401", w3.Code)
+	}
+
+	// 4) API routes still require auth (valid token = 200)
+	w4 := doReq(t, r, "GET", "/admin/clusters", "good-token", nil)
+	if w4.Code != http.StatusOK {
+		t.Errorf("GET /admin/clusters with good-token: got %d, want 200", w4.Code)
+	}
+}
+
 func TestListClusters(t *testing.T) {
 	r, _, _ := newTestAdmin(t)

@@ -91,15 +91,29 @@
    .form-actions { display: flex; gap: .5rem; margin-top: 1rem; }
    .bool { flex-direction: row; align-items: center; gap: .5rem; }
    .bool input { width: auto; }
+    dialog.modal {
+      border: 1px solid var(--border);
+      background: var(--panel);
+      color: var(--text);
+      border-radius: 6px;
+      padding: 1.5rem;
+      min-width: 360px;
+      max-width: 90vw;
+    }
+    dialog.modal::backdrop { background: rgba(0, 0, 0, .65); }
+    dialog.modal h2 { margin: 0 0 .5rem; font-size: 1.1rem; color: var(--accent); }
+    dialog.modal form { display: flex; flex-direction: column; gap: .75rem; }
+    dialog.modal label { color: var(--muted); font-size: .9rem; }
+    dialog.modal input { width: 100%; }
+    .modal-error { color: var(--danger); font-size: .9rem; margin: 0; }
+    .modal-actions { display: flex; gap: .5rem; justify-content: flex-end; }
  </style>
 </head>
 <body>
  <header>
    <h1>spark-mcp-go Admin</h1>
    <div class='auth'>
-      <label for='token'>Token</label>
-      <input id='token' type='password' placeholder='Bearer token' autocomplete='off'>
-      <button id='save-token'>Save</button>
+      <button id='set-token' class='secondary'>Set Token</button>
      <button id='logout' class='danger'>Logout</button>
    </div>
  </header>
@@ -199,12 +213,25 @@
    </section>
  </main>

+  <dialog id='token-modal' class='modal'>
+    <h2>Admin Token Required</h2>
+    <p id='modal-error' class='modal-error hidden'></p>
+    <form id='token-form'>
+      <label for='modal-token'>Bearer token</label>
+      <input id='modal-token' type='password' autocomplete='off' required>
+      <div class='modal-actions'>
+        <button type='submit' id='modal-save'>Save</button>
+        <button type='button' id='modal-cancel' class='secondary hidden'>Cancel</button>
+      </div>
+    </form>
+  </dialog>
+
  <script>
    const $ = (sel) => document.querySelector(sel);
    const API = '/admin';

-    const tokenInput = $('#token');
-    tokenInput.value = localStorage.getItem('adminToken') || '';
+    let pendingRetry = null;
+    let cancelBlocker = null;

    function headers() {
      return {
@@ -219,11 +246,67 @@
      el.classList.toggle('hidden', !msg);
    }

+    function showModalError(msg) {
+      const el = $('#modal-error');
+      el.textContent = msg || '';
+      el.classList.toggle('hidden', !msg);
+    }
+
+    function openTokenModal(opts = {}) {
+      const dialog = $('#token-modal');
+      const cancelBtn = $('#modal-cancel');
+      showModalError(opts.error || '');
+      cancelBtn.classList.toggle('hidden', !opts.allowCancel);
+      if (cancelBlocker) {
+        dialog.removeEventListener('cancel', cancelBlocker);
+        cancelBlocker = null;
+      }
+      if (!opts.allowCancel) {
+        cancelBlocker = (e) => e.preventDefault();
+        dialog.addEventListener('cancel', cancelBlocker);
+      }
+      dialog.showModal();
+      $('#modal-token').focus();
+    }
+
+    function closeTokenModal() {
+      const dialog = $('#token-modal');
+      showModalError('');
+      if (cancelBlocker) {
+        dialog.removeEventListener('cancel', cancelBlocker);
+        cancelBlocker = null;
+      }
+      dialog.close();
+      $('#modal-token').value = '';
+    }
+
+    function saveTokenFromModal(e) {
+      e.preventDefault();
+      localStorage.setItem('adminToken', $('#modal-token').value.trim());
+      closeTokenModal();
+      if (pendingRetry) {
+        const fn = pendingRetry;
+        pendingRetry = null;
+        fn();
+      } else {
+        loadClusters();
+      }
+    }
+
    async function api(method, path, body) {
      const opts = { method, headers: headers() };
      if (body !== undefined) opts.body = JSON.stringify(body);
      const res = await fetch(API + path, opts);
      if (!res.ok) {
+        if (res.status === 401) {
+          localStorage.removeItem('adminToken');
+          pendingRetry = () => api(method, path, body);
+          openTokenModal({
+            error: 'Token rejected by server (401). Enter a valid token.',
+            allowCancel: false
+          });
+          throw new Error('401 Unauthorized');
+        }
        let detail = res.statusText;
        try {
          const j = await res.json();
@@ -368,16 +451,18 @@
    $('#refresh').addEventListener('click', loadClusters);
    $('#btn-cancel').addEventListener('click', hideForm);

-    $('#save-token').addEventListener('click', () => {
-      localStorage.setItem('adminToken', tokenInput.value.trim());
-      loadClusters();
-    });
+    $('#set-token').addEventListener('click', () => openTokenModal({ allowCancel: true }));
    $('#logout').addEventListener('click', () => {
      localStorage.removeItem('adminToken');
-      tokenInput.value = '';
-      showError('Token cleared; reload the page after supplying a new token.');
+      openTokenModal({
+        error: 'Token cleared. Enter a new token to continue.',
+        allowCancel: false
+      });
    });

+    $('#token-form').addEventListener('submit', saveTokenFromModal);
+    $('#modal-cancel').addEventListener('click', closeTokenModal);
+
    $('#cluster-form').addEventListener('submit', async (e) => {
      e.preventDefault();
      showError('');
@@ -395,7 +480,11 @@
      }
    });

-    loadClusters();
+    if (!localStorage.getItem('adminToken')) {
+      openTokenModal({ allowCancel: false });
+    } else {
+      loadClusters();
+    }
  </script>
 </body>
 </html>
@@ -1,12 +1,12 @@
 package logging

 import (
-	"log/slog"
 	"context"
 	"crypto/rand"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -41,6 +41,7 @@ func shortHexID(n int) (string, error) {

 type toolCallLogger struct {
 	file      *os.File
+	base      *slog.Logger
 	toolName  string
 	startedAt time.Time
 	resultRaw json.RawMessage
@@ -54,6 +55,12 @@ func (t *toolCallLogger) WithResult(result any) {

 func (t *toolCallLogger) WithError(err error) {
 	t.err = err
+	if t.base != nil && err != nil {
+		t.base.Error("tool.call.error",
+			"tool", t.toolName,
+			"err", err.Error(),
+		)
+	}
 }

 func (t *toolCallLogger) End() {
@@ -82,6 +89,14 @@ func (t *toolCallLogger) End() {
 		}
 		_, _ = t.file.Write(b)
 		_, _ = t.file.WriteString("\n")
+
+		if t.base != nil {
+			t.base.Info("tool.call.end",
+				"tool", t.toolName,
+				"duration_ms", durationMs,
+				"error", errStr,
+			)
+		}
 	})
 }

@@ -145,6 +160,7 @@ func StartToolCall(ctx context.Context, base *slog.Logger, toolName string, para

 	return &toolCallLogger{
 		file:      f,
+		base:      base,
 		toolName:  toolName,
 		startedAt: startedAt,
 	}, nil
@@ -0,0 +1,195 @@
+package logging
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+// newCapturingLogger returns a slog.Logger that writes JSON to buf and a
+// pointer to buf so tests can assert on what was emitted.
+func newCapturingLogger() (*slog.Logger, *bytes.Buffer) {
+	buf := &bytes.Buffer{}
+	return slog.New(slog.NewJSONHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})), buf
+}
+
+// initLogDir sets the package-level logDir for StartToolCall to discover.
+// Tests run in a temp directory; tear down via t.Cleanup.
+func initLogDir(t *testing.T) string {
+	t.Helper()
+	dir := t.TempDir()
+	logDirMu.Lock()
+	prev := logDir
+	logDir = dir
+	logDirMu.Unlock()
+	t.Cleanup(func() {
+		logDirMu.Lock()
+		logDir = prev
+		logDirMu.Unlock()
+	})
+	return dir
+}
+
+// TestStartToolCall_EmitsErrorAndEnd verifies that WithError and End each
+// emit a structured event through the main base logger, in addition to
+// writing the per-tool file.  This is what makes "tail -f spark-mcp.log"
+// useful for debugging tool-call failures.
+func TestStartToolCall_EmitsErrorAndEnd(t *testing.T) {
+	dir := initLogDir(t)
+
+	base, buf := newCapturingLogger()
+
+	cl, err := StartToolCall(context.Background(), base, "list_applications", map[string]any{"cluster_id": "x"})
+	if err != nil {
+		t.Fatalf("StartToolCall: %v", err)
+	}
+	if cl == nil {
+		t.Fatal("nil logger")
+	}
+
+	cl.WithError(os.ErrNotExist)
+	cl.End()
+
+	// Per-tool file is still written (regression check).
+	entries, err := os.ReadDir(filepath.Join(dir, "tools"))
+	if err != nil {
+		t.Fatalf("read tools dir: %v", err)
+	}
+	if len(entries) == 0 {
+		t.Fatalf("no per-tool log file written")
+	}
+
+	// Decode each JSON line in the captured buffer.
+	var sawError, sawEnd bool
+	for line := range strings.SplitSeq(strings.TrimSpace(buf.String()), "\n") {
+		if line == "" {
+			continue
+		}
+		var rec map[string]any
+		if err := json.Unmarshal([]byte(line), &rec); err != nil {
+			t.Fatalf("invalid JSON in log: %q (err=%v)", line, err)
+		}
+		switch rec["msg"] {
+		case "tool.call.error":
+			sawError = true
+			if rec["level"] != "ERROR" {
+				t.Errorf("tool.call.error level=%v, want ERROR", rec["level"])
+			}
+			if rec["tool"] != "list_applications" {
+				t.Errorf("tool.call.error tool=%v, want list_applications", rec["tool"])
+			}
+			if !strings.Contains(rec["err"].(string), "file does not exist") {
+				t.Errorf("tool.call.error err=%v, missing underlying err text", rec["err"])
+			}
+		case "tool.call.end":
+			sawEnd = true
+			if rec["level"] != "INFO" {
+				t.Errorf("tool.call.end level=%v, want INFO", rec["level"])
+			}
+			if rec["tool"] != "list_applications" {
+				t.Errorf("tool.call.end tool=%v, want list_applications", rec["tool"])
+			}
+			if _, ok := rec["duration_ms"].(float64); !ok {
+				t.Errorf("tool.call.end missing numeric duration_ms: %v", rec)
+			}
+			if rec["error"] != "file does not exist" {
+				t.Errorf("tool.call.end error=%v, want file does not exist", rec["error"])
+			}
+		}
+	}
+
+	if !sawError {
+		t.Errorf("expected tool.call.error in main logger; got: %s", buf.String())
+	}
+	if !sawEnd {
+		t.Errorf("expected tool.call.end in main logger; got: %s", buf.String())
+	}
+}
+
+// TestStartToolCall_NilBaseSafe verifies the noop-friendly path: if base is
+// nil (e.g. tests where deps.Logger is unset), WithError and End must not
+// panic and the per-tool file must still be written.
+func TestStartToolCall_NilBaseSafe(t *testing.T) {
+	dir := initLogDir(t)
+
+	cl, err := StartToolCall(context.Background(), nil, "noop_tool", map[string]any{"k": "v"})
+	if err != nil {
+		t.Fatalf("StartToolCall: %v", err)
+	}
+	// Should not panic.
+	cl.WithError(os.ErrPermission)
+	cl.End()
+
+	// Per-tool file still written.
+	entries, err := os.ReadDir(filepath.Join(dir, "tools"))
+	if err != nil {
+		t.Fatalf("read tools dir: %v", err)
+	}
+	if len(entries) == 0 {
+		t.Errorf("expected per-tool file even with nil base")
+	}
+}
+
+// TestStartToolCall_SuccessfulCall_EmitsEndOnly verifies that a happy-path
+// call emits exactly one end event (no error event), with an empty error
+// string and a non-negative duration.
+func TestStartToolCall_SuccessfulCall_EmitsEndOnly(t *testing.T) {
+	_ = initLogDir(t)
+
+	base, buf := newCapturingLogger()
+	cl, err := StartToolCall(context.Background(), base, "list_clusters", nil)
+	if err != nil {
+		t.Fatalf("StartToolCall: %v", err)
+	}
+	cl.End()
+
+	lines := strings.SplitSeq(strings.TrimSpace(buf.String()), "\n")
+	// Exactly one event: tool.call.start (from StartToolCall itself).
+	// No tool.call.error, and one tool.call.end.
+	var endCount, errorCount int
+	for line := range lines {
+		if line == "" {
+			continue
+		}
+		var rec map[string]any
+		if err := json.Unmarshal([]byte(line), &rec); err != nil {
+			t.Fatalf("invalid JSON: %q", line)
+		}
+		switch rec["msg"] {
+		case "tool.call.end":
+			endCount++
+			if rec["error"] != "" {
+				t.Errorf("successful end should have empty error; got %v", rec["error"])
+			}
+		case "tool.call.error":
+			errorCount++
+		}
+	}
+	if errorCount != 0 {
+		t.Errorf("expected 0 tool.call.error, got %d; buf=%s", errorCount, buf.String())
+	}
+	if endCount != 1 {
+		t.Errorf("expected exactly 1 tool.call.end, got %d; buf=%s", endCount, buf.String())
+	}
+	// Sanity: duration_ms is non-negative.
+	for line := range lines {
+		var rec map[string]any
+		if err := json.Unmarshal([]byte(line), &rec); err == nil {
+			if msg, _ := rec["msg"].(string); msg == "tool.call.end" {
+				if d, ok := rec["duration_ms"].(float64); !ok || d < 0 {
+					t.Errorf("duration_ms invalid: %v", rec["duration_ms"])
+				}
+			}
+		}
+	}
+}
+
+// ensure time is referenced even if no test uses it directly (avoids unused
+// import churn when tests are edited).
+var _ = time.Now
@@ -15,7 +15,7 @@ const defaultTailBytes = 1 << 20
 // NewGetApplicationLogsTool returns the schema for the get_application_logs MCP Tool.
 func NewGetApplicationLogsTool() mcp.Tool {
 	return mcp.NewTool(GetApplicationLogsName,
-		mcp.WithDescription("Fetch YARN application logs. Tries amContainerLogs first, then aggregated-logs, then the legacy /logs endpoint. Returns {source, text}."),
+		mcp.WithDescription("Fetch YARN application logs. Tries amContainerLogs (RM endpoint, often 307 to a NodeManager) first, then aggregated-logs, then the legacy /logs endpoint, and finally the AM container's direct NodeManager URL parsed from the app response. Returns {source, text}."),
 		mcp.WithString("cluster_id",
 			mcp.Required(),
 			mcp.Description("ID of the configured cluster (from list_clusters)"),
@@ -13,7 +13,7 @@ const GetApplicationStatusName = "get_application_status"
 // NewGetApplicationStatusTool returns the schema for the get_application_status MCP Tool.
 func NewGetApplicationStatusTool() mcp.Tool {
 	return mcp.NewTool(GetApplicationStatusName,
-		mcp.WithDescription("Get the detailed status of a single YARN application from the ResourceManager. Returns the raw RM JSON for the app."),
+		mcp.WithDescription("Get the detailed status of a single YARN application from the ResourceManager. Returns the raw RM JSON for the app, including the amContainerLogs field (a direct NodeManager URL usable with fetch_url as a last-resort log source if all get_application_logs fallback endpoints fail)."),
 		mcp.WithString("cluster_id",
 			mcp.Required(),
 			mcp.Description("ID of the configured cluster (from list_clusters)"),
@@ -24,6 +24,13 @@ func NewListApplicationsTool() mcp.Tool {
 		mcp.WithString("user",
 			mcp.Description("Filter by submitting user."),
 		),
+		mcp.WithString("queue",
+			mcp.Description("Filter by YARN queue name. Matches the YARN RM ?queue= parameter."),
+		),
+		mcp.WithNumber("limit",
+			mcp.Description("Maximum number of applications to return. Defaults to 100 if not provided; set to 0 to request the RM default (no limit)."),
+			mcp.DefaultNumber(100),
+		),
 	)
 }

@@ -43,11 +50,24 @@ func (d *Deps) ListApplicationsHandler(ctx context.Context, req mcp.CallToolRequ
 	if v, ok := args["user"].(string); ok {
 		user = v
 	}
+	queue := ""
+	if v, ok := args["queue"].(string); ok {
+		queue = v
+	}
+	limit := 0
+	if v, ok := args["limit"].(float64); ok {
+		limit = int(v)
+	}
+	if limit < 0 {
+		limit = 0
+	}

 	callLog := startToolCall(ctx, d.Logger, ListApplicationsName, map[string]any{
 		"cluster_id": clusterID,
 		"state":      state,
 		"user":       user,
+		"queue":      queue,
+		"limit":      limit,
 	})
 	defer callLog.End()

@@ -58,7 +78,7 @@ func (d *Deps) ListApplicationsHandler(ctx context.Context, req mcp.CallToolRequ
 	}

 	rmc := rm.New(d.HTTPClient, cl)
-	raw, err := rmc.ListApps(ctx, state, user)
+	raw, err := rmc.ListApps(ctx, state, user, queue, limit)
 	if err != nil {
 		callLog.WithError(err)
 		return errResult("list_applications: " + err.Error()), nil
@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"

 	"spark-mcp-go/internal/cluster"
@@ -24,8 +25,8 @@ func New(hc *httpclient.Client, c *cluster.Cluster) *Client {
 	return &Client{hc: hc, cluster: c}
 }

-// ListApps returns raw JSON from GET /ws/v1/cluster/apps[?state=&user=].
-func (r *Client) ListApps(ctx context.Context, state, user string) (json.RawMessage, error) {
+// ListApps returns raw JSON from GET /ws/v1/cluster/apps[?state=&user=&queue=&limit=].
+func (r *Client) ListApps(ctx context.Context, state, user, queue string, limit int) (json.RawMessage, error) {
 	base := r.cluster.RMURL + "/ws/v1/cluster/apps"
 	u, err := url.Parse(base)
 	if err != nil {
@@ -38,6 +39,12 @@ func (r *Client) ListApps(ctx context.Context, state, user string) (json.RawMess
 	if user != "" {
 		q.Set("user", user)
 	}
+	if queue != "" {
+		q.Set("queue", queue)
+	}
+	if limit > 0 {
+		q.Set("limit", strconv.Itoa(limit))
+	}
 	u.RawQuery = q.Encode()
 	return r.do(ctx, http.MethodGet, u.String(), nil)
 }
@@ -55,9 +62,10 @@ func (r *Client) KillApp(ctx context.Context, appID string) (json.RawMessage, er
 }

 // GetLogs walks a fallback chain to retrieve YARN application logs:
-//  1. GET /ws/v1/cluster/apps/{id}/amContainerLogs (RM often 307 to a NodeManager)
+//  1. GET /ws/v1/cluster/apps/{id}/amContainerLogs (RM endpoint, often 307 to a NodeManager)
 //  2. GET /ws/v1/cluster/apps/{id}/aggregated-logs?container=...
 //  3. GET /ws/v1/cluster/apps/{id}/logs
+//  4. GET the NodeManager URL from the app's amContainerLogs field.
 //
 // The first successful endpoint wins.  The returned source names the endpoint
 // that produced the body.
@@ -69,12 +77,31 @@ func (r *Client) GetLogs(ctx context.Context, appID, container string) (body []b
 	}
 	var lastErr error
 	for _, ep := range endpoints {
-		b, e := r.doBytes(ctx, http.MethodGet, r.cluster.RMURL+ep.path, nil)
+		b, e := r.doBytesWithHosts(ctx, http.MethodGet, r.cluster.RMURL+ep.path, nil)
 		if e == nil {
 			return b, ep.name, nil
 		}
 		lastErr = e
 	}
+	// Try fetching the NodeManager's stdout URL directly.
+	app, err := r.GetApp(ctx, appID)
+	if err == nil {
+		var resp struct {
+			App struct {
+				AMContainerLogs string `json:"amContainerLogs"`
+			} `json:"app"`
+		}
+		if err := json.Unmarshal(app, &resp); err == nil && resp.App.AMContainerLogs != "" {
+			nmHost := extractHost(resp.App.AMContainerLogs)
+			b, e := r.doBytesWithHosts(ctx, http.MethodGet, resp.App.AMContainerLogs, nil, nmHost)
+			if e == nil {
+				return b, "amContainerLogs-direct", nil
+			}
+			lastErr = e
+		}
+	} else {
+		lastErr = err
+	}
 	return nil, "", fmt.Errorf("rm: all log endpoints failed: %w", lastErr)
 }

@@ -92,7 +119,7 @@ func (r *Client) do(ctx context.Context, method, rawURL string, body io.Reader)
 	return raw, nil
 }

-func (r *Client) doBytes(ctx context.Context, method, rawURL string, body io.Reader) ([]byte, error) {
+func (r *Client) doBytesWithHosts(ctx context.Context, method, rawURL string, body io.Reader, extraHosts ...string) ([]byte, error) {
 	req, err := http.NewRequestWithContext(ctx, method, rawURL, body)
 	if err != nil {
 		return nil, err
@@ -100,7 +127,7 @@ func (r *Client) doBytes(ctx context.Context, method, rawURL string, body io.Rea
 	if err := httpclient.ApplyAuth(req, "", r.cluster); err != nil {
 		return nil, err
 	}
-	resp, err := r.hc.DoWithRedirect(ctx, req, 5, r.allowedHosts()...)
+	resp, err := r.hc.DoWithRedirect(ctx, req, 5, append(r.allowedHosts(), extraHosts...)...)
 	if err != nil {
 		return nil, err
 	}
@@ -111,6 +138,10 @@ func (r *Client) doBytes(ctx context.Context, method, rawURL string, body io.Rea
 	return io.ReadAll(resp.Body)
 }

+func (r *Client) doBytes(ctx context.Context, method, rawURL string, body io.Reader) ([]byte, error) {
+	return r.doBytesWithHosts(ctx, method, rawURL, body)
+}
+
 func (r *Client) allowedHosts() []string {
 	hosts := []string{extractHost(r.cluster.RMURL)}
 	for _, pat := range r.cluster.URLAllowlist {
@@ -3,6 +3,7 @@ package rm
 import (
 	"context"
 	"encoding/base64"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -30,7 +31,7 @@ func TestListApps(t *testing.T) {
 	defer srv.Close()

 	cl := &cluster.Cluster{RMURL: srv.URL}
-	raw, err := testClient(srv, cl).ListApps(context.Background(), "", "")
+	raw, err := testClient(srv, cl).ListApps(context.Background(), "", "", "", 0)
 	if err != nil {
 		t.Fatalf("ListApps: %v", err)
 	}
@@ -54,7 +55,7 @@ func TestListApps_WithQuery(t *testing.T) {
 	defer srv.Close()

 	cl := &cluster.Cluster{RMURL: srv.URL}
-	raw, err := testClient(srv, cl).ListApps(context.Background(), "RUNNING", "yarn")
+	raw, err := testClient(srv, cl).ListApps(context.Background(), "RUNNING", "yarn", "", 0)
 	if err != nil {
 		t.Fatalf("ListApps: %v", err)
 	}
@@ -212,7 +213,7 @@ func TestApplyAuth_SimpleUserName(t *testing.T) {
 	defer srv.Close()

 	cl := &cluster.Cluster{RMURL: srv.URL, AuthType: cluster.AuthSimple, AuthUsername: "alice"}
-	_, err := testClient(srv, cl).ListApps(context.Background(), "", "")
+	_, err := testClient(srv, cl).ListApps(context.Background(), "", "", "", 0)
 	if err != nil {
 		t.Fatalf("ListApps: %v", err)
 	}
@@ -227,7 +228,7 @@ func TestApplyAuth_BasicHeader(t *testing.T) {
 	defer srv.Close()

 	cl := &cluster.Cluster{RMURL: srv.URL, AuthType: cluster.AuthBasic, AuthUsername: "u", AuthPassword: "p"}
-	_, err := testClient(srv, cl).ListApps(context.Background(), "", "")
+	_, err := testClient(srv, cl).ListApps(context.Background(), "", "", "", 0)
 	if err != nil {
 		t.Fatalf("ListApps: %v", err)
 	}
@@ -236,3 +237,114 @@ func TestApplyAuth_BasicHeader(t *testing.T) {
 		t.Errorf("Authorization=%q, want %q", captured, want)
 	}
 }
+
+func TestListApps_WithLimit(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Query().Get("limit") != "50" {
+			t.Errorf("limit=%q, want 50", r.URL.Query().Get("limit"))
+		}
+		w.Write([]byte(`{"apps":{"app":[]}}`))
+	}))
+	defer srv.Close()
+	cl := &cluster.Cluster{RMURL: srv.URL}
+	_, err := testClient(srv, cl).ListApps(context.Background(), "", "", "", 50)
+	if err != nil {
+		t.Fatalf("ListApps: %v", err)
+	}
+}
+
+func TestListApps_WithQueue(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Query().Get("queue") != "root.users.alice" {
+			t.Errorf("queue=%q, want root.users.alice", r.URL.Query().Get("queue"))
+		}
+		w.Write([]byte(`{"apps":{"app":[]}}`))
+	}))
+	defer srv.Close()
+	cl := &cluster.Cluster{RMURL: srv.URL}
+	_, err := testClient(srv, cl).ListApps(context.Background(), "", "", "root.users.alice", 0)
+	if err != nil {
+		t.Fatalf("ListApps: %v", err)
+	}
+}
+
+func TestListApps_LimitZeroNoParam(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Query().Get("limit") != "" {
+			t.Errorf("limit should not be sent when 0; got %q", r.URL.Query().Get("limit"))
+		}
+		w.Write([]byte(`{"apps":{"app":[]}}`))
+	}))
+	defer srv.Close()
+	cl := &cluster.Cluster{RMURL: srv.URL}
+	_, err := testClient(srv, cl).ListApps(context.Background(), "", "", "", 0)
+	if err != nil {
+		t.Fatalf("ListApps: %v", err)
+	}
+}
+
+func TestGetLogs_FallbackToAMDirect(t *testing.T) {
+	// Mock NM that we want GetLogs to hit after standard endpoints fail.
+	nmSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("nm direct logs"))
+	}))
+	defer nmSrv.Close()
+	nmHost := nmSrv.Listener.Addr().String()
+
+	rmSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123/amContainerLogs":
+			w.WriteHeader(http.StatusInternalServerError)
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123/aggregated-logs":
+			w.WriteHeader(http.StatusNotImplemented)
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123/logs":
+			w.WriteHeader(http.StatusNotFound)
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123" && r.Method == http.MethodGet:
+			// GetApp returns amContainerLogs pointing to the mock NM.
+			w.Write([]byte(fmt.Sprintf(`{"app":{"id":"app_123","amContainerLogs":"%s/node/containerlogs/container_1/root"}}`, nmSrv.URL)))
+		default:
+			t.Errorf("unexpected RM path %q", r.URL.Path)
+		}
+	}))
+	defer rmSrv.Close()
+
+	cl := &cluster.Cluster{
+		RMURL:        rmSrv.URL,
+		URLAllowlist: []string{nmHost}, // allow NM host for SSRF
+	}
+	body, source, err := testClient(rmSrv, cl).GetLogs(context.Background(), "app_123", "container_1")
+	if err != nil {
+		t.Fatalf("GetLogs: %v", err)
+	}
+	if source != "amContainerLogs-direct" {
+		t.Errorf("source=%q, want amContainerLogs-direct", source)
+	}
+	if string(body) != "nm direct logs" {
+		t.Errorf("body=%q", string(body))
+	}
+}
+
+func TestGetLogs_FallbackFailsWhenAMFieldMissing(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123/amContainerLogs":
+			w.WriteHeader(http.StatusInternalServerError)
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123/aggregated-logs":
+			w.WriteHeader(http.StatusNotImplemented)
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123/logs":
+			w.WriteHeader(http.StatusNotFound)
+		case r.URL.Path == "/ws/v1/cluster/apps/app_123" && r.Method == http.MethodGet:
+			// amContainerLogs is empty
+			w.Write([]byte(`{"app":{"id":"app_123"}}`))
+		default:
+			t.Errorf("unexpected path %q", r.URL.Path)
+		}
+	}))
+	defer srv.Close()
+
+	cl := &cluster.Cluster{RMURL: srv.URL}
+	_, _, err := testClient(srv, cl).GetLogs(context.Background(), "app_123", "container_1")
+	if err == nil {
+		t.Fatal("expected error when all endpoints fail and amContainerLogs is empty")
+	}
+}
Author	SHA1	Message	Date
tao.chenandClaude	1a7c692e79	gitignore: 跨平台 build 产物 spark-mcp-- 覆盖 GOOS-GOARCH 命名约定的所有二进制变体 (spark-mcp-linux-amd64, spark-mcp-darwin-arm64 等). 之前每次交叉编译都会留一个未跟踪的二进制在仓库根. Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-13 15:28:01 +08:00
tao.chenandClaude	9bcfe558cf	logging: tool.call.error/end 接入主 logger (双写终端 + spark-mcp.log) 之前所有 tool 调用的 error 和 end 事件只写进 per-tool 文件 (data/logs/tools/<ts>_<tool>_<id>.log), 主日志文件 spark-mcp.log 完全看不到。tail -f spark-mcp.log 排查问题时形同虚设, 必须去找具体的 per-tool 文件才能看到错误。修复: toolCallLogger struct 加 base *slog.Logger 字段, StartToolCall 把传入的 base 写进 struct. WithError 在记录错误的同时调 base.Error ("tool.call.error", tool, name, err), End 在写完 per-tool 文件后调 base.Info("tool.call.end", tool, name, duration_ms, error). base 走的是 main logger (multiHandler 包装), 自动同时写终端 (遵循 LOG_FORMAT, 默认 text) 和主日志文件 (永远 JSON). LOG_LEVEL 过滤对这两条事件仍生效, 行为统一. 新增 3 个单元测试 (internal/logging/tool_call_test.go): - TestStartToolCall_EmitsErrorAndEnd: 验证错误路径两条事件都进主 logger (用 bytes.Buffer 捕获 slog 输出, 解码 JSON 校验字段) - TestStartToolCall_NilBaseSafe: 验证 base=nil (测试场景) 不 panic 且 per-tool 文件仍写入 - TestStartToolCall_SuccessfulCall_EmitsEndOnly: 验证成功路径只发 end 不发 error, duration_ms >= 0 附带: gofmt 重排 internal/logging/tool_call.go import 块 (按字母序) 未提交: spark-mcp-linux-amd64 (本地 build 产物, .gitignore 未拦, 需另决定); .env (gitignored 内容) Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-13 15:27:28 +08:00
tao.chenandClaude	2537976722	rm: list_applications 加 limit/queue + 日志 fallback 链补 NM 直连 P0 (Python parity 缺口): - rm.Client.ListApps 新增 queue + limit 查询参数 - list_applications MCP tool schema 暴露 queue (string) 和 limit (number, default 100), handler 透传给 RM client - 限流原因: 生产 RM 上无 limit 会拉回 N MB JSON 撞 MaxResponseBytes - limit <= 0 不发送 query 参数, 跟 RM 默认行为一致; 老的调用方 (TestListApplications_EndToEnd 等) 不传参时行为不变 - 4 个旧 ListApps 测试调用点跟着更新, 加 3 个新测试: TestListApps_WithLimit, TestListApps_WithQueue, TestListApps_LimitZeroNoParam P1.3 (Python fallback 行为补全): - rm.Client.GetLogs 加第 4 步 fallback: 当前 3 步 (amContainerLogs/aggregated-logs/logs) 全失败时, 调 GetApp 解析 app.amContainerLogs 字段, GET 该 URL 直连 NodeManager - 新 source 名 'amContainerLogs-direct' 区分 RM-level endpoint - 提取 doBytesWithHosts(extraHosts...) 支持 per-call host 白名单, NM host 动态加进 allowedHosts, SSRF 保护不破 (URL 是 RM 响应里回来的, 不是 LLM 任意填的) - 新增 TestGetLogs_FallbackToAMDirect (成功路径) 和 TestGetLogs_FallbackFailsWhenAMFieldMissing (amContainerLogs 字段为空时正常返回 error) P1.4 (零代码改动 + 文档化): - get_application_logs / get_application_status 工具 description 更新, 提示 LLM raw RM JSON 里包含 amContainerLogs 字段, 可在 aggregated-logs 全部失败时自己用 fetch_url 直连 NM - get_application_status 函数体不变 (本就是透传 raw JSON) 验证: - go build / go vet / go test 全部通过 - 5 个新测试全 PASS, 老的 TestListApplications_EndToEnd 仍 PASS - 单 ListApps 调用点 (list_applications.go:81) 编译过未提交: spark-mcp-linux-amd64 (本地 build 产物, 当前 .gitignore 没拦, 需另行决定) Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-13 15:15:21 +08:00
tao.chenandClaude	bf2640d8fd	admin: /admin HTML 公开可访问，让 token 弹窗真正能触发上一轮 (`90eba22`) 把 admin 页改成 token 弹窗模式，但 webHandler 还挂在带 AdminAuth 的 group 里。结果无 token 访问 /admin 拿到的不是 HTML，而是 {"error":"missing bearer token"} 的 401 JSON，弹窗变成死代码。修复: 拆出 gPublic group，只放 webHandler (GET /admin 和 /admin/)。 API 路由 (/clusters*, /audit) 和 OpenAPI docs (/docs, /docs/spec) 继续受 AdminAuth 保护 (后者暴露内部 API 结构，不该公网可见)。 HTML 自身的 token 输入流不变 — 弹窗仍然在客户端拦截 401、清旧 token、记录 pendingRetry、保存新 token 后自动重试。补一个 TestAdminWeb_NoTokenReturnsHTML 锁定新契约: - GET /admin 无 token -> 200 + text/html (含 token-modal 元素) - GET /admin/ 同上 - GET /admin/clusters 无 token -> 401 (回归保护) - GET /admin/clusters good-token -> 200 (回归保护) 旧 TestMount_RequiresAuth 测的全是 API 路径, 不受影响, 全绿。验证: - go build / go vet / go test 全部通过 - 端到端 curl 复现用户报告场景: /admin 无 token -> 200 + 弹窗 HTML - API + docs 鉴权行为不变 Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-13 14:16:49 +08:00
tao.chenandClaude	90eba221d4	admin: token 输入改为弹窗模式 (modal) 把 header 内嵌的 token 输入框 + Save 按钮拆掉，改用原生 <dialog> 弹窗。三种触发场景: 首访无 token 自动弹、API 401 自动弹 (清旧 token 后弹带错误信息)、用户主动点 Set Token 按钮弹。无有效 token 时阻止 Escape 关闭 (cancel 事件 preventDefault), 防止误操作丢失输入。 localStorage 键名 adminToken 不变, 零迁移成本。401 路径在 api() 内部清 token + 弹窗 + 记录 pendingRetry 闭包, 用户保存新 token 后自动重试失败请求。设计取舍: - 选原生 <dialog> 而非自造 div + JS, 自带 modal/背板/焦点陷阱/a11y - 选 <form> + preventDefault 而非 method='dialog', 避免自动关闭 - CSS 全部复用现有 dark theme 变量, 0 视觉割裂 - 0 新依赖 (纯 HTML5 + 原生 JS) 验证: - go build / go vet / go test 全部通过 - 端到端 curl 确认新元素嵌入、旧元素删除、401/200 路径正常 - router_test.go 无回归 (HTML 改动不涉及 Go) Co-Authored-By: Claude <noreply@anthropic.com>	2026-07-13 14:08:16 +08:00