diff --git a/CHANGES/7806.bugfix b/CHANGES/7806.bugfix new file mode 100644 index 00000000000..357015a17cc --- /dev/null +++ b/CHANGES/7806.bugfix @@ -0,0 +1 @@ +Fixed proxying large S3 and Azure artifacts without buffering their complete contents. diff --git a/pulp_file/tests/functional/api/test_download_policies.py b/pulp_file/tests/functional/api/test_download_policies.py index 46020243f1b..882457a008c 100644 --- a/pulp_file/tests/functional/api/test_download_policies.py +++ b/pulp_file/tests/functional/api/test_download_policies.py @@ -7,10 +7,15 @@ from urllib.parse import urljoin import pytest +import requests from aiohttp.client_exceptions import ClientResponseError from bs4 import BeautifulSoup -from pulpcore.client.pulp_file import FileFilePublication, FileRepositorySyncURL +from pulpcore.client.pulp_file import ( + FileFilePublication, + FileRepositorySyncURL, + RepositoryAddRemoveContent, +) from pulpcore.tests.functional.utils import download_file, get_files_in_manifest OBJECT_STORAGES = ( @@ -37,6 +42,74 @@ def _do_range_request_download_and_assert(url, range_header, expected_bytes): ) +@pytest.mark.parametrize( + "storage_class", + ( + pytest.param("storages.backends.s3.S3Storage", id="s3"), + pytest.param("storages.backends.s3boto3.S3Boto3Storage", id="s3boto3"), + pytest.param("storages.backends.azure_storage.AzureStorage", id="azure"), + ), +) +def test_proxied_object_storage_artifact_streaming( + storage_class, + pulp_settings, + domain_factory, + random_artifact_factory, + file_bindings, + file_repository_factory, + file_publication_factory, + file_distribution_factory, + distribution_base_url, + gen_object_with_cleanup, + monitor_task, +): + """Serve full and ranged object-storage content through Pulp without redirects.""" + + if pulp_settings.STORAGES["default"]["BACKEND"] != storage_class: + pytest.skip("The functional environment does not provide this object-storage configuration") + + domain = domain_factory(storage_class=storage_class, redirect_to_object_storage=False) + artifact = random_artifact_factory(pulp_domain=domain.name, size=32) + content = gen_object_with_cleanup( + file_bindings.ContentFilesApi, + artifact=artifact.pulp_href, + relative_path=str(uuid.uuid4()), + pulp_domain=domain.name, + ) + repository = file_repository_factory(pulp_domain=domain.name) + monitor_task( + file_bindings.RepositoriesFileApi.modify( + repository.pulp_href, + RepositoryAddRemoveContent(add_content_units=[content.pulp_href]), + ).task + ) + publication = file_publication_factory( + pulp_domain=domain.name, + repository=repository.pulp_href, + ) + distribution = file_distribution_factory( + pulp_domain=domain.name, + publication=publication.pulp_href, + ) + content_url = urljoin(distribution_base_url(distribution.base_url), content.relative_path) + + full_response = requests.get(content_url, allow_redirects=False) + assert full_response.status_code == requests.codes.ok + assert not full_response.is_redirect + assert hashlib.sha256(full_response.content).hexdigest() == artifact.sha256 + assert full_response.headers["Content-Length"] == str(len(full_response.content)) + assert full_response.headers["Accept-Ranges"] == "bytes" + + range_response = requests.get( + content_url, headers={"Range": "bytes=1-4"}, allow_redirects=False + ) + assert range_response.status_code == requests.codes.partial_content + assert not range_response.is_redirect + assert range_response.content == full_response.content[1:5] + assert range_response.headers["Content-Length"] == "4" + assert range_response.headers["Content-Range"] == f"bytes 1-4/{len(full_response.content)}" + + @pytest.mark.parallel @pytest.mark.parametrize("download_policy", ["immediate", "on_demand", "streamed"]) def test_download_policy( diff --git a/pulpcore/responses.py b/pulpcore/responses.py index 1b1fac62a0d..8f45a50dd85 100644 --- a/pulpcore/responses.py +++ b/pulpcore/responses.py @@ -1,4 +1,5 @@ import asyncio +from contextlib import asynccontextmanager from aiohttp import hdrs from aiohttp.web import StreamResponse @@ -9,6 +10,15 @@ from pulpcore.app.models import Artifact +STREAMING_STORAGE_CLASSES = frozenset( + ( + "storages.backends.s3.S3Storage", + "storages.backends.s3boto3.S3Boto3Storage", + "storages.backends.azure_storage.AzureStorage", + "storages.backends.gcloud.GoogleCloudStorage", + ) +) + class ArtifactResponse(StreamResponse): """A response object can be used to send artifacts.""" @@ -33,6 +43,12 @@ def __init__( self._chunk_size = chunk_size async def _sendfile(self, request, fobj, offset, count): + if self._artifact.pulp_domain.storage_class in STREAMING_STORAGE_CLASSES: + storage = self._artifact.pulp_domain.get_storage() + return await self._sendfile_storage_stream( + request, storage.open_stream, fobj.name, offset, count + ) + # To keep memory usage low, fobj is transferred in chunks # controlled by the constructor's chunk_size argument. @@ -54,6 +70,56 @@ async def _sendfile(self, request, fobj, offset, count): await writer.drain() return writer + @staticmethod + @asynccontextmanager + async def _storage_stream(stream_opener, name, offset, count): + """Bridge django-storages' synchronous streaming context manager. + + The django-storages fork supplies ``open_stream()`` on the remote + backends selected above. Its provider I/O, including opening and + closing the context, must stay off the content app's event loop. + Propagate exception details to the synchronous context manager so it + retains normal ``with`` semantics. Once django-storages releases this + API upstream, replace the forked dependency with that release. + """ + + stream_context = await asyncio.to_thread(stream_opener, name, start=offset, length=count) + stream = await asyncio.to_thread(stream_context.__enter__) + try: + yield stream + except BaseException as exc: + if not await asyncio.to_thread( + stream_context.__exit__, type(exc), exc, exc.__traceback__ + ): + raise + else: + await asyncio.to_thread(stream_context.__exit__, None, None, None) + + async def _sendfile_storage_stream(self, request, stream_opener, name, offset, count): + """Write an ``open_stream()`` response in bounded chunks. + + The storage API is synchronous while aiohttp writes are asynchronous. + Bound each provider read to the remaining HTTP range and use aiohttp's + normal backpressure for every write. + """ + + writer = await super().prepare(request) + assert writer is not None + + async with self._storage_stream(stream_opener, name, offset, count) as stream: + remaining = count + while remaining: + chunk = await asyncio.to_thread(stream.read, min(self._chunk_size, remaining)) + if not chunk: + break + if len(chunk) > remaining: + chunk = chunk[:remaining] + await writer.write(chunk) + remaining -= len(chunk) + + await writer.drain() + return writer + async def prepare(self, request): if self._artifact is None: self._artifact = await Artifact.objects.select_related("pulp_domain").aget( diff --git a/pulpcore/tests/unit/test_storage_streaming.py b/pulpcore/tests/unit/test_storage_streaming.py new file mode 100644 index 00000000000..66e16cca099 --- /dev/null +++ b/pulpcore/tests/unit/test_storage_streaming.py @@ -0,0 +1,159 @@ +"""Tests for ArtifactResponse's opt-in django-storages streaming path.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, Mock, call + +import pytest + + +@pytest.mark.parametrize( + "storage_class", + ( + "storages.backends.s3.S3Storage", + "storages.backends.s3boto3.S3Boto3Storage", + "storages.backends.azure_storage.AzureStorage", + "storages.backends.gcloud.GoogleCloudStorage", + ), +) +def test_artifact_response_uses_open_stream_for_a_domain_storage(storage_class, monkeypatch): + """Use the explicit storage API instead of opening a seekable temporary file.""" + + asyncio.run( + _test_artifact_response_uses_open_stream_for_a_domain_storage(storage_class, monkeypatch) + ) + + +async def _test_artifact_response_uses_open_stream_for_a_domain_storage(storage_class, monkeypatch): + from aiohttp.web import StreamResponse + + from pulpcore.responses import ArtifactResponse + + writer = Mock() + writer.write = AsyncMock() + writer.drain = AsyncMock() + stream = Mock() + stream.read.side_effect = (b"abc", b"def", b"") + stream_context = MagicMock() + stream_context.__enter__.return_value = stream + stream_opener = Mock(return_value=stream_context) + storage = Mock(open_stream=stream_opener) + domain = Mock(storage_class=storage_class) + domain.get_storage.return_value = storage + artifact = Mock(pulp_domain=domain) + file_object = Mock() + file_object.name = "artifact/name" + + async def prepare(_self, _request): + return writer + + monkeypatch.setattr(StreamResponse, "prepare", prepare) + response = ArtifactResponse(artifact=artifact, chunk_size=3) + + assert await response._sendfile("request", file_object, 11, 6) is writer + stream_opener.assert_called_once_with("artifact/name", start=11, length=6) + stream.read.assert_has_calls([call(3), call(3)]) + writer.write.assert_has_awaits([call(b"abc"), call(b"def")]) + stream_context.__exit__.assert_called_once_with(None, None, None) + file_object.seek.assert_not_called() + file_object.read.assert_not_called() + + +def test_artifact_response_limits_storage_stream_to_http_range(monkeypatch): + """Never send bytes beyond the headers' selected HTTP range.""" + + asyncio.run(_test_artifact_response_limits_storage_stream_to_http_range(monkeypatch)) + + +async def _test_artifact_response_limits_storage_stream_to_http_range(monkeypatch): + from aiohttp.web import StreamResponse + + from pulpcore.responses import ArtifactResponse + + writer = Mock() + writer.write = AsyncMock() + writer.drain = AsyncMock() + stream = Mock() + stream.read.return_value = b"provider returned too much" + stream_context = MagicMock() + stream_context.__enter__.return_value = stream + + async def prepare(_self, _request): + return writer + + monkeypatch.setattr(StreamResponse, "prepare", prepare) + response = ArtifactResponse(artifact=Mock(), chunk_size=10) + + assert ( + await response._sendfile_storage_stream( + "request", Mock(return_value=stream_context), "artifact/name", 0, 3 + ) + is writer + ) + writer.write.assert_awaited_once_with(b"pro") + stream_context.__exit__.assert_called_once_with(None, None, None) + + +def test_artifact_response_closes_stream_context_after_write_error(monkeypatch): + """Give the storage context the exception needed to release provider resources.""" + + asyncio.run(_test_artifact_response_closes_stream_context_after_write_error(monkeypatch)) + + +async def _test_artifact_response_closes_stream_context_after_write_error(monkeypatch): + from aiohttp.web import StreamResponse + + from pulpcore.responses import ArtifactResponse + + writer = Mock() + writer.write = AsyncMock(side_effect=RuntimeError("client disconnected")) + stream = Mock() + stream.read.return_value = b"abc" + stream_context = MagicMock() + stream_context.__enter__.return_value = stream + + async def prepare(_self, _request): + return writer + + monkeypatch.setattr(StreamResponse, "prepare", prepare) + response = ArtifactResponse(artifact=Mock(), chunk_size=3) + + with pytest.raises(RuntimeError, match="client disconnected"): + await response._sendfile_storage_stream( + "request", Mock(return_value=stream_context), "artifact/name", 0, 3 + ) + + assert stream_context.__exit__.call_args.args[0] is RuntimeError + assert str(stream_context.__exit__.call_args.args[1]) == "client disconnected" + + +def test_artifact_response_keeps_file_fallback_without_open_stream(monkeypatch): + """Retain existing storage-file serving for filesystems and unsupported backends.""" + + asyncio.run(_test_artifact_response_keeps_file_fallback_without_open_stream(monkeypatch)) + + +async def _test_artifact_response_keeps_file_fallback_without_open_stream(monkeypatch): + from aiohttp.web import StreamResponse + + from pulpcore.responses import ArtifactResponse + + writer = Mock() + writer.write = AsyncMock() + writer.drain = AsyncMock() + storage = Mock(spec=[]) + domain = Mock(storage_class="pulpcore.app.models.storage.FileSystem") + domain.get_storage.return_value = storage + artifact = Mock(pulp_domain=domain) + file_object = Mock() + file_object.read.return_value = b"payload" + + async def prepare(_self, _request): + return writer + + monkeypatch.setattr(StreamResponse, "prepare", prepare) + response = ArtifactResponse(artifact=artifact, chunk_size=8) + + assert await response._sendfile("request", file_object, 3, 7) is writer + file_object.seek.assert_called_once_with(3) + file_object.read.assert_called_once_with(7) + writer.write.assert_awaited_once_with(b"payload") diff --git a/pyproject.toml b/pyproject.toml index e242c59ab3b..3fc29f6d7d7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,10 +67,10 @@ dependencies = [ ] [project.optional-dependencies] -sftp = ["django-storages[sftp]==1.14.6"] -s3 = ["django-storages[boto3]==1.14.6"] -google = ["django-storages[google]==1.14.6"] -azure = ["django-storages[azure]==1.14.6"] +sftp = ["django-storages[sftp] @ git+https://github.com/dralley/django-storages.git@open-stream"] +s3 = ["django-storages[boto3] @ git+https://github.com/dralley/django-storages.git@open-stream"] +google = ["django-storages[google] @ git+https://github.com/dralley/django-storages.git@open-stream"] +azure = ["django-storages[azure] @ git+https://github.com/dralley/django-storages.git@open-stream"] prometheus = ["django-prometheus"] saml2 = ["djangosaml2>=1.12.0,<1.13"] kafka = [