diff --git a/roar/cli/commands/proxy.py b/roar/cli/commands/proxy.py index edab3628..b46312ba 100644 --- a/roar/cli/commands/proxy.py +++ b/roar/cli/commands/proxy.py @@ -82,8 +82,9 @@ def proxy_disable() -> None: def proxy_start() -> None: """Start a standalone S3 proxy daemon. - The daemon runs in the background. Use its port with - AWS_ENDPOINT_URL to route S3 traffic through the proxy. + The daemon runs in the background. Point S3 clients at its port with + AWS_ENDPOINT_URL_S3 (SDKs) / S3_ENDPOINT_URL (s5cmd) to route S3 traffic + through the proxy. """ from ...execution.cluster.proxy import ProxyService from ...integrations.config import get_roar_dir @@ -104,7 +105,10 @@ def proxy_start() -> None: click.echo(f"Proxy daemon started (pid={info['pid']}, port={info['port']}).") click.echo("") click.echo("To use it:") - click.echo(f" export AWS_ENDPOINT_URL=http://127.0.0.1:{info['port']}") + click.echo( + f" export AWS_ENDPOINT_URL_S3=http://127.0.0.1:{info['port']} # boto3, aws cli" + ) + click.echo(f" export S3_ENDPOINT_URL=http://127.0.0.1:{info['port']} # s5cmd") except RuntimeError as e: raise click.ClickException(str(e)) from e diff --git a/roar/execution/runtime/proxy_resource.py b/roar/execution/runtime/proxy_resource.py index 7fed3a6c..95c4dd47 100644 --- a/roar/execution/runtime/proxy_resource.py +++ b/roar/execution/runtime/proxy_resource.py @@ -49,7 +49,20 @@ def active(self) -> bool: def start(self, ctx: RunContext, environ: Mapping[str, str]) -> RuntimeResourceStart: del ctx - existing_endpoint = str(environ.get("AWS_ENDPOINT_URL") or "").strip() or None + # Detect a pre-existing S3 endpoint (e.g. MinIO/LocalStack) to chain to. + # Prefer the S3-scoped vars we also inject below, but fall back to the + # generic AWS_ENDPOINT_URL — that is what most MinIO/LocalStack users + # actually set, so ignoring it here would leave the proxy with no upstream + # and drop their S3 traffic on the floor. + existing_endpoint = ( + str( + environ.get("AWS_ENDPOINT_URL_S3") + or environ.get("S3_ENDPOINT_URL") + or environ.get("AWS_ENDPOINT_URL") + or "" + ).strip() + or None + ) try: self._handle = self._service.start_for_run(upstream_url=existing_endpoint) except Exception as exc: @@ -57,8 +70,15 @@ def start(self, ctx: RunContext, environ: Mapping[str, str]) -> RuntimeResourceS self.logger.warning("Failed to start proxy: %s", exc) return RuntimeResourceStart() + proxy_url = f"http://127.0.0.1:{self._handle.port}" + # Point S3 clients at the proxy with the S3-*scoped* endpoint override, + # not the service-agnostic AWS_ENDPOINT_URL. AWS_ENDPOINT_URL_S3 is + # honored by boto3 / aws CLI / aws-sdk-go-v2 and, unlike the global var, + # does not redirect non-S3 services (STS, EC2, CloudWatch, ...) into the + # S3-only proxy. S3_ENDPOINT_URL covers s5cmd, which ignores the SDK vars. env = { - "AWS_ENDPOINT_URL": f"http://127.0.0.1:{self._handle.port}", + "AWS_ENDPOINT_URL_S3": proxy_url, + "S3_ENDPOINT_URL": proxy_url, } if existing_endpoint: env["ROAR_UPSTREAM_S3_ENDPOINT"] = existing_endpoint diff --git a/roar/execution/runtime/tracer.py b/roar/execution/runtime/tracer.py index e2d62e55..462c1615 100644 --- a/roar/execution/runtime/tracer.py +++ b/roar/execution/runtime/tracer.py @@ -528,7 +528,7 @@ def execute( except Exception: pass # Best-effort - # Merge extra env (e.g. AWS_ENDPOINT_URL from proxy) + # Merge extra env (e.g. AWS_ENDPOINT_URL_S3 / S3_ENDPOINT_URL from proxy) if extra_env: env.update(extra_env) diff --git a/tests/integration/test_proxy_integration.py b/tests/integration/test_proxy_integration.py index 7544694a..9e61151e 100644 --- a/tests/integration/test_proxy_integration.py +++ b/tests/integration/test_proxy_integration.py @@ -70,7 +70,7 @@ def test_proxy_captures_s3_put_as_job_output( script = proxy_repo / "s3_put.py" script.write_text( "import os, urllib.request\n" - 'endpoint = os.environ["AWS_ENDPOINT_URL"]\n' + 'endpoint = os.environ["AWS_ENDPOINT_URL_S3"]\n' "req = urllib.request.Request(\n" ' f"{endpoint}/test-bucket/output.csv",\n' ' data=b"result_data",\n' @@ -130,7 +130,7 @@ def test_proxy_captures_s3_get_as_job_input( script = proxy_repo / "s3_get.py" script.write_text( "import os, urllib.request\n" - 'endpoint = os.environ["AWS_ENDPOINT_URL"]\n' + 'endpoint = os.environ["AWS_ENDPOINT_URL_S3"]\n' 'urllib.request.urlopen(f"{endpoint}/test-bucket/input.csv")\n' ) git_commit("add s3 get script") @@ -184,7 +184,7 @@ def test_proxy_captures_mixed_get_and_put( script = proxy_repo / "s3_mix.py" script.write_text( "import os, urllib.request\n" - 'endpoint = os.environ["AWS_ENDPOINT_URL"]\n' + 'endpoint = os.environ["AWS_ENDPOINT_URL_S3"]\n' "# Read input\n" 'urllib.request.urlopen(f"{endpoint}/mix-bucket/src.csv")\n' "# Write output\n" @@ -255,7 +255,7 @@ def test_proxy_chains_existing_aws_endpoint_url( script = proxy_repo / "s3_chain.py" script.write_text( "import os, urllib.request\n" - 'endpoint = os.environ["AWS_ENDPOINT_URL"]\n' + 'endpoint = os.environ["AWS_ENDPOINT_URL_S3"]\n' "req = urllib.request.Request(\n" ' f"{endpoint}/chain-test/data.csv",\n' ' data=b"chain_payload",\n' @@ -296,7 +296,7 @@ def test_proxy_etags_match_server_response( script = proxy_repo / "s3_etag.py" script.write_text( "import os, urllib.request\n" - 'endpoint = os.environ["AWS_ENDPOINT_URL"]\n' + 'endpoint = os.environ["AWS_ENDPOINT_URL_S3"]\n' 'urllib.request.urlopen(f"{endpoint}/etag-bucket/verify.dat")\n' ) git_commit("add etag verify script") @@ -402,7 +402,7 @@ def test_proxy_captures_operations_on_script_failure( script = proxy_repo / "s3_fail.py" script.write_text( "import os, sys, urllib.request\n" - 'endpoint = os.environ["AWS_ENDPOINT_URL"]\n' + 'endpoint = os.environ["AWS_ENDPOINT_URL_S3"]\n' "req = urllib.request.Request(\n" ' f"{endpoint}/fail-bucket/before_crash.csv",\n' ' data=b"partial_output",\n' diff --git a/tests/unit/test_proxy_runtime_resource.py b/tests/unit/test_proxy_runtime_resource.py index 2e6216db..706601cb 100644 --- a/tests/unit/test_proxy_runtime_resource.py +++ b/tests/unit/test_proxy_runtime_resource.py @@ -23,22 +23,53 @@ def test_init_raises_when_proxy_binary_is_missing(): ProxyRuntimeResource(service=service) -def test_start_passes_existing_aws_endpoint_as_upstream(ctx): +def test_start_passes_existing_s3_endpoint_as_upstream(ctx): service = MagicMock() service.find_proxy.return_value = "/tmp/roar-proxy" service.start_for_run.return_value = ProxyHandle(process=MagicMock(), port=9090) resource = ProxyRuntimeResource(service=service) - result = resource.start(ctx, {"AWS_ENDPOINT_URL": "http://localhost:4566"}) + result = resource.start(ctx, {"AWS_ENDPOINT_URL_S3": "http://localhost:4566"}) service.start_for_run.assert_called_once_with(upstream_url="http://localhost:4566") assert result.env == { - "AWS_ENDPOINT_URL": "http://127.0.0.1:9090", + "AWS_ENDPOINT_URL_S3": "http://127.0.0.1:9090", + "S3_ENDPOINT_URL": "http://127.0.0.1:9090", "ROAR_UPSTREAM_S3_ENDPOINT": "http://localhost:4566", } assert resource.active is True +def test_start_detects_s5cmd_endpoint_var_as_upstream(ctx): + service = MagicMock() + service.find_proxy.return_value = "/tmp/roar-proxy" + service.start_for_run.return_value = ProxyHandle(process=MagicMock(), port=9090) + + resource = ProxyRuntimeResource(service=service) + result = resource.start(ctx, {"S3_ENDPOINT_URL": "http://localhost:4566"}) + + service.start_for_run.assert_called_once_with(upstream_url="http://localhost:4566") + assert result.env["ROAR_UPSTREAM_S3_ENDPOINT"] == "http://localhost:4566" + + +def test_start_detects_generic_aws_endpoint_url_as_upstream(ctx): + """The generic AWS_ENDPOINT_URL (the common MinIO/LocalStack setup) is + detected as the upstream to chain to, even though we inject the S3-scoped + vars. Regression guard for the case where a user only set AWS_ENDPOINT_URL.""" + service = MagicMock() + service.find_proxy.return_value = "/tmp/roar-proxy" + service.start_for_run.return_value = ProxyHandle(process=MagicMock(), port=9090) + + resource = ProxyRuntimeResource(service=service) + result = resource.start(ctx, {"AWS_ENDPOINT_URL": "http://localhost:4566"}) + + service.start_for_run.assert_called_once_with(upstream_url="http://localhost:4566") + assert result.env["ROAR_UPSTREAM_S3_ENDPOINT"] == "http://localhost:4566" + # still redirects S3 clients via the scoped vars, not the generic one + assert result.env["AWS_ENDPOINT_URL_S3"] == "http://127.0.0.1:9090" + assert "AWS_ENDPOINT_URL" not in result.env + + def test_start_without_existing_endpoint_only_sets_proxy_url(ctx): service = MagicMock() service.find_proxy.return_value = "/tmp/roar-proxy" @@ -48,7 +79,10 @@ def test_start_without_existing_endpoint_only_sets_proxy_url(ctx): result = resource.start(ctx, {}) service.start_for_run.assert_called_once_with(upstream_url=None) - assert result.env == {"AWS_ENDPOINT_URL": "http://127.0.0.1:9090"} + assert result.env == { + "AWS_ENDPOINT_URL_S3": "http://127.0.0.1:9090", + "S3_ENDPOINT_URL": "http://127.0.0.1:9090", + } def test_start_failure_returns_empty_env_and_keeps_resource_inactive(ctx):