Resolve storage paths through a backend registry

.ci/ansible/inventory.yaml

@@ -6,7 +6,6 @@ all:
         pulp:
         pulp-fixtures:
         rustfs:
-        ci-sftp:
       vars:
         ansible_connection: docker
 ...

CHANGES/+stable-domain-storage-backends.feature

@@ -0,0 +1 @@
+Storage backends can now be replaced without changing Domain records or the Domain API.

docs/dev/learn/domains/storage_backend_implementations.md

@@ -0,0 +1,68 @@
+# Replacing a Domain storage implementation
+
+Domain records retain historical storage class paths so older Pulp processes can read them during
+a rolling upgrade. Pulp resolves those paths to stable backend IDs (`filesystem`, `s3`, `azure`, and
+`gcs`), then selects the installed class through `DOMAIN_STORAGE_BACKENDS`. The API still advertises
+the historical class paths and returns them on reads. It also accepts stable IDs in raw requests,
+but generated clients may reject those IDs because the schema still lists the historical paths.
+Existing `S3Boto3Storage` paths are preserved rather than rewritten to the newer S3 path.
+
+This boundary replaces the code for an existing storage type; it does not register a new type.
+Only filesystem, S3, and Azure are currently offered for new Domains. GCS is recognized for
+existing configurations, and SFTP is not supported.
+
+## Select an implementation
+
+Set the class path for each non-default Domain storage type you are replacing:
+
+```python
+DOMAIN_STORAGE_BACKENDS = {
+    "filesystem": "pulpcore.app.models.storage.FileSystem",
+    "s3": "my_storage.backends.s3.S3Storage",
+    "azure": "storages.backends.azure_storage.AzureStorage",
+    "gcs": "storages.backends.gcloud.GoogleCloudStorage",
+}
+```
+
+The default Domain instead uses Django's `STORAGES["default"]`. When replacing that class, set
+both its `BACKEND` and the corresponding `DOMAIN_STORAGE_BACKENDS` entry. The mapping lets Pulp
+identify its storage type without requiring the replacement's Python path in a Domain record.
+An unrecognized default class can serve redirects only if it implements `pulp_url`.
+
+For direct content redirects, a replacement may implement
+`pulp_url(name, headers=None, method=None)` to sign URLs with Pulp's response-header overrides.
+Without that hook, Pulp calls the django-storages-style `url()` method for recognized object storage
+types. For ranged content reads, a replacement may expose a synchronous
+`open_stream(name, start=0, length=None)` context manager. Pulp falls back to Django's `open()` and
+`seek()` if the hook is absent; that fallback bounds the returned bytes but may buffer a whole
+remote object. The stream must release its resources when the context exits.
+
+## Use storage from a plugin
+
+Plugin code should use `pulpcore.plugin.util.get_local_storage_path(file, domain=None)` when it
+needs an actual filesystem path. It returns `None` when the backend does not provide one; other
+storage errors are propagated. Use `get_direct_storage_url(file, headers=None, http_method=None,
+domain=None)` when a consumer needs a URL directly to object storage. It returns `None` when the
+Domain uses filesystem storage or has redirects disabled, and raises `NotImplementedError` when
+redirects are requested from an unsupported backend. Both functions accept a Pulp-managed file
+such as `Artifact.file` or `PulpTemporaryFile.file`. If the file belongs to another Domain, pass
+that Domain explicitly; otherwise they use the current Domain.
+
+The URL function accepts generic `Content-Type` and `Content-Disposition` response headers and
+translates them into the selected provider's signed URL parameters. Plugins should not inspect
+`storage_class` or pass provider-specific parameters to `storage.url()`. The existing
+`get_artifact_url(artifact, ...)` differs: it always returns a URL, falling back to the content
+app when a direct URL is unavailable.
+
+## Roll out a replacement
+
+Keep the Domain's storage type, bucket or container, prefix, object names, credentials, and URL
+behavior compatible. Install the replacement and its dependencies on every API, content, and worker
+process, and test reads, writes, range requests, and signed redirects against existing objects.
+Then change the deployment settings. A code-only change needs no Domain data migration. Moving
+objects to a different location still requires a storage migration.
+
+During a rolling upgrade, new processes continue to store and return historical class paths so
+older processes can read the same Domain rows. Keep the old implementation installed until those
+processes and the rollback window are gone. Do not switch persisted `storage_class` values to IDs
+while any process still requires the historical paths.

