From d6e3952a4b4874a427f3254538f81dd45b053e85 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:45:01 +0800 Subject: [PATCH] executor: parse tracking_url and error on missing application_id Co-Authored-By: Claude --- internal/executor/spark_submit.go | 89 +++++++++++++++----- internal/executor/spark_submit_parse_test.go | 79 +++++++++++++++++ internal/executor/spark_submit_test.go | 22 +++-- 3 files changed, 160 insertions(+), 30 deletions(-) create mode 100644 internal/executor/spark_submit_parse_test.go diff --git a/internal/executor/spark_submit.go b/internal/executor/spark_submit.go index b4fd4eb..f5b7b09 100644 --- a/internal/executor/spark_submit.go +++ b/internal/executor/spark_submit.go @@ -23,10 +23,13 @@ import ( // application application_xxx" line and a few progress lines. const stdoutCap = 10 << 20 -// appIDPattern matches the YARN line printed on successful submission: -// -// "Submitted application application_12345_0001" -var appIDPattern = regexp.MustCompile(`application_\d+_\d+`) +// submittedAppPattern matches the explicit success line, including the +// application id captured in group 1. This mirrors the Python reference. +var submittedAppPattern = regexp.MustCompile(`Submitted application (\S+)`) + +// trackingURLPattern matches "tracking URL: " on stderr. The URL tail +// is used as a fallback source for the application id. +var trackingURLPattern = regexp.MustCompile(`tracking URL:\s+(\S+)`) // SparkSubmitOpts is the resolved command line for a single spark-submit run. // @@ -43,11 +46,12 @@ type SparkSubmitOpts struct { // Result is what spark_submit returns to the LLM. type Result struct { - AppID string `json:"app_id,omitempty"` - ExitCode int `json:"exit_code"` - StdoutTail string `json:"stdout_tail"` - StderrTail string `json:"stderr_tail"` - DurationMS int64 `json:"duration_ms"` + AppID string `json:"app_id,omitempty"` + TrackingURL string `json:"tracking_url,omitempty"` + ExitCode int `json:"exit_code"` + StdoutTail string `json:"stdout_tail"` + StderrTail string `json:"stderr_tail"` + DurationMS int64 `json:"duration_ms"` } // ErrBinaryEmpty is returned when the cluster has no spark-submit binary @@ -90,27 +94,70 @@ func Run(ctx context.Context, opts SparkSubmitOpts) (*Result, error) { DurationMS: dur.Milliseconds(), } - // app_id: prefer stdout, fall back to stderr. Both are searched only on - // the tail (cap-bounded) so we don't pay for the full output. - if id := matchAppID(res.StdoutTail); id != "" { - res.AppID = id - } else if id := matchAppID(res.StderrTail); id != "" { - res.AppID = id - } + appID, trackingURL, ok := parseSparkSubmitOutput(res.StdoutTail, res.StderrTail) + res.AppID = appID + res.TrackingURL = trackingURL + if !ok { + return res, fmt.Errorf("executor: spark-submit ran but no application_id in output (exit_code=%d)", res.ExitCode) + } if err != nil { - // Wrap the underlying error but keep the parsed result so the LLM - // can see exit_code and stderr even on failure. return res, fmt.Errorf("executor: spark-submit failed: %w", err) } return res, nil } -func matchAppID(s string) string { - if s == "" { +// parseSparkSubmitOutput extracts the application id and tracking URL from +// spark-submit output. It mirrors the Python reference implementation: first +// look for "Submitted application ", then fall back to "tracking URL:", +// deriving the id from the last path segment. The returned ok is false when +// no application id could be determined. +func parseSparkSubmitOutput(stdout, stderr string) (appID, trackingURL string, ok bool) { + // Stage 1: explicit success line, prefer stdout then stderr. + var trackingFound bool + if m := submittedAppPattern.FindStringSubmatch(stdout); m != nil { + appID = m[1] + ok = true + } else if m := submittedAppPattern.FindStringSubmatch(stderr); m != nil { + appID = m[1] + ok = true + } + + // Stage 2: tracking URL fallback (stderr only, like the Python reference). + if !ok { + if m := trackingURLPattern.FindStringSubmatch(stderr); m != nil { + trackingURL = m[1] + trackingFound = true + appID = extractAppIDFromTrackingURL(trackingURL) + ok = appID != "" + if !ok { + trackingURL = "" + } + } + } + + // Tracking URL is independent: if present anywhere on stderr, capture it. + if trackingURL == "" && !trackingFound { + if m := trackingURLPattern.FindStringSubmatch(stderr); m != nil { + trackingURL = m[1] + } + } + return +} + +// extractAppIDFromTrackingURL returns the last path segment if it starts with +// "application_", otherwise an empty string. +func extractAppIDFromTrackingURL(rawURL string) string { + trimmed := strings.TrimRight(rawURL, "/") + idx := strings.LastIndex(trimmed, "/") + if idx < 0 { return "" } - return appIDPattern.FindString(s) + tail := trimmed[idx+1:] + if strings.HasPrefix(tail, "application_") { + return tail + } + return "" } // exitCodeFromErr returns 0 on nil, the process's exit code on *exec.ExitError, diff --git a/internal/executor/spark_submit_parse_test.go b/internal/executor/spark_submit_parse_test.go new file mode 100644 index 0000000..8a8bd73 --- /dev/null +++ b/internal/executor/spark_submit_parse_test.go @@ -0,0 +1,79 @@ +package executor + +import "testing" + +func TestParseAppID_FromStdout(t *testing.T) { + stdout := "some prefix\nSubmitted application application_123_456\n" + stderr := "random log\n" + appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr) + if !ok { + t.Fatal("expected ok") + } + if appID != "application_123_456" { + t.Errorf("appID=%q, want application_123_456", appID) + } + if trackingURL != "" { + t.Errorf("trackingURL=%q, want empty", trackingURL) + } +} + +func TestParseAppID_FromStderr(t *testing.T) { + stdout := "" + stderr := "Submitted application application_123_456\n" + appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr) + if !ok { + t.Fatal("expected ok") + } + if appID != "application_123_456" { + t.Errorf("appID=%q, want application_123_456", appID) + } + if trackingURL != "" { + t.Errorf("trackingURL=%q, want empty", trackingURL) + } +} + +func TestParseAppID_FromTrackingURL(t *testing.T) { + stdout := "" + stderr := "Log line\ntracking URL: http://rm.example.com:8088/proxy/application_123_456/\n" + appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr) + if !ok { + t.Fatal("expected ok") + } + if appID != "application_123_456" { + t.Errorf("appID=%q, want application_123_456", appID) + } + wantURL := "http://rm.example.com:8088/proxy/application_123_456/" + if trackingURL != wantURL { + t.Errorf("trackingURL=%q, want %q", trackingURL, wantURL) + } +} + +func TestParseAppID_NoAppID(t *testing.T) { + stdout := "" + stderr := "some random log" + appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr) + if ok { + t.Fatal("expected not ok") + } + if appID != "" { + t.Errorf("appID=%q, want empty", appID) + } + if trackingURL != "" { + t.Errorf("trackingURL=%q, want empty", trackingURL) + } +} + +func TestParseAppID_BadTrackingURL(t *testing.T) { + stdout := "" + stderr := "tracking URL: http://rm.example.com:8088/proxy/not-an-app/\n" + appID, trackingURL, ok := parseSparkSubmitOutput(stdout, stderr) + if ok { + t.Fatal("expected not ok") + } + if appID != "" { + t.Errorf("appID=%q, want empty", appID) + } + if trackingURL != "" { + t.Errorf("trackingURL=%q, want empty", trackingURL) + } +} diff --git a/internal/executor/spark_submit_test.go b/internal/executor/spark_submit_test.go index eff7939..a464b29 100644 --- a/internal/executor/spark_submit_test.go +++ b/internal/executor/spark_submit_test.go @@ -30,7 +30,7 @@ func TestRun_FakeBinary(t *testing.T) { res, err := Run(ctx, SparkSubmitOpts{ Binary: "/bin/echo", - Args: []string{"--master", "yarn; rm -rf /tmp/this-should-not-exist"}, + Args: []string{"--master", "yarn; rm -rf /tmp/this-should-not-exist", "Submitted application application_12345_0001"}, Timeout: 3 * time.Second, }) if err != nil { @@ -43,8 +43,8 @@ func TestRun_FakeBinary(t *testing.T) { if !strings.Contains(res.StdoutTail, "yarn; rm -rf /tmp/this-should-not-exist") { t.Errorf("stdout missing literal arg, got: %q", res.StdoutTail) } - if res.AppID != "" { - t.Errorf("AppID should be empty for echo, got %q", res.AppID) + if res.AppID != "application_12345_0001" { + t.Errorf("AppID = %q, want application_12345_0001", res.AppID) } } @@ -66,22 +66,26 @@ func TestRun_NonZeroExit(t *testing.T) { } } -// TestMatchAppID locks down the regex so a YARN output tweak doesn't +// TestSubmittedAppPattern locks down the regex so a YARN output tweak doesn't // silently break app_id extraction. -func TestMatchAppID(t *testing.T) { +func TestSubmittedAppPattern(t *testing.T) { cases := []struct { in, want string }{ {"Submitted application application_12345_0001", "application_12345_0001"}, - {"... some prefix application_999_42 ... tail", "application_999_42"}, + {"Submitted application application_999_42 extra", "application_999_42"}, {"no app id here", ""}, {"", ""}, - {"application_1_2 application_3_4", "application_1_2"}, // first match + {"Submitted application app_1_2 Submitted application app_3_4", "app_1_2"}, // first match } for _, tc := range cases { - got := matchAppID(tc.in) + m := submittedAppPattern.FindStringSubmatch(tc.in) + var got string + if m != nil { + got = m[1] + } if got != tc.want { - t.Errorf("matchAppID(%q) = %q, want %q", tc.in, got, tc.want) + t.Errorf("submittedAppPattern(%q) = %q, want %q", tc.in, got, tc.want) } } }