ACS support

Adding model/serializer and refresh endpoint.

closes: #9358
https://pulp.plan.io/issues/9358

CHANGES/9358.feature

@@ -0,0 +1 @@
+Added support for Alternate Content Sources.

docs/workflows/alternate-content-source.rst

@@ -0,0 +1,54 @@
+Alternate Content Sources
+=========================
+
+Alternate Content Sources (ACS) can help speed up populating of new repositories.
+If you have content stored locally or geographically near you which matches
+the remote content, Alternate Content Sources will allow you to substitute
+this content, allowing for faster data transfer.
+
+`Alternate Content Sources <https://docs.pulpproject.org/pulpcore/workflows/alternate-content-sources.html>`_
+base is provided by pulpcore plugin.
+
+To use an Alternate Content Source you need a ``RPMRemote`` with path of your ACS.
+
+.. warning::
+
+    Remotes with mirrorlist URLs cannot be used as an Alternative Content Source.
+
+.. code-block:: bash
+
+    http POST $BASE_ADDR/pulp/api/v3/remotes/rpm/rpm/ name="myRemoteAcs" policy="on_demand" url="http://fixtures.pulpproject.org/rpm-unsigned/"
+
+Create Alternate Content Source
+-------------------------------
+
+Create an Alternate Content Source.
+
+.. code-block:: bash
+
+    http POST $BASE_ADDR/pulp/api/v3/acs/rpm/rpm/ name="myAcs" remote=$REMOTE_HREF
+
+Alternate Content Source Paths
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+If you have more places with ACS within one base path you can specify them
+by paths and all of them will be considered as a ACS.
+
+.. code-block:: bash
+
+    http POST $BASE_ADDR/pulp/api/v3/remotes/rpm/rpm/ name="myRemoteACS" policy="on_demand" url="http://fixtures.pulpproject.org/"
+    http POST $BASE_ADDR/pulp/api/v3/acs/file/file/ name="myAcs" remote=$REMOTE_HREF paths:='["rpm-unsigned/", "rpm-distribution-tree/"]'
+
+Refresh Alternate Content Source
+--------------------------------
+
+To make your ACS available for future syncs you need to call ``refresh`` endpoint
+on your ACS. This create a catalogue of available content which will be used instead
+new content if found.
+
+.. code-block:: bash
+
+    http POST $BASE_ADDR/pulp/api/v3/acs/rpm/rpm/<ACS-UUID>/refresh/
+
+Alternate Content Source has a global scope so if any content is found in ACS it
+will be used in all future syncs.

docs/workflows/index.rst

@@ -31,3 +31,4 @@ library with:
    manage
    copy
    metadata_signing
+   alternate-content-source

pulp_rpm/app/migrations/0038_rpmalternatecontentsource.py

@@ -0,0 +1,36 @@
+# Generated by Django 3.2.7 on 2021-09-17 10:47
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0076_remove_reserved_resource"),
+        ("rpm", "0037_update_json_field"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="RpmAlternateContentSource",
+            fields=[
+                (
+                    "alternatecontentsource_ptr",
+                    models.OneToOneField(
+                        auto_created=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        parent_link=True,
+                        primary_key=True,
+                        related_name="rpm_rpmalternatecontentsource",
+                        serialize=False,
+                        to="core.alternatecontentsource",
+                    ),
+                ),
+            ],
+            options={
+                "default_related_name": "%(app_label)s_%(model_name)s",
+            },
+            bases=("core.alternatecontentsource",),
+        ),
+    ]

pulp_rpm/app/models/__init__.py

@@ -10,3 +10,6 @@
 from .modulemd import Modulemd, ModulemdDefaults  # noqa
 from .package import Package  # noqa
 from .repository import RpmDistribution, RpmPublication, RpmRemote, UlnRemote, RpmRepository  # noqa
+
+# at the end to avoid circular import as ACS needs import RpmRemote
+from .acs import RpmAlternateContentSource  # noqa

pulp_rpm/app/models/acs.py

@@ -0,0 +1,19 @@
+from logging import getLogger
+
+from pulpcore.plugin.models import AlternateContentSource
+from pulp_rpm.app.models import RpmRemote
+
+
+log = getLogger(__name__)
+
+
+class RpmAlternateContentSource(AlternateContentSource):
+    """
+    Alternate Content Source for 'RPM" content.
+    """
+
+    TYPE = "rpm"
+    REMOTE_TYPES = [RpmRemote]
+
+    class Meta:
+        default_related_name = "%(app_label)s_%(model_name)s"

pulp_rpm/app/serializers/__init__.py

