diff --git a/spark_executor/core/yarn_client.py b/spark_executor/core/yarn_client.py index 7073b90..391084d 100644 --- a/spark_executor/core/yarn_client.py +++ b/spark_executor/core/yarn_client.py @@ -14,9 +14,12 @@ Endpoints used (YARN 2.6+): When aggregated logs are unavailable (HTTP 404/501, e.g. log aggregation disabled), fall back to the amContainerLogs field reported by the app -endpoint and fetch the AM (driver) container's /stdout directly from the -NodeManager. This returns driver logs only; executor container logs still -require yarn.log-aggregation-enable=true (the primary /aggregated-logs path). +endpoint and fetch the AM (driver) container's /stdout from the +NodeManager. Some clusters (Knox-proxied, wrapped YARN builds) wrap +/stdout in an HTML directory listing; in that case we fetch the bare +{amContainerLogs} URL, parse the listing, and pull each log file. This +returns driver logs only; executor container logs still require +yarn.log-aggregation-enable=true (the primary /aggregated-logs path). The ResourceManager URL is passed in per call (snapshotted on the Job at confirm_submit_job time) and falls back to the YARN_RESOURCE_MANAGER_URL env @@ -26,6 +29,8 @@ field was designed for, but no longer requires the `yarn` CLI to interpret it. import json from dataclasses import dataclass +import re + import httpx import httpx_kerberos @@ -163,11 +168,50 @@ def _logs_unavailable_error(application_id: str) -> YarnError: ) +def _looks_like_html(text: str) -> bool: + """Cheap heuristic: True if the body starts with an HTML/XHTML prolog. + + Intentionally lightweight (no full parser) — used only to decide whether + a NodeManager log response is raw text or an HTML directory listing. + """ + head = text.lstrip()[:200].lower() + return head.startswith((" list[str]: + """Extract plain log-file names from a NodeManager directory-listing page. + + Keeps simple filenames (stdout, stderr, syslog, prelaunch.err, + launch_container.sh, ...); drops parent-dir links, absolute URLs, and + sort-toggle query strings. Returns hrefs in document order, deduplicated. + """ + hrefs = re.findall(r'href="([^"]*)"', html, re.IGNORECASE) + seen: set[str] = set() + out: list[str] = [] + for href in hrefs: + if href in ("../", "/") or href.startswith("/"): + continue + if "://" in href or "?" in href: + continue + if href in seen: + continue + seen.add(href) + out.append(href) + return out + + def _fetch_logs_via_am_container(application_id: str, config: YarnClientConfig) -> str: """ - Fetch ApplicationMaster (driver) container stdout as a fallback when the + Fetch ApplicationMaster (driver) container logs as a fallback when the aggregated-logs endpoint is unavailable. + Fast path: GET {amContainerLogs}/stdout and return its text directly when + the NodeManager serves raw log bytes (modern YARN). + + Fallback path: when /stdout returns an HTML directory listing (Knox / + wrapped YARN), GET the bare {amContainerLogs} URL, parse the listing for + log-file names, and fetch each one individually. + Returns AM (driver) container logs only. Executor container logs require yarn.log-aggregation-enable=true, which is covered by the primary /aggregated-logs path. @@ -183,12 +227,30 @@ def _fetch_logs_via_am_container(application_id: str, config: YarnClientConfig) if not am_container_logs: raise _logs_unavailable_error(application_id) - log_url = f"{am_container_logs.rstrip('/')}/stdout" - resp = _request("GET", log_url, timeout=60.0, verify=config.verify_for_httpx(), auth=config.auth_for_httpx()) - if resp.status_code >= 400: - logger.error(f"YARN GET {log_url} -> {resp.status_code}: {resp.text[:500]}") - raise _logs_unavailable_error(application_id) - return resp.text + am_container_logs = am_container_logs.rstrip("/") + verify = config.verify_for_httpx() + auth = config.auth_for_httpx() + + # Fast path: modern YARN serves raw log bytes at /stdout. + stdout_url = f"{am_container_logs}/stdout" + resp = _request("GET", stdout_url, timeout=60.0, verify=verify, auth=auth) + if resp.status_code < 400 and not _looks_like_html(resp.text): + return resp.text + + # Fallback path: Knox / wrapped YARN serves an HTML directory listing for + # every URL on the NodeManager log endpoint. Parse it and fetch each file. + resp = _request("GET", am_container_logs, timeout=60.0, verify=verify, auth=auth) + if resp.status_code < 400: + parts: list[str] = [] + for filename in _parse_log_listing_html(resp.text): + file_url = f"{am_container_logs}/{filename}" + file_resp = _request("GET", file_url, timeout=60.0, verify=verify, auth=auth) + if file_resp.status_code < 400 and not _looks_like_html(file_resp.text): + parts.append(f"=== {filename} ===\n{file_resp.text}") + if parts: + return "\n\n".join(parts) + + raise _logs_unavailable_error(application_id) def get_application_logs(application_id: str, config: YarnClientConfig) -> str: diff --git a/tests/unit/test_yarn_client.py b/tests/unit/test_yarn_client.py index 6379e42..1807fa8 100644 --- a/tests/unit/test_yarn_client.py +++ b/tests/unit/test_yarn_client.py @@ -12,6 +12,7 @@ from spark_executor.core.yarn_client import ( YarnConfigError, YarnError, YarnClientConfig, + _parse_log_listing_html, get_application_logs, get_application_status, kill_application, @@ -201,6 +202,139 @@ def test_logs_5xx_on_aggregated_endpoint_raises_immediately(): assert m.call_count == 1 +def test_logs_falls_back_to_directory_listing_when_stdout_returns_html(): + """Knox / wrapped YARN: GET {amContainerLogs}/stdout returns an HTML + directory listing instead of raw log bytes. The fallback should fetch the + bare URL, parse the listing, and pull each log file individually.""" + listing = ( + "" + 'Parent Directory' + 'Name' + 'stdout' + 'stderr' + 'syslog' + "" + ) + html_stdout = ( + "wrapper around /stdout endpoint" + ) + responses = [ + _resp(404), # aggregated-logs missing + _resp( + 200, + json_data={ + "app": { + "amContainerLogs": "http://nm1:8042/node/containerlogs/container_1/hdfs", + } + }, + ), + _resp(200, text=html_stdout), # /stdout -> HTML wrapper + _resp(200, text=listing), # bare URL -> directory listing + _resp(200, text="driver stdout line\n"), # fetch stdout + _resp(200, text="driver stderr line\n"), # fetch stderr + _resp(200, text="syslog line\n"), # fetch syslog + ] + with patch( + "spark_executor.core.yarn_client.httpx.request", side_effect=responses + ) as m: + out = get_application_logs("application_1", YarnClientConfig(yarn_rm_url=RM)) + + # Header + body for each file in document order. + assert "=== stdout ===" in out + assert "=== stderr ===" in out + assert "=== syslog ===" in out + assert "driver stdout line" in out + assert "driver stderr line" in out + assert "syslog line" in out + # stdout must come before stderr (document order preserved). + assert out.index("=== stdout ===") < out.index("=== stderr ===") < out.index("=== syslog ===") + + # 7 HTTP calls: aggregated-logs, app, /stdout (HTML), bare URL, stdout file, stderr file, syslog file + assert m.call_count == 7 + assert m.call_args_list[2].args == ( + "GET", + "http://nm1:8042/node/containerlogs/container_1/hdfs/stdout", + ) + assert m.call_args_list[3].args == ( + "GET", + "http://nm1:8042/node/containerlogs/container_1/hdfs", + ) + assert m.call_args_list[4].args == ( + "GET", + "http://nm1:8042/node/containerlogs/container_1/hdfs/stdout", + ) + assert m.call_args_list[5].args == ( + "GET", + "http://nm1:8042/node/containerlogs/container_1/hdfs/stderr", + ) + assert m.call_args_list[6].args == ( + "GET", + "http://nm1:8042/node/containerlogs/container_1/hdfs/syslog", + ) + + +def test_logs_html_filter_strips_sort_links_and_parent_dir(): + """_parse_log_listing_html must drop ../, sort-toggle query strings, and + absolute URLs, while keeping plain log-file names.""" + html = ( + 'Parent' + 'Name' + 'Last modified' + 'Size' + 'Description' + 'absolute' + 'rooted' + 'stdout' + 'stderr' + 'syslog' + 'stdout-dup' # duplicate + ) + assert _parse_log_listing_html(html) == ["stdout", "stderr", "syslog"] + + +def test_logs_html_filter_handles_prelaunch_files(): + """Some YARN versions list prelaunch.out / prelaunch.err / launch_container.sh + in the directory listing alongside stdout/stderr. Keep them — we want to + surface every log file the container produced.""" + html = ( + 'prelaunch.out' + 'prelaunch.err' + 'launch_container.sh' + 'directory-info' + ) + assert _parse_log_listing_html(html) == [ + "prelaunch.out", + "prelaunch.err", + "launch_container.sh", + "directory-info", + ] + + +def test_logs_raises_when_directory_listing_empty(): + """If /stdout returns HTML AND the directory listing has no parseable log + file names, raise _logs_unavailable_error.""" + html_stdout = "wrapper" + empty_listing = 'Parent' + responses = [ + _resp(404), # aggregated-logs missing + _resp( + 200, + json_data={ + "app": { + "amContainerLogs": "http://nm1:8042/node/containerlogs/container_1/hdfs", + } + }, + ), + _resp(200, text=html_stdout), # /stdout -> HTML + _resp(200, text=empty_listing), # bare URL -> only parent link + ] + with patch( + "spark_executor.core.yarn_client.httpx.request", side_effect=responses + ): + with pytest.raises(YarnError, match="log-aggregation-enable"): + get_application_logs("application_1", YarnClientConfig(yarn_rm_url=RM)) + + # --- kill_application --- def test_kill_sends_put_with_killed_state():