Add sync optimization support to pulp_file

Whe the manifest hasn't changed relative to the previous sync, we skip
the sync, so long as the sync modes are compatible and such.

This requires storing a bit of hidden metadata on the repository object.

Author: dralley

CHANGES/pulp_file/+sync-optimization.misc

@@ -0,0 +1,3 @@
+Add backend support for sync-optimization for the file plugin
+
+We can't actually use it yet due to zero-downtime upgrade restrictions

pulp_file/app/migrations/0020_filerepository_last_sync_details.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("file", "0019_add_filegitremote"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="filerepository",
+            name="last_sync_details",
+            field=models.JSONField(default=dict),
+        ),
+    ]

pulp_file/app/models.py

@@ -91,6 +91,7 @@ class FileRepository(Repository, AutoAddObjPermsMixin):
 
     manifest = models.TextField(default="PULP_MANIFEST", null=True)
     autopublish = models.BooleanField(default=False)
+    last_sync_details = models.JSONField(default=dict)
 
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"

pulp_file/app/serializers.py

@@ -16,6 +16,7 @@
     PublicationSerializer,
     RemoteSerializer,
     RepositorySerializer,
+    RepositorySyncURLSerializer,
     SingleArtifactContentUploadSerializer,
 )
 from pulpcore.plugin.util import get_domain_pk
@@ -139,11 +140,28 @@ class FileRepositorySerializer(RepositorySerializer):
         allow_null=True,
     )
 
+    last_sync_details = serializers.JSONField(
+        help_text=_("Details about the last sync of this repository."),
+        read_only=True,
+    )
+
     class Meta:
-        fields = RepositorySerializer.Meta.fields + ("autopublish", "manifest")
+        fields = RepositorySerializer.Meta.fields + ("autopublish", "manifest", "last_sync_details")
         model = FileRepository
 
 
+class FileRepositorySyncURLSerializer(RepositorySyncURLSerializer):
+    """
+    Serializer for File Repository Sync URL.
+    """
+
+    optimize = serializers.BooleanField(
+        help_text=_("Whether or not to optimize sync."),
+        required=False,
+        default=True,
+    )
+
+
 class FileRemoteSerializer(RemoteSerializer):
     """
     Serializer for File Remotes.

pulp_file/app/tasks/synchronizing.py

@@ -3,6 +3,7 @@
 import re
 import shutil
 import tempfile
+from hashlib import sha256
 
 from gettext import gettext as _
 from urllib.parse import quote, urlparse, urlunparse
@@ -36,7 +37,51 @@
 metadata_files = []
 
 
-def synchronize(remote_pk, repository_pk, mirror, url=None):
+def _get_sha256(file_path):
+    """Compute the SHA256 hex digest of a file."""
+    with open(file_path, "rb") as f:
+        return sha256(f.read()).hexdigest()
+
+
+def _should_optimize_sync(sync_details, last_sync_details):
+    """
+    Check whether the sync can be skipped by comparing with the previous sync.
+
+    Args:
+        sync_details (dict): Details about the current sync configuration.
+        last_sync_details (dict): Details about the previous sync configuration.
+
+    Returns:
+        bool: True if sync can be skipped; False otherwise.
+
+    """
+    if not last_sync_details:
+        return False
+
+    # If switching to immediate download, we may need to download content
+    if (
+        last_sync_details.get("download_policy") != "immediate"
+        and sync_details["download_policy"] == "immediate"
+    ):
+        return False
+
+    # If switching to mirror mode, we need to create a publication
+    if not last_sync_details.get("mirror") and sync_details["mirror"]:
+        return False
+
+    if last_sync_details.get("url") != sync_details["url"]:
+        return False
+
+    if last_sync_details.get("most_recent_version") != sync_details["most_recent_version"]:
+        return False
+
+    if last_sync_details.get("manifest_checksum") != sync_details["manifest_checksum"]:
+        return False
+
+    return True
+
+
+def synchronize(remote_pk, repository_pk, mirror, optimize=False, url=None, **kwargs):
     """
     Sync content from the remote repository.
 
@@ -46,6 +91,7 @@ def synchronize(remote_pk, repository_pk, mirror, url=None):
         remote_pk (str): The remote PK.
         repository_pk (str): The repository PK.
         mirror (bool): True for mirror mode, False for additive.
+        optimize (bool): Whether to skip sync if nothing has changed.
         url (str): The url to synchronize. If omitted, the url of the remote is used.
 
     Raises:
@@ -67,10 +113,42 @@ def synchronize(remote_pk, repository_pk, mirror, url=None):
             for stage in old_pipeline_stages(new_version)
             if not isinstance(stage, (RemoteArtifactSaver))
         ]
+        rv = dv.create()
     else:
