fix(yarn_client): manually follow 3xx Location redirects on log fetch

httpx follows redirects by default, but it does NOT forward the
Authorization header across host boundaries. In YARN deployments where
the ResourceManager 307-redirects the NodeManager log fetch to a
different host (load balancer, Knox, NM selection), the follow-up
request lands unauthenticated and returns 401/403.

Replace the _request call in _fetch_logs_via_am_container with
_request_following_redirects, which:
  - Walks the 3xx Location chain (up to 3 hops).
  - Re-applies auth + verify on every hop.
  - Resolves relative Location URLs against the current request URL.
  - Raises YarnError on 3xx-without-Location (misconfigured server) and
    on hop count overflow (redirect loop protection).

301/302/303/307/308 are all treated as 'follow the Location' per
RFC 7231 — the method/body handling for the rest is up to httpx when
we re-issue the request.

Tests: 4 new (cross-host 307, no-Location 307, two-hop chain, relative
Location). Full suite 247 passed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/unit/test_yarn_client.py

@@ -43,10 +43,13 @@ def _restore_settings():
     config.settings.ssl_ca_bundle_default = snapshot.ssl_ca_bundle_default
 
 
-def _resp(status: int, *, json_data=None, text: str | None = None) -> httpx.Response:
+def _resp(status: int, *, json_data=None, text: str | None = None, headers: dict | None = None) -> httpx.Response:
+    kwargs: dict = {}
+    if headers:
+        kwargs["headers"] = headers
     if json_data is not None:
-        return httpx.Response(status, json=json_data)
-    return httpx.Response(status, text=text or "")
+        return httpx.Response(status, json=json_data, **kwargs)
+    return httpx.Response(status, text=text or "", **kwargs)
 
 
 # --- get_application_status ---
@@ -191,6 +194,110 @@ def test_logs_returns_am_container_stdout():
     )
 
 
+def test_logs_follows_307_redirect_to_other_host():
+    """RM 307-redirects the log fetch to the actual NodeManager host. The
+    helper must follow Location and re-apply auth/verify on the new host."""
+    responses = [
+        _resp(404),  # aggregated-logs missing
+        _resp(
+            200,
+            json_data={
+                "app": {
+                    "amContainerLogs": "http://rm:8088/node/containerlogs/container_1/hdfs",
+                }
+            },
+        ),
+        _resp(307, headers={"Location": "http://nm2:8042/node/containerlogs/container_1/hdfs/stdout"}),
+        _resp(200, text="driver stdout via redirect"),
+    ]
+    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))
+    assert out == "driver stdout via redirect"
+    calls = m.call_args_list
+    assert calls[2].args == ("GET", "http://rm:8088/node/containerlogs/container_1/hdfs/stdout")
+    assert calls[3].args == ("GET", "http://nm2:8042/node/containerlogs/container_1/hdfs/stdout")
+    # Auth must be re-applied on the redirected request.
+    assert calls[3].kwargs["auth"] == calls[2].kwargs["auth"]
+
+
+def test_logs_307_without_location_raises_unavailable():
+    """3xx without a Location header is a misconfigured server — surface as
+    YarnError rather than silently returning empty text."""
+    responses = [
+        _resp(404),  # aggregated-logs missing
+        _resp(
+            200,
+            json_data={
+                "app": {
+                    "amContainerLogs": "http://rm:8088/node/containerlogs/container_1/hdfs",
+                }
+            },
+        ),
+        _resp(307),  # no Location header
+    ]
+    with patch(
+        "spark_executor.core.yarn_client.httpx.request", side_effect=responses
+    ):
+        with pytest.raises(YarnError, match="no Location header"):
+            get_application_logs("application_1", YarnClientConfig(yarn_rm_url=RM))
+
+
+def test_logs_follows_redirect_chain_then_200():
+    """Two consecutive 307s followed by a 200 — the helper should walk the
+    chain and return the final body."""
+    responses = [
+        _resp(404),  # aggregated-logs missing
+        _resp(
+            200,
+            json_data={
+                "app": {
+                    "amContainerLogs": "http://lb:8088/node/containerlogs/container_1/hdfs",
+                }
+            },
+        ),
+        _resp(307, headers={"Location": "http://rm:8088/node/containerlogs/container_1/hdfs/stdout"}),
+        _resp(307, headers={"Location": "http://nm2:8042/node/containerlogs/container_1/hdfs/stdout"}),
+        _resp(200, text="chained redirect body"),
+    ]
+    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))
+    assert out == "chained redirect body"
+    urls = [c.args[1] for c in m.call_args_list[2:]]
+    assert urls == [
+        "http://lb:8088/node/containerlogs/container_1/hdfs/stdout",
+        "http://rm:8088/node/containerlogs/container_1/hdfs/stdout",
+        "http://nm2:8042/node/containerlogs/container_1/hdfs/stdout",
+    ]
+
+
+def test_logs_relative_location_resolved_against_request_url():
+    """A relative Location header (e.g. '/redirected/stdout') must be
+    resolved against the request URL, not treated as an absolute path."""
+    responses = [
+        _resp(404),  # aggregated-logs missing
+        _resp(
+            200,
+            json_data={
+                "app": {
+                    "amContainerLogs": "http://nm1:8042/node/containerlogs/container_1/hdfs",
+                }
+            },
+        ),
+        _resp(307, headers={"Location": "/redirected/stdout"}),
+        _resp(200, text="resolved body"),
+    ]
+    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))
+    assert out == "resolved body"
+    assert m.call_args_list[3].args == ("GET", "http://nm1:8042/redirected/stdout")
+
+
 def test_logs_5xx_on_aggregated_endpoint_raises_immediately():
     with patch(
         "spark_executor.core.yarn_client.httpx.request",