diff --git a/spark_executor/core/yarn_client.py b/spark_executor/core/yarn_client.py index 7ce94f4..8636bc2 100644 --- a/spark_executor/core/yarn_client.py +++ b/spark_executor/core/yarn_client.py @@ -83,16 +83,73 @@ def get_application_status(application_id: str, yarn_rm_url: str | None) -> tupl return state, json.dumps(data, indent=2) +def _logs_unavailable_error(application_id: str) -> YarnError: + """Consistent error when logs cannot be retrieved from either path.""" + return YarnError( + f"YARN aggregated logs not available for {application_id!r}. " + f"The application may not be in FINISHED state, or " + f"yarn.log-aggregation-enable is false on the cluster." + ) + + +def _fetch_logs_via_nodemanager(application_id: str, yarn_rm_url: str | None) -> str: + """ + Hadoop 2.x fallback: walk app attempts -> containers -> NodeManager + container logs and concatenate the results. + """ + base = _base_url(yarn_rm_url) + app_url = f"{base}/ws/v1/cluster/apps/{application_id}" + + # Latest (or all) app attempts. + attempts_url = f"{app_url}/appattempts" + resp = _request("GET", attempts_url, timeout=30.0) + if resp.status_code >= 400: + raise _logs_unavailable_error(application_id) + attempts = resp.json().get("appAttempts", {}).get("appAttempt", []) + if not attempts: + raise _logs_unavailable_error(application_id) + + # Collect containers across attempts. + containers: list[dict] = [] + for attempt in attempts: + attempt_id = attempt.get("id") + if not attempt_id: + continue + containers_url = f"{app_url}/appattempts/{attempt_id}/containers" + resp = _request("GET", containers_url, timeout=30.0) + if resp.status_code >= 400: + raise _logs_unavailable_error(application_id) + containers.extend(resp.json().get("containers", {}).get("container", [])) + + if not containers: + raise _logs_unavailable_error(application_id) + + parts = [] + for container in containers: + container_id = container.get("id") + user = container.get("user") + node_http_address = container.get("nodeHttpAddress") + if not all((container_id, user, node_http_address)): + continue + nm_url = ( + f"http://{node_http_address}/node/containerlogs/{container_id}/{user}/" + ) + resp = _request("GET", nm_url, timeout=60.0) + if resp.status_code >= 400: + raise _logs_unavailable_error(application_id) + parts.append(f"=== container: {container_id} ===\n{resp.text}") + + if not parts: + raise _logs_unavailable_error(application_id) + return "\n\n".join(parts) + + def get_application_logs(application_id: str, yarn_rm_url: str | None) -> str: """Return aggregated container logs for an application as text.""" url = f"{_base_url(yarn_rm_url)}/ws/v1/cluster/apps/{application_id}/aggregated-logs" resp = _request("GET", url, timeout=60.0) - if resp.status_code == 404: - raise YarnError( - f"YARN aggregated logs not available for {application_id!r}. " - f"The application may not be in FINISHED state, or " - f"yarn.log-aggregation-enable is false on the cluster." - ) + if resp.status_code in (404, 501): + return _fetch_logs_via_nodemanager(application_id, yarn_rm_url) if resp.status_code >= 400: logger.error(f"YARN GET {url} -> {resp.status_code}: {resp.text[:500]}") raise YarnError(f"YARN GET logs returned HTTP {resp.status_code}") diff --git a/tests/unit/test_yarn_client.py b/tests/unit/test_yarn_client.py index 6a170c3..f340795 100644 --- a/tests/unit/test_yarn_client.py +++ b/tests/unit/test_yarn_client.py @@ -96,27 +96,123 @@ def test_status_rejects_non_http_url(): # --- get_application_logs --- -def test_logs_returns_text(): +def test_logs_returns_text_on_h3_aggregated_endpoint(): fake = _resp(200, text="log line 1\nlog line 2\n") with patch("spark_executor.core.yarn_client.httpx.request", return_value=fake) as m: out = get_application_logs("application_1", RM) assert out == "log line 1\nlog line 2\n" - assert m.call_args.args == ("GET", f"{RM}/ws/v1/cluster/apps/application_1/aggregated-logs") + assert m.call_args.args == ( + "GET", + f"{RM}/ws/v1/cluster/apps/application_1/aggregated-logs", + ) -def test_logs_raises_on_404_with_explanation(): - with patch("spark_executor.core.yarn_client.httpx.request", return_value=_resp(404)): +def test_logs_falls_back_to_nodemanager_on_404(): + responses = [ + _resp(404), # aggregated-logs H3 endpoint missing + _resp(200, json_data={"appAttempts": {"appAttempt": [{"id": "attempt_1"}]}}), + _resp( + 200, + json_data={ + "containers": { + "container": [ + { + "id": "container_1", + "user": "hdfs", + "nodeHttpAddress": "nm1:8042", + } + ] + } + }, + ), + _resp(200, text="container log content"), + ] + with patch( + "spark_executor.core.yarn_client.httpx.request", side_effect=responses + ) as m: + out = get_application_logs("application_1", RM) + assert "container_1" in out + assert "container log content" in out + calls = m.call_args_list + assert calls[0].args == ( + "GET", + f"{RM}/ws/v1/cluster/apps/application_1/aggregated-logs", + ) + assert calls[1].args == ("GET", f"{RM}/ws/v1/cluster/apps/application_1/appattempts") + assert calls[2].args == ( + "GET", + f"{RM}/ws/v1/cluster/apps/application_1/appattempts/attempt_1/containers", + ) + assert calls[3].args == ( + "GET", + "http://nm1:8042/node/containerlogs/container_1/hdfs/", + ) + + +def test_logs_raises_on_404_when_nm_also_empty(): + responses = [ + _resp(404), # aggregated-logs H3 endpoint missing + _resp(200, json_data={"appAttempts": {"appAttempt": []}}), + ] + 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", RM) -def test_logs_raises_on_5xx(): +def test_logs_handles_multiple_containers(): + responses = [ + _resp(404), # aggregated-logs H3 endpoint missing + _resp(200, json_data={"appAttempts": {"appAttempt": [{"id": "attempt_1"}]}}), + _resp( + 200, + json_data={ + "containers": { + "container": [ + { + "id": "container_1", + "user": "hdfs", + "nodeHttpAddress": "nm1:8042", + }, + { + "id": "container_2", + "user": "hdfs", + "nodeHttpAddress": "nm2:8042", + }, + ] + } + }, + ), + _resp(200, text="log one"), + _resp(200, text="log two"), + ] + with patch( + "spark_executor.core.yarn_client.httpx.request", side_effect=responses + ) as m: + out = get_application_logs("application_1", RM) + assert "log one" in out + assert "log two" in out + assert out.index("log one") < out.index("log two") + calls = m.call_args_list + assert calls[3].args == ( + "GET", + "http://nm1:8042/node/containerlogs/container_1/hdfs/", + ) + assert calls[4].args == ( + "GET", + "http://nm2:8042/node/containerlogs/container_2/hdfs/", + ) + + +def test_logs_5xx_on_aggregated_endpoint_raises_immediately(): with patch( "spark_executor.core.yarn_client.httpx.request", return_value=_resp(500, text="boom"), - ): - with pytest.raises(YarnError): + ) as m: + with pytest.raises(YarnError, match="500"): get_application_logs("application_1", RM) + assert m.call_count == 1 # --- kill_application ---