[PULP-2500] Deactivate local DistributedPublication in favour of pulpcore's

CHANGES/1430.distributed-publication.misc

@@ -0,0 +1,3 @@
+Pulp Deb now is guaranteed to use pulpcore's distributed-publication tracking.
+This change deactivates deb's DistributedPublication code and does the data migration.
+Table deletion should be done in a future release do enable zero-downtime upgrades.

pulp_deb/app/migrations/0044_remove_distributedpublication.py

@@ -0,0 +1,65 @@
+import logging
+from itertools import islice
+
+from django.db import migrations, transaction
+
+logger = logging.getLogger(__name__)
+BATCH_SIZE = 5000
+
+
+def copy_to_core(apps, schema_editor):
+    """Best-effort data backfill."""
+    try:
+        with transaction.atomic(using=schema_editor.connection.alias):
+            DebDistributedPublication = apps.get_model("deb", "DistributedPublication")
+            CoreDistributedPublication = apps.get_model("core", "DistributedPublication")
+            db_alias = schema_editor.connection.alias
+            rows = (
+                DebDistributedPublication.objects.using(db_alias)
+                .values(
+                    "pulp_id",
+                    "pulp_created",
+                    "pulp_last_updated",
+                    "expires_at",
+                    "distribution_id",
+                    "publication_id",
+                )
+                .iterator(chunk_size=BATCH_SIZE)
+            )
+            created_field = CoreDistributedPublication._meta.get_field("pulp_created")
+            updated_field = CoreDistributedPublication._meta.get_field("pulp_last_updated")
+            auto_now_add = created_field.auto_now_add
+            auto_now = updated_field.auto_now
+            try:
+                # bulk_create normally refreshes these fields; preserve their legacy values.
+                created_field.auto_now_add = False
+                updated_field.auto_now = False
+                while True:
+                    batch = list(islice(rows, BATCH_SIZE))
+                    if not batch:
+                        break
+                    CoreDistributedPublication.objects.using(db_alias).bulk_create(
+                        [CoreDistributedPublication(**row) for row in batch],
+                        batch_size=BATCH_SIZE,
+                        ignore_conflicts=True,
+                    )
+            finally:
+                created_field.auto_now_add = auto_now_add
+                updated_field.auto_now = auto_now
+    except Exception:
+        logger.warning(
+            "Could not copy distributed publication data to pulpcore; "
+            "continuing because the legacy table is retained.",
+            exc_info=True,
+        )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0149_distributedpublication"),
+        ("deb", "0043_alter_release_unique_together_and_more"),
+    ]
+
+    operations = [
+        migrations.RunPython(copy_to_core),
+    ]

pulp_deb/app/models/publication.py

@@ -1,42 +1,17 @@
-from contextlib import suppress
-from datetime import timedelta
-
 from django.contrib.postgres.fields import ArrayField
 from django.db import models
-from django.utils import timezone
-from django_lifecycle import AFTER_CREATE, AFTER_UPDATE, hook
 
 from pulpcore.plugin.models import (
     AutoAddObjPermsMixin,
     BaseModel,
     Distribution,
     Publication,
-    PublishedArtifact,
-    RepositoryVersion,
 )
 
 from pulp_deb.app.constants import LAYOUT_CHOICES, LAYOUT_TYPES
 from pulp_deb.app.models.signing_service import AptReleaseSigningService
 
 BOOL_CHOICES = [(True, "yes"), (False, "no")]
-PUBLICATION_CACHE_DURATION = timedelta(days=3)
-
-
-def latest_publication(repo_pk):
-    """
-    Find the latest publication for a repository.
-
-    This function is based on the logic in pulpcore's content handler.
-
-    https://github.com/pulp/pulpcore/blob/3bfd35c76e29944b622d275be52c0d5ebbdfbf72/pulpcore/content/handler.py#L601-L607
-    """
-    versions = RepositoryVersion.objects.filter(repository=repo_pk)
-    with suppress(Publication.DoesNotExist):
-        return (
-            Publication.objects.filter(repository_version__in=versions, complete=True)
-            .latest("repository_version", "pulp_created")
-            .cast()
-        )
 
 
 class VerbatimPublication(Publication, AutoAddObjPermsMixin):
@@ -48,12 +23,6 @@ class VerbatimPublication(Publication, AutoAddObjPermsMixin):
 
     TYPE = "verbatim-publication"
 
-    @hook(AFTER_UPDATE, when="complete", has_changed=True, is_now=True)
-    def set_distributed_publication(self):
-        for distro in AptDistribution.objects.filter(repository__pk=self.repository.pk):
-            if self == latest_publication(self.repository.pk):
-                DistributedPublication(distribution=distro, publication=self).save()
-
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
         permissions = [
@@ -78,12 +47,6 @@ class AptPublication(Publication, AutoAddObjPermsMixin):
         AptReleaseSigningService, on_delete=models.PROTECT, null=True
     )
 