@@ -1,3 +1,4 @@
+from .acs import RpmAlternateContentSourceSerializer  # noqa
 from .advisory import (  # noqa
     MinimalUpdateRecordSerializer,
     UpdateCollectionSerializer,

pulp_rpm/app/serializers/acs.py

@@ -0,0 +1,24 @@
+from gettext import gettext as _
+from rest_framework import serializers
+
+from pulpcore.plugin.serializers import AlternateContentSourceSerializer
+from pulp_rpm.app.models import RpmAlternateContentSource
+
+
+class RpmAlternateContentSourceSerializer(AlternateContentSourceSerializer):
+    """
+    Serializer for RPM alternate content source.
+    """
+
+    def validate_paths(self, paths):
+        """Validate that paths do not start with /."""
+        for path in paths:
+            if path.startswith("/"):
+                raise serializers.ValidationError(_("Path cannot start with a slash."))
+            if not path.endswith("/"):
+                raise serializers.ValidationError(_("Path must end with a slash."))
+        return paths
+
+    class Meta:
+        fields = AlternateContentSourceSerializer.Meta.fields
+        model = RpmAlternateContentSource

pulp_rpm/app/tasks/synchronizing.py

@@ -31,6 +31,7 @@
     PublishedMetadata,
 )
 from pulpcore.plugin.stages import (
+    ACSArtifactHandler,
     ArtifactDownloader,
     ArtifactSaver,
     ContentSaver,
@@ -271,34 +272,40 @@ def fetch_mirror(remote):
     return None
 
 
-def fetch_remote_url(remote):
+def fetch_remote_url(remote, custom_url=None):
     """Fetch a single remote from which can be content synced."""
 
-    def normalize_url(url):
-        return url.rstrip("/") + "/"
+    def normalize_url(url_to_normalize):
+        return url_to_normalize.rstrip("/") + "/"
+
+    url = custom_url or remote.url
 
     try:
-        normalized_remote_url = normalize_url(remote.url)
+        normalized_remote_url = normalize_url(url)
         get_repomd_file(remote, normalized_remote_url)
         # just check if the metadata exists
         return normalized_remote_url
     except ClientResponseError as exc:
-        log.info(
-            _("Attempting to resolve a true url from potential mirrolist url '{}'").format(
-                remote.url
+        # If 'custom_url' is passed it is a call from ACS refresh
+        # which doesn't support mirror lists.
+        if custom_url:
+            raise ValueError(
+                _("Remote URL {} for Alternate Content Source is invalid").format(custom_url)
             )
+        log.info(
+            _("Attempting to resolve a true url from potential mirrolist url '{}'").format(url)
         )
         remote_url = fetch_mirror(remote)
         if remote_url:
             log.info(
                 _("Using url '{}' from mirrorlist in place of the provided url {}").format(
-                    remote_url, remote.url
+                    remote_url, url
                 )
             )
             return normalize_url(remote_url)
 
         if exc.status == 404:
-            raise ValueError(_("An invalid remote URL was provided: {}").format(remote.url))
+            raise ValueError(_("An invalid remote URL was provided: {}").format(url))
 
         raise exc
 
@@ -344,7 +351,7 @@ def should_optimize_sync(sync_details, last_sync_details):
     return True
 
 
-def synchronize(remote_pk, repository_pk, sync_policy, skip_types, optimize):
+def synchronize(remote_pk, repository_pk, sync_policy, skip_types, optimize, url=None):
     """
     Sync content from the remote repository.
 
@@ -366,6 +373,7 @@ def synchronize(remote_pk, repository_pk, sync_policy, skip_types, optimize):
         sync_policy (str): How to perform the sync.
         skip_types (list): List of content to skip.
         optimize(bool): Optimize mode.
+        url(str): Custom URL to use instead of Remote's URL
 
     Raises:
         ValueError: If the remote does not specify a url to sync.
@@ -377,7 +385,7 @@ def synchronize(remote_pk, repository_pk, sync_policy, skip_types, optimize):
         remote = UlnRemote.objects.get(pk=remote_pk)
     repository = RpmRepository.objects.get(pk=repository_pk)
 
-    if not remote.url:
+    if not remote.url and not url:
         raise ValueError(_("A remote must have a url specified to synchronize."))
 
     log.info(_("Synchronizing: repository={r} remote={p}").format(r=repository.name, p=remote.name))
@@ -442,7 +450,7 @@ def get_sync_details(remote, url, sync_policy, version):
         }
 
     with tempfile.TemporaryDirectory("."):
-        remote_url = fetch_remote_url(remote)
+        remote_url = fetch_remote_url(remote, url)
         sync_details = get_sync_details(
             remote, remote_url, sync_policy, repository.latest_version()
         )
@@ -547,6 +555,15 @@ class RpmDeclarativeVersion(DeclarativeVersion):
     Subclassed Declarative version creates a custom pipeline for RPM sync.
     """
 
+    def __init__(self, *args, **kwargs):
+        """
+        Adding support for ACS.
+
+        Adding it here, because we call RpmDeclarativeVersion multiple times in sync.
+        """
+        kwargs["acs"] = True
+        super().__init__(*args, **kwargs)
+
     def pipeline_stages(self, new_version):
         """
         Build a list of stages feeding into the ContentUnitAssociation stage.
@@ -564,13 +581,19 @@ def pipeline_stages(self, new_version):
         pipeline = [
             self.first_stage,
             QueryExistingArtifacts(),
-            ArtifactDownloader(),
-            ArtifactSaver(),
-            QueryExistingContents(),
-            RpmContentSaver(),
-            RpmInterrelateContent(),
-            RemoteArtifactSaver(fix_mismatched_remote_artifacts=True),
         ]
+        if self.acs:
+            pipeline.append(ACSArtifactHandler())
+        pipeline.extend(
+            [
+                ArtifactDownloader(),
+                ArtifactSaver(),
+                QueryExistingContents(),
+                RpmContentSaver(),
+                RpmInterrelateContent(),
+                RemoteArtifactSaver(fix_mismatched_remote_artifacts=True),
+            ]
+        )
         return pipeline