mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Fix proxy endpoint: improve timeout and error handling
- Use streaming responses (client.send with stream=True) to avoid buffering entire response body before proxying - Replace generic Exception with UpstreamProxyError and GatewayTimeoutException using proper HTTP status codes - Include exception type and URL in error messages for better diagnostics Fixes https://github.com/galaxyproject/galaxy/issues/22293
This commit is contained in:
@@ -308,6 +308,16 @@ class ServerNotConfiguredForRequest(MessageException):
|
||||
err_code = error_codes_by_name["SERVER_NOT_CONFIGURED_FOR_REQUEST"]
|
||||
|
||||
|
||||
class UpstreamProxyError(MessageException):
|
||||
status_code = 502
|
||||
err_code = error_codes_by_name["UPSTREAM_PROXY_ERROR"]
|
||||
|
||||
|
||||
class GatewayTimeoutException(MessageException):
|
||||
status_code = 504
|
||||
err_code = error_codes_by_name["UPSTREAM_PROXY_TIMEOUT"]
|
||||
|
||||
|
||||
class HandlerAssignmentError(Exception):
|
||||
def __init__(self, msg=None, obj=None, **kwargs):
|
||||
super().__init__(msg)
|
||||
|
||||
@@ -218,5 +218,15 @@
|
||||
"name": "SERVER_NOT_CONFIGURED_FOR_REQUEST",
|
||||
"code": 501002,
|
||||
"message": "Server not configured for the request. The Galaxy admin may be able to resolve the problem by installing additional dependencies or setting up new infrastructure."
|
||||
},
|
||||
{
|
||||
"name": "UPSTREAM_PROXY_ERROR",
|
||||
"code": 502001,
|
||||
"message": "An error occurred while proxying a request to an upstream server."
|
||||
},
|
||||
{
|
||||
"name": "UPSTREAM_PROXY_TIMEOUT",
|
||||
"code": 504001,
|
||||
"message": "The upstream server did not respond in time."
|
||||
}
|
||||
]
|
||||
|
||||
@@ -3,6 +3,7 @@ API Controller to proxy remote files.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
from functools import partial
|
||||
from urllib.parse import (
|
||||
urljoin,
|
||||
@@ -21,7 +22,9 @@ from starlette.responses import (
|
||||
)
|
||||
|
||||
from galaxy.exceptions import (
|
||||
GatewayTimeoutException,
|
||||
RequestParameterInvalidException,
|
||||
UpstreamProxyError,
|
||||
UserRequiredException,
|
||||
)
|
||||
from galaxy.files.uris import validate_uri_access
|
||||
@@ -42,6 +45,8 @@ URLQueryParam: str = Query(
|
||||
|
||||
ALLOWED_SCHEMES = ("https", "http")
|
||||
MAX_REDIRECTS = 5
|
||||
MAX_STREAM_BYTES = 1 * 1024 * 1024 # 1 MB
|
||||
MAX_STREAM_SECONDS = 10
|
||||
|
||||
|
||||
def is_valid_url(url: str) -> bool:
|
||||
@@ -83,7 +88,9 @@ class FastAPIProxy:
|
||||
# This is to prevent the server from hanging indefinitely
|
||||
timeout = httpx.Timeout(10.0, connect=60.0)
|
||||
|
||||
response = None
|
||||
client = httpx.AsyncClient(timeout=timeout)
|
||||
streaming = False
|
||||
try:
|
||||
response = await self._handle_redirects_validation(request, url, trans, headers, client)
|
||||
|
||||
@@ -93,13 +100,32 @@ class FastAPIProxy:
|
||||
|
||||
async def stream_with_cleanup():
|
||||
"""Stream response chunks and ensure cleanup on completion or error."""
|
||||
total_bytes = 0
|
||||
start_time = time.monotonic()
|
||||
try:
|
||||
async for chunk in response.aiter_bytes():
|
||||
total_bytes += len(chunk)
|
||||
if total_bytes > MAX_STREAM_BYTES:
|
||||
log.warning(
|
||||
"Proxy stream to %s exceeded max size of %d bytes",
|
||||
url,
|
||||
MAX_STREAM_BYTES,
|
||||
)
|
||||
break
|
||||
elapsed = time.monotonic() - start_time
|
||||
if elapsed > MAX_STREAM_SECONDS:
|
||||
log.warning(
|
||||
"Proxy stream to %s exceeded max time of %d seconds",
|
||||
url,
|
||||
MAX_STREAM_SECONDS,
|
||||
)
|
||||
break
|
||||
yield chunk
|
||||
finally:
|
||||
await response.aclose()
|
||||
await client.aclose()
|
||||
|
||||
streaming = True
|
||||
# StreamingResponse will handle chunked transfer encoding automatically
|
||||
return StreamingResponse(
|
||||
stream_with_cleanup(),
|
||||
@@ -116,11 +142,13 @@ class FastAPIProxy:
|
||||
except httpx.InvalidURL as e:
|
||||
# Catch any URL validation errors that slip through our pre-validation
|
||||
raise RequestParameterInvalidException(f"Invalid URL format: {e}")
|
||||
except httpx.TimeoutException as e:
|
||||
raise GatewayTimeoutException(f"Timeout proxying request to {url}: {type(e).__name__}")
|
||||
except httpx.RequestError as e:
|
||||
raise Exception(f"Request error: {e}")
|
||||
raise UpstreamProxyError(f"Error proxying request to {url}: {type(e).__name__}: {e}")
|
||||
finally:
|
||||
# Only cleanup for non-GET requests (GET cleanup happens in the stream generator)
|
||||
if request.method != "GET":
|
||||
# Only cleanup if we're NOT handing off to the stream generator
|
||||
if not streaming:
|
||||
if response is not None:
|
||||
await response.aclose()
|
||||
await client.aclose()
|
||||
@@ -139,9 +167,12 @@ class FastAPIProxy:
|
||||
redirect_count = 0
|
||||
|
||||
while redirect_count <= MAX_REDIRECTS:
|
||||
response = await client.request(
|
||||
method=request.method, url=current_url, headers=headers, follow_redirects=False
|
||||
req = client.build_request(
|
||||
method=request.method,
|
||||
url=current_url,
|
||||
headers=headers,
|
||||
)
|
||||
response = await client.send(req, follow_redirects=False, stream=True)
|
||||
|
||||
if self._is_redirect_response(response):
|
||||
redirect_count += 1
|
||||
|
||||
@@ -148,7 +148,8 @@ class TestProxyApi(ApiTestCase):
|
||||
|
||||
# Setup mock client
|
||||
mock_client = MagicMock()
|
||||
mock_client.request = AsyncMock(return_value=redirect_response)
|
||||
mock_client.build_request = MagicMock(return_value=MagicMock())
|
||||
mock_client.send = AsyncMock(return_value=redirect_response)
|
||||
mock_client.aclose = AsyncMock()
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
@@ -181,7 +182,8 @@ class TestProxyApi(ApiTestCase):
|
||||
|
||||
# Setup mock client to return redirect first, then final response
|
||||
mock_client = MagicMock()
|
||||
mock_client.request = AsyncMock(side_effect=[redirect_response, final_response])
|
||||
mock_client.build_request = MagicMock(return_value=MagicMock())
|
||||
mock_client.send = AsyncMock(side_effect=[redirect_response, final_response])
|
||||
mock_client.aclose = AsyncMock()
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
@@ -203,7 +205,8 @@ class TestProxyApi(ApiTestCase):
|
||||
|
||||
# Setup mock client
|
||||
mock_client = MagicMock()
|
||||
mock_client.request = AsyncMock(return_value=redirect_response)
|
||||
mock_client.build_request = MagicMock(return_value=MagicMock())
|
||||
mock_client.send = AsyncMock(return_value=redirect_response)
|
||||
mock_client.aclose = AsyncMock()
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
|
||||
Reference in New Issue
Block a user