docs/user/guides/create-domains.md

@@ -85,8 +85,7 @@ Specific examples for different storage backends:
 
 
 !!! warning
-    Changing the `storage-class` or `storage-settings` of an in-use domain is dangerous and can result in a broken domain.
-    TODO Migration feature
+    Changing a domain's storage location without migrating its artifacts can break the domain.
 
 ## Using Domains
 

pulpcore/app/apps.py

@@ -11,6 +11,7 @@
 from django.db.models.signals import post_migrate, pre_migrate
 from django.utils.module_loading import module_has_submodule
 
+from pulpcore.constants import STORAGE_BACKENDS
 from pulpcore.exceptions.plugin import MissingPlugin
 
 VIEWSETS_MODULE_NAME = "viewsets"
@@ -333,18 +334,30 @@ def _ensure_default_domain(sender, apps, **kwargs):
     with suppress(LookupError, ProgrammingError):
         table_names = connection.introspection.table_names()
         if "core_domain" in table_names:
+            from pulpcore.app.storage_backends import (
+                backend_id,
+                default_backend_id,
+                legacy_backend_path,
+            )
             from pulpcore.app.util import get_default_domain
 
             default = get_default_domain()  # Cache the default domain
+            # A class-path change within one backend must not rewrite the legacy row.
+            storage_changed = default_backend_id() != (
+                backend_id(default.storage_class) or default.storage_class
+            )
             # Match the Pulp settings
             if (
                 settings.HIDE_GUARDED_DISTRIBUTIONS != default.hide_guarded_distributions
                 or settings.REDIRECT_TO_OBJECT_STORAGE != default.redirect_to_object_storage
-                or settings.STORAGES["default"]["BACKEND"] != default.storage_class
+                or storage_changed
             ):
                 default.hide_guarded_distributions = settings.HIDE_GUARDED_DISTRIBUTIONS
                 default.redirect_to_object_storage = settings.REDIRECT_TO_OBJECT_STORAGE
-                default.storage_class = settings.STORAGES["default"]["BACKEND"]
+                if storage_changed:
+                    default.storage_class = legacy_backend_path(
+                        settings.STORAGES["default"]["BACKEND"]
+                    )
                 default.save(skip_hooks=True)
 
 
@@ -408,8 +421,10 @@ def _get_permission(perm):
 
 
 def _populate_artifact_serving_distribution(sender, apps, verbosity, **kwargs):
+    from pulpcore.app.storage_backends import default_backend_id
+
     if (
-        settings.STORAGES["default"]["BACKEND"] == "pulpcore.app.models.storage.FileSystem"
+        default_backend_id() == STORAGE_BACKENDS.FILESYSTEM
         or not settings.REDIRECT_TO_OBJECT_STORAGE
     ):
         try:

pulpcore/app/checks.py

@@ -15,10 +15,45 @@
     Warning as CheckWarning,
 )
 from django.db.models import Q
+from django.utils.module_loading import import_string
 
 from pulpcore import constants
 
 
+@register()
+def storage_redirect_backend(app_configs, **kwargs):
+    """Reject default storage that the content app cannot serve with redirects."""
+    from pulpcore.app.storage_backends import backend_id
+
+    if not settings.REDIRECT_TO_OBJECT_STORAGE:
+        return []
+    path = settings.STORAGES["default"]["BACKEND"]
+    backend = backend_id(path)
+    if backend in {
+        constants.STORAGE_BACKENDS.FILESYSTEM,
+        constants.STORAGE_BACKENDS.S3,
+        constants.STORAGE_BACKENDS.AZURE,
+        constants.STORAGE_BACKENDS.GCS,
+    }:
+        return []
+
+    # An unregistered default backend can still redirect through Pulp's URL hook.
+    try:
+        implementation = import_string(path)
+    except (ImportError, AttributeError):
+        implementation = None
+    if callable(getattr(implementation, "pulp_url", None)):
+        return []
+
+    return [
+        CheckError(
+            "STORAGES['default']['BACKEND'] must map to a supported storage type "
+            "or implement pulp_url when REDIRECT_TO_OBJECT_STORAGE is enabled.",
+            id="pulpcore.E003",
+        )
+    ]
+
+
 @register(deploy=True)
 def secret_key_check(app_configs, **kwargs):
     messages = []