-        first_stage = FileFirstStage(remote, url)
+        sync_url = url or remote.url
+        version = repository.latest_version()
+
+        # Download the manifest to compute its checksum for optimization
+        downloader = remote.get_downloader(url=sync_url)
+        manifest_result = downloader.fetch()
+
+        sync_details = {
+            "url": remote.url,
+            "download_policy": remote.policy,
+            "mirror": mirror,
+            "most_recent_version": version.number,
+            "manifest_checksum": _get_sha256(manifest_result.path),
+        }
+
+        if optimize and _should_optimize_sync(sync_details, repository.last_sync_details):
+            with ProgressReport(
+                message="Skipping Sync (no change from previous sync)",
+                code="sync.was_skipped",
+            ) as pb:
+                pb.total = 1
+                pb.done = 1
+            return
+
+        first_stage = FileFirstStage(remote, url, manifest_path=manifest_result.path)
         dv = DeclarativeVersion(first_stage, repository, mirror=mirror, acs=True)
-    rv = dv.create()
+        rv = dv.create()
+
+        # Update last_sync_details after sync
+        if rv:
+            sync_details["most_recent_version"] = rv.number
+        repository.last_sync_details = sync_details
+        repository.save()
+
     if rv and mirror:
         # TODO: this is awful, we really should rewrite the DeclarativeVersion API to
         # accomodate this use case
@@ -98,18 +176,21 @@ class FileFirstStage(Stage):
     The first stage of a pulp_file sync pipeline.
     """
 
-    def __init__(self, remote, url):
+    def __init__(self, remote, url, manifest_path=None):
         """
         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
+            manifest_path (str): Path to an already-downloaded manifest file. If provided,
+                the manifest will not be downloaded again.
 
         """
         super().__init__()
         self.remote = remote
         self.url = url if url else remote.url
+        self.manifest_path = manifest_path
 
     async def run(self):
         """
@@ -123,15 +204,19 @@ async def run(self):
         ) as pb:
             parsed_url = urlparse(self.url)
             root_dir = os.path.dirname(parsed_url.path)
-            downloader = self.remote.get_downloader(url=self.url)
-            result = await downloader.run()
+            if self.manifest_path:
+                result_path = self.manifest_path
+            else:
+                downloader = self.remote.get_downloader(url=self.url)
+                result = await downloader.run()
+                result_path = result.path
             await pb.aincrement()
-            metadata_files.append((result.path, self.url.split("/")[-1]))
+            metadata_files.append((result_path, self.url.split("/")[-1]))
 
         async with ProgressReport(
             message="Parsing Metadata Lines", code="sync.parsing.metadata"
         ) as pb:
-            manifest = Manifest(result.path)
+            manifest = Manifest(result_path)
             entries = list(manifest.read())
 
             pb.total = len(entries)

pulp_file/app/viewsets.py

@@ -16,7 +16,6 @@
 )
 from pulpcore.plugin.serializers import (
     AsyncOperationResponseSerializer,
-    RepositorySyncURLSerializer,
     TaskGroupOperationResponseSerializer,
 )
 from pulpcore.plugin.tasking import dispatch
@@ -52,6 +51,7 @@
     FileGitRemoteSerializer,
     FileRemoteSerializer,
     FileRepositorySerializer,
+    FileRepositorySyncURLSerializer,
     FilePublicationSerializer,
 )
 
@@ -252,14 +252,14 @@ class FileRepositoryViewSet(RepositoryViewSet, ModifyRepositoryActionMixin, Role
         summary="Sync from a remote",
         responses={202: AsyncOperationResponseSerializer},
     )
-    @action(detail=True, methods=["post"], serializer_class=RepositorySyncURLSerializer)
+    @action(detail=True, methods=["post"], serializer_class=FileRepositorySyncURLSerializer)
     def sync(self, request, pk):
         """
         Synchronizes a repository.
 
         The ``repository`` field has to be provided.
         """
-        serializer = RepositorySyncURLSerializer(
+        serializer = FileRepositorySyncURLSerializer(
             data=request.data, context={"request": request, "repository_pk": pk}
         )
         serializer.is_valid(raise_exception=True)
@@ -268,6 +268,7 @@ def sync(self, request, pk):
         remote = serializer.validated_data.get("remote", repository.remote)
 
         mirror = serializer.validated_data.get("mirror", False)
+        optimize = serializer.validated_data.get("optimize", True)  # noqa
         result = dispatch(
             tasks.synchronize,
             shared_resources=[remote],
@@ -276,6 +277,7 @@ def sync(self, request, pk):
                 "remote_pk": str(remote.pk),
                 "repository_pk": str(repository.pk),
                 "mirror": mirror,
+                # "optimize": optimize,  # TODO: uncomment in 3.109
             },
         )
         return OperationPostponedResponse(result, request)