ACS refresh

Added `refresh` endpoint to support ACS.

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

Required PR: pulp/pulpcore#1496

CHANGES/9097.feature

@@ -0,0 +1 @@
+Added refresh endpoint for Alernate Content Sources.

docs/workflows/alternate-content-source.rst

@@ -0,0 +1,29 @@
+Alternate Content Sources
+=========================
+
+To use an Alternate Content Source you need a ``FileRemote`` with base path of your
+ACS.
+
+.. code-block:: bash
+
+    http POST localhost:24817/pulp/api/v3/remotes/file/file/ name="remoteForACS" policy="on_demand" url="http://fixtures.pulpproject.org/"
+
+Create Alternate Content Source
+-------------------------------
+
+.. code-block:: bash
+
+    http POST localhost:24817/pulp/api/v3/acs/file/file/ name="fileAcs" remote=<pulp-remote-href> paths:='["file/PULP_MANIFEST", "backup/MANIFEST"]'
+
+Refresh ACS
+-----------
+
+To make your ACS available for future syncs you need to call ``refresh`` endpoint
+on your ACS.
+
+.. code-block:: bash
+
+    http POST localhost:24817/pulp/api/v3/acs/file/file/<ACS-UUID>/refresh/
+
+Alternate Content Source has a global scope so if any content found in ACS it
+will be used in all future syncs.

docs/workflows/index.rst

@@ -41,3 +41,4 @@ library with:
    sync
    upload
    publish-host
+   alternate-content-source

pulp_file/app/tasks/__init__.py

@@ -1,2 +1,5 @@
 from .publishing import publish  # noqa
 from .synchronizing import synchronize  # noqa
+
+# file_acs_refresh must be imported after synchronize to avoid circular import
+from .acs import file_acs_refresh  # noqa

pulp_file/app/tasks/acs.py

@@ -0,0 +1,55 @@
+import os
+
+from datetime import datetime
+
+from pulpcore.plugin.models import (
+    AlternateContentSource,
+    AlternateContentSourcePath,
+    ProgressReport,
+    TaskGroup,
+)
+from pulpcore.plugin.tasking import dispatch
+
+from pulp_file.app.tasks import synchronize
+from pulp_file.app.models import FileRepository
+
+metadata_files = []
+
+
+def file_acs_refresh(acs_pk):
+    """
+    Refresh alternative content sources.
+    """
+    acs = AlternateContentSource.objects.get(pk=acs_pk)
+    acs_paths = AlternateContentSourcePath.objects.filter(alternate_content_source=acs_pk)
+    with ProgressReport(
+        message="Refreshing ACS metadata", code="acs.refreshing.metadata", total=len(acs_paths)
+    ) as pb:
+        for acs_path in acs_paths:
+            # Create or get repository for each path
+            repo_data = {
+                "name": f"{acs.name}--{acs_path.pk}--repository",
+                "retain_repo_versions": 1,
+                "user_hidden": True,
+            }
+            repo, created = FileRepository.objects.get_or_create(**repo_data)
+            if created:
+                acs_path.repository = repo
+                acs_path.save()
+            acs_url = os.path.join(acs.remote.url, acs_path.path)
+            # Dispatch each ACS path to own task and assign it to common TaskGroup
+            task_group = TaskGroup.objects.create(description=f"Refresh of {acs_path.path}")
+            dispatch(
+                synchronize,
+                shared_resources=[acs.remote],
+                exclusive_resources=[acs],
+                task_group=task_group,
+                kwargs={
+                    "remote_pk": str(acs.remote.pk),
+                    "repository_pk": str(acs_path.repository.pk),
+                    "mirror": False,
+                    "url": acs_url,
+                },
+            )
+            pb.increment()
+    acs.last_refreshed = datetime.now()

pulp_file/app/tasks/synchronizing.py

@@ -24,7 +24,7 @@
 metadata_files = []
 
 
-def synchronize(remote_pk, repository_pk, mirror):
+def synchronize(remote_pk, repository_pk, mirror, url=None):
     """
     Sync content from the remote repository.
 
@@ -34,6 +34,7 @@ def synchronize(remote_pk, repository_pk, mirror):
         remote_pk (str): The remote PK.
         repository_pk (str): The repository PK.
         mirror (bool): True for mirror mode, False for additive.
+        url (str): The url to synchronize. If omitted, the url of the remote is used.
 
     Raises:
         ValueError: If the remote does not specify a URL to sync.
@@ -45,8 +46,8 @@ def synchronize(remote_pk, repository_pk, mirror):
     if not remote.url:
         raise ValueError(_("A remote must have a url specified to synchronize."))
 
-    first_stage = FileFirstStage(remote)
-    dv = DeclarativeVersion(first_stage, repository, mirror=mirror)
+    first_stage = FileFirstStage(remote, url)
+    dv = DeclarativeVersion(first_stage, repository, mirror=mirror, acs=True)
     rv = dv.create()
     if rv and mirror:
         # TODO: this is awful, we really should rewrite the DeclarativeVersion API to
@@ -72,16 +73,18 @@ class FileFirstStage(Stage):
     The first stage of a pulp_file sync pipeline.
     """
 
-    def __init__(self, remote):
+    def __init__(self, remote, url):
         """
         The first stage of a pulp_file sync pipeline.
 
         Args:
             remote (FileRemote): The remote data to be used when syncing
+            url (str): The base url of custom remote
 
         """
         super().__init__()
         self.remote = remote
+        self.url = url if url else remote.url
 
     async def run(self):
         """
@@ -93,12 +96,12 @@ async def run(self):
         async with ProgressReport(
             message="Downloading Metadata", code="sync.downloading.metadata"
         ) as pb:
-            parsed_url = urlparse(self.remote.url)
+            parsed_url = urlparse(self.url)
             root_dir = os.path.dirname(parsed_url.path)
-            downloader = self.remote.get_downloader(url=self.remote.url)
+            downloader = self.remote.get_downloader(url=self.url)
             result = await downloader.run()
             await pb.aincrement()
-            metadata_files.append((result.path, self.remote.url.split("/")[-1]))
+            metadata_files.append((result.path, self.url.split("/")[-1]))
 
         async with ProgressReport(
             message="Parsing Metadata Lines", code="sync.parsing.metadata"

pulp_file/app/viewsets.py

@@ -183,6 +183,10 @@ class FileDistributionViewSet(DistributionViewSet):
     serializer_class = FileDistributionSerializer
 
 
+@extend_schema(
+    description="Trigger an asynchronous task to create Alternate Content Source content.",
+    responses={202: AsyncOperationResponseSerializer},
+)
 class FileAlternateContentSourceViewSet(AlternateContentSourceViewSet):
     """
     Alternate Content Source ViewSet for File
@@ -193,3 +197,18 @@ class FileAlternateContentSourceViewSet(AlternateContentSourceViewSet):
     endpoint_name = "file"
     queryset = FileAlternateContentSource.objects.all()
     serializer_class = FileAlternateContentSourceSerializer
+
+    @action(methods=["post"], detail=True)
+    def refresh(self, request, pk):
+        """
+        Refresh ACS metadata.
+        """
+        acs = self.get_object()
+        result = dispatch(
+            tasks.file_acs_refresh,
+            [acs],
+            kwargs={
+                "acs_pk": pk,
+            },
+        )
+        return OperationPostponedResponse(result, request)