Stream proxied S3 and Azure artifacts

CHANGES/7806.bugfix

@@ -0,0 +1 @@
+Fixed proxying large S3 and Azure artifacts without buffering their complete contents.

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(

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(

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")

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 = [