-    @hook(AFTER_UPDATE, when="complete", has_changed=True, is_now=True)
-    def set_distributed_publication(self):
-        for distro in AptDistribution.objects.filter(repository__pk=self.repository.pk):
-            if self == latest_publication(self.repository.pk):
-                DistributedPublication(distribution=distro, publication=self).save()
-
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
         permissions = [
@@ -99,35 +62,6 @@ class AptDistribution(Distribution, AutoAddObjPermsMixin):
     TYPE = "apt-distribution"
     SERVE_FROM_PUBLICATION = True
 
-    @hook(AFTER_CREATE)
-    @hook(AFTER_UPDATE, when="publication", has_changed=True, is_not=None)
-    @hook(AFTER_UPDATE, when="repository", has_changed=True, is_not=None)
-    def set_distributed_publication(self):
-        if self.publication:
-            DistributedPublication(distribution=self, publication=self.publication)
-        elif self.repository:
-            if publication := latest_publication(self.repository):
-                DistributedPublication(distribution=self, publication=publication).save()
-
-    def content_handler(self, path):
-        recent_dp = self.distributedpublication_set.filter(
-            models.Q(expires_at__gte=timezone.now()) | models.Q(expires_at__isnull=True)
-        ).order_by("pulp_created")
-        pa = (
-            PublishedArtifact.objects.filter(
-                relative_path=path, publication__distributedpublication__pk__in=recent_dp
-            )
-            .order_by("-publication__distributedpublication__pulp_created")
-            .select_related(
-                "content_artifact",
-                "content_artifact__artifact",
-            )
-        ).first()
-
-        if pa:
-            return pa.content_artifact
-        return
-
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
         permissions = [
@@ -149,11 +83,3 @@ class DistributedPublication(BaseModel):
     distribution = models.ForeignKey(Distribution, on_delete=models.CASCADE)
     publication = models.ForeignKey(Publication, on_delete=models.CASCADE)
     expires_at = models.DateTimeField(null=True)
-
-    @hook(AFTER_CREATE)
-    def cleanup(self):
-        """Set expires_at on any older DistributedPublication and cleanup any expired ones."""
-        DistributedPublication.objects.filter(expires_at__lt=timezone.now()).delete()
-        DistributedPublication.objects.exclude(pk=self.pk).filter(
-            distribution=self.distribution, expires_at__isnull=True
-        ).update(expires_at=(timezone.now() + PUBLICATION_CACHE_DURATION))

pulp_deb/tests/unit/test_distributed_publication_migration.py

@@ -0,0 +1,79 @@
+import unittest
+from importlib import import_module
+from types import SimpleNamespace
+from uuid import uuid4
+
+from django.db import connection
+from django.db.migrations.loader import MigrationLoader
+from django.test import TestCase
+from django.utils import timezone
+
+from pulp_deb.app.models import AptDistribution, AptPublication, AptRepository
+
+
+class TestDistributedPublicationMigration(TestCase):
+    @unittest.skip("FIXME: remove when DistributedPublication table is dropped")
+    def test_copies_all_rows(self):
+        migration = import_module("pulp_deb.app.migrations.0044_remove_distributedpublication")
+        # Create Repository, Publication and Distribution
+        repository = AptRepository.objects.create(name=f"migration-{uuid4()}")
+        repository_version = repository.latest_version()
+        publication = AptPublication.objects.create(repository_version=repository_version)
+        distribution = AptDistribution.objects.create(
+            name=f"migration-{uuid4()}",
+            base_path=f"migration-{uuid4()}",
+            publication=publication,
+        )
+
+        # Get models from before the migration
+        migration_apps = (
+            MigrationLoader(connection)
+            .project_state(
+                [
+                    ("core", "0149_distributedpublication"),
+                    ("deb", "0043_alter_release_unique_together_and_more"),
+                ]
+            )
+            .apps
+        )
+        DebDistributedPublication = migration_apps.get_model("deb", "DistributedPublication")
+        CoreDistributedPublication = migration_apps.get_model("core", "DistributedPublication")
+
+        # Create a set of deb's DistributedPublicaitions
+        source_rows = [
+            DebDistributedPublication(
+                distribution_id=distribution.pk,
+                publication_id=publication.pk,
+                expires_at=timezone.now(),
+            )
+            for _ in range(migration.BATCH_SIZE + 2)
+        ]
+        DebDistributedPublication.objects.bulk_create(source_rows)
+        source_ids = [row.pulp_id for row in source_rows]
+
+        # Migrate data and assert it's copied correctly
+        migration.copy_to_core(
+            migration_apps,
+            SimpleNamespace(connection=connection),
+        )
+        source_data = set(
+            DebDistributedPublication.objects.filter(pulp_id__in=source_ids).values_list(
+                "pulp_id",
+                "pulp_created",
+                "pulp_last_updated",
+                "expires_at",
+                "distribution_id",
+                "publication_id",
+            )
+        )
+        copied_data = set(
+            CoreDistributedPublication.objects.filter(pulp_id__in=source_ids).values_list(
+                "pulp_id",
+                "pulp_created",
+                "pulp_last_updated",
+                "expires_at",
+                "distribution_id",
+                "publication_id",
+            )
+        )
+        self.assertEqual(copied_data, source_data)