@@ -38,7 +73,9 @@ def secret_key_check(app_configs, **kwargs):
 def storage_paths(app_configs, **kwargs):
     warnings = []
 
-    if settings.STORAGES["default"]["BACKEND"] == "pulpcore.app.models.storage.FileSystem":
+    from pulpcore.app.storage_backends import default_backend_id
+
+    if default_backend_id() == constants.STORAGE_BACKENDS.FILESYSTEM:
         try:
             media_root_dev = Path(settings.MEDIA_ROOT).stat().st_dev
         except OSError:

pulpcore/app/importexport.py

@@ -17,7 +17,7 @@
 from pulpcore.app.models.content import Artifact
 from pulpcore.app.models.progress import ProgressReport
 from pulpcore.app.models.repository import Repository
-from pulpcore.constants import EXPORT_BATCH_SIZE, TASK_STATES
+from pulpcore.constants import EXPORT_BATCH_SIZE, STORAGE_BACKENDS, TASK_STATES
 
 log = logging.getLogger(__name__)
 
@@ -118,7 +118,9 @@ def export_artifacts(export, artifact_pks):
     with ProgressReport(**data) as pb:
         pb.BATCH_INTERVAL = 5000
 
-        if settings.STORAGES["default"]["BACKEND"] != "pulpcore.app.models.storage.FileSystem":
+        from pulpcore.app.storage_backends import default_backend_id
+
+        if default_backend_id() != STORAGE_BACKENDS.FILESYSTEM:
             with tempfile.TemporaryDirectory(dir=".") as temp_dir:
                 for offset in range(0, len(artifact_pks), EXPORT_BATCH_SIZE):
                     batch = artifact_pks[offset : offset + EXPORT_BATCH_SIZE]

pulpcore/app/models/domain.py

@@ -28,7 +28,7 @@ class Domain(BaseModel, AutoAddObjPermsMixin):
         name (models.SlugField): Unique name of domain
         pulp_labels (HStoreField): Dictionary of string values
         description (models.TextField): Optional description of domain
-        storage_class (models.TextField): Required storage class for backend
+        storage_class (models.TextField): Historical class-path identifier for the backend
         storage_settings (EncryptedJSONField): Settings needed to configure storage backend
         redirect_to_object_storage (models.BooleanField): Redirect to object storage in content app
         hide_guarded_distributions (models.BooleanField): Hide guarded distributions in content app
@@ -41,7 +41,7 @@ class Domain(BaseModel, AutoAddObjPermsMixin):
     name = models.SlugField(null=False, unique=True)
     pulp_labels = HStoreField(default=dict)
     description = models.TextField(null=True)
-    # Storage class is required, optional settings are validated by serializer
+    # Class paths remain persisted so older Pulp processes can read Domain rows.
     storage_class = models.TextField(null=False)
     storage_settings = EncryptedJSONField(default=dict)
     # Pulp settings that are appropriate to be set on a "per domain" level
@@ -53,21 +53,32 @@ class Domain(BaseModel, AutoAddObjPermsMixin):
     )
 
     def get_storage(self):
-        """Returns this domain's instantiated storage class."""
+        """Return the deployment-selected storage implementation for this domain."""
         if self.name == "default":
             return default_storage
 
+        from pulpcore.app.storage_backends import backend_path
+
+        implementation = backend_path(self.storage_class)
         if date_storage_tuple := storages.get(self.pulp_id):
-            last_updated, storage = date_storage_tuple
-            if self.pulp_last_updated == last_updated:
+            last_updated, path, storage = date_storage_tuple
+            # A settings-only implementation swap must invalidate the process cache.
+            if self.pulp_last_updated == last_updated and path == implementation:
                 return storage
 
         from pulpcore.app.serializers import DomainSerializer
 
         storage = DomainSerializer(instance=self).create_storage()
-        storages[self.pulp_id] = (self.pulp_last_updated, storage)
+        storages[self.pulp_id] = (self.pulp_last_updated, implementation, storage)
         return storage
 
+    @property
+    def storage_backend(self):
+        """Return the stable backend ID for a historical class path or an ID."""
+        from pulpcore.app.storage_backends import backend_id
+
+        return backend_id(self.storage_class)
+
     @hook(BEFORE_DELETE, when="name", is_now="default")
     @hook(BEFORE_UPDATE, when="name", was="default")
     def prevent_default_deletion(self):