Fix migration of multiple plugins simultaneously

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

CHANGES/5978.bugfix

@@ -0,0 +1 @@
+Fix migration of multiple plugins.

pulp_2to3_migration/app/migration.py

@@ -54,10 +54,12 @@ async def migrate_repositories(plan):
         message='Creating repositories in Pulp 3', code='creating.repositories', total=0
     )
     with ProgressReport(**progress_data) as pb:
-        pulp2repos_qs = Pulp2Repository.objects.filter(pulp3_repository_version=None,
-                                                       not_in_plan=False)
-
         for plugin in plan.get_plugin_plans():
+            pulp2repos_qs = Pulp2Repository.objects.filter(
+                pulp3_repository_version=None,
+                not_in_plan=False,
+                type=plugin.type,
+            )
             repos_to_create = plugin.get_repo_creation_setup()
 
             # no specific migration plan for repositories
@@ -177,17 +179,19 @@ async def migrate_repo_distributor(pb, dist_migrator, pulp2dist, repo_version=No
         message='Migrating distributors to Pulp 3', code='migrating.distributors', total=0
     )
     with ProgressReport(**progress_data) as pb:
-        pulp2distributors_qs = Pulp2Distributor.objects.filter(
-            pulp3_distribution=None,
-            pulp3_publication=None,
-            not_in_plan=False)
-        pb.total = pulp2distributors_qs.count()
-        pb.save()
+        for plugin in plan.get_plugin_plans():
+            pulp2distributors_qs = Pulp2Distributor.objects.filter(
+                pulp3_distribution=None,
+                pulp3_publication=None,
+                not_in_plan=False,
+                pulp2_type_id=plugin.type
+            )
+            pb.total = pulp2distributors_qs.count()
+            pb.save()
 
-        # gather all needed plugin distributor migrators
-        distributor_migrators = {}
+            # gather all needed plugin distributor migrators
+            distributor_migrators = {}
 
-        for plugin in plan.get_plugin_plans():
             distributor_migrators.update(**plugin.migrator.distributor_migrators)
 
             pulp3_repo_setup = plugin.get_repo_creation_setup()

pulp_2to3_migration/app/models/base.py

@@ -52,18 +52,6 @@ def get_repositories(self):
         """
         return self.plan_view.all_repositories_to_migrate
 
-    def get_importers_repos(self):
-        """
-        Pulp2 repositories to migrate importers for or empty list if all should be migrated.
-        """
-        return self.plan_view.all_repositories_importers_to_migrate
-
-    def get_distributors_repos(self):
-        """
-        Pulp2 repositories to migrate distributors for or empty list if all should be migrated.
-        """
-        return self.plan_view.all_repositories_distributors_to_migrate
-
     def get_missing_resources(self):
         """
         Return a dict of any resources listed in the plan but missing from Pulp 2.
@@ -165,6 +153,24 @@ def __init__(self, plugin_migration_plan):
 
         self._parse_plugin_plan(plugin_migration_plan)
 
+    def get_repositories(self):
+        """
+        Return a list of pulp2 repositories to migrate or empty list if all should be migrated.
+        """
+        return self.repositories_to_migrate
+
+    def get_importers_repos(self):
+        """
+        Pulp2 repositories to migrate importers for or empty list if all should be migrated.
+        """
+        return self.repositories_importers_to_migrate
+
+    def get_distributors_repos(self):
+        """
+        Pulp2 repositories to migrate distributors for or empty list if all should be migrated.
+        """
+        return self.repositories_distributors_to_migrate
+
     def get_repo_creation_setup(self):
         """
         Returns a structure that defines the Pulp 3 repositories to be created.

pulp_2to3_migration/app/pre_migration.py

@@ -204,50 +204,46 @@ async def pre_migrate_all_without_content(plan, type_to_repo_ids, repo_id_to_typ
     _logger.debug('Pre-migrating Pulp 2 repositories')
 
     with ProgressReport(message='Processing Pulp 2 repositories, importers, distributors',
-                        code='processing.repositories') as pb:
-        repos = plan.get_repositories()
-
-        # filter by repo type
-        repos_to_check = []
-        for repo_type in plan.get_plugins():
-            repos_to_check += type_to_repo_ids[repo_type]
-
-        mongo_repo_q = mongo_Q(repo_id__in=repos_to_check)
-        mongo_repo_qs = Repository.objects(mongo_repo_q)
-
-        pb.total = mongo_repo_qs.count()
-        pb.save()
-
-        importers_repos = plan.get_importers_repos()
-        distributors_repos = plan.get_distributors_repos()
-
-        importer_types = []
-        distributor_types = []
-        for plugin in plan.get_plugin_plans():
-            distributor_types.extend(plugin.migrator.distributor_migrators.keys())
-            importer_types.extend(plugin.migrator.importer_migrators.keys())
-
-        for repo_data in mongo_repo_qs.only('id',
-                                            'repo_id',
-                                            'last_unit_added',
-                                            'last_unit_removed',
-                                            'description',
-                                            'notes'):
-            repo = None
-            repo_id = repo_data.repo_id
-            with transaction.atomic():
-                if not repos or repos and repo_id in repos:
-                    repo = await pre_migrate_repo(repo_data, repo_id_to_type)
-                # do not pre-migrate importers/distributors in case of special repo setup
-                # and no importers/distributors were specified in the MP
-                if not repos or repos and importers_repos:
-                    await pre_migrate_importer(repo_id, importers_repos, importer_types, repo)
-                if not repos or repos and distributors_repos:
-                    await pre_migrate_distributor(repo_id, distributors_repos, distributor_types,
-                                                  repo)
-                if repo:
-                    await pre_migrate_repocontent(repo)
-            pb.increment()
+                        code='processing.repositories', total=0) as pb:
+
+        for plugin_plan in plan.get_plugin_plans():
+            repos = plugin_plan.get_repositories()
+            # filter by repo type
+            repos_to_check = type_to_repo_ids[plugin_plan.type]
+
+            mongo_repo_q = mongo_Q(repo_id__in=repos_to_check)
+            mongo_repo_qs = Repository.objects(mongo_repo_q)
+
+            pb.total += mongo_repo_qs.count()
+            pb.save()
+
+            importers_repos = plugin_plan.get_importers_repos()
+            distributors_repos = plugin_plan.get_distributors_repos()
+
+            distributor_types = list(plugin_plan.migrator.distributor_migrators.keys())
+            importer_types = list(plugin_plan.migrator.importer_migrators.keys())
+
+            for repo_data in mongo_repo_qs.only('id',
+                                                'repo_id',
+                                                'last_unit_added',
+                                                'last_unit_removed',
+                                                'description',
+                                                'notes'):
+                repo = None
+                repo_id = repo_data.repo_id
+                with transaction.atomic():
+                    if not repos or repos and repo_id in repos:
+                        repo = await pre_migrate_repo(repo_data, repo_id_to_type)
+                    # do not pre-migrate importers/distributors in case of special repo setup
+                    # and no importers/distributors were specified in the MP
+                    if not repos or repos and importers_repos:
+                        await pre_migrate_importer(repo_id, importers_repos, importer_types, repo)
+                    if not repos or repos and distributors_repos:
+                        await pre_migrate_distributor(
+                            repo_id, distributors_repos, distributor_types, repo)
+                    if repo:
+                        await pre_migrate_repocontent(repo)
+                pb.increment()
 
 
 async def pre_migrate_repo(record, repo_id_to_type):
@@ -427,36 +423,35 @@ async def mark_removed_resources(plan, type_to_repo_ids):
         plan(MigrationPlan): A Migration Plan
         type_to_repo_ids(dict): A mapping from a pulp 2 repo type to a list of pulp 2 repo_ids
     """
-    repos = plan.get_repositories()
+    for plugin_plan in plan.get_plugin_plans():
+        repos = plugin_plan.get_repositories()
 
-    # filter by repo type
-    repos_to_consider = []
-
-    for repo_type in plan.get_plugins():
-        repos_to_consider += type_to_repo_ids[repo_type]
+        # filter by repo type
+        repos_to_consider = type_to_repo_ids[plugin_plan.type]
 
-    # in case only certain repositories are specified in the migration plan
-    if repos:
-        repos_to_consider = set(repos).intersection(repos_to_consider)
+        # in case only certain repositories are specified in the migration plan
+        if repos:
+            repos_to_consider = set(repos).intersection(repos_to_consider)
 
-    mongo_repo_q = mongo_Q(repo_id__in=repos_to_consider)
+        mongo_repo_q = mongo_Q(repo_id__in=repos_to_consider)
 
-    mongo_repo_object_ids = set(str(i.id) for i in Repository.objects(mongo_repo_q).only('id'))
+        mongo_repo_object_ids = set(
+            str(i.id) for i in Repository.objects(mongo_repo_q).only('id'))
 
-    psql_repo_types = plan.get_plugins()
-    premigrated_repos = Pulp2Repository.objects.filter(type__in=psql_repo_types)
-    premigrated_repo_object_ids = set(premigrated_repos.values_list('pulp2_object_id',
-                                                                    flat=True))
-    removed_repo_object_ids = premigrated_repo_object_ids - mongo_repo_object_ids
+        premigrated_repos = Pulp2Repository.objects.filter(type=plugin_plan.type)
+        premigrated_repo_object_ids = set(premigrated_repos.values_list('pulp2_object_id',
+                                                                        flat=True))
+        removed_repo_object_ids = premigrated_repo_object_ids - mongo_repo_object_ids
 
-    removed_repos = []
-    for pulp2repo in Pulp2Repository.objects.filter(pulp2_object_id__in=removed_repo_object_ids):
-        pulp2repo.not_in_plan = True
-        removed_repos.append(pulp2repo)
+        removed_repos = []
+        for pulp2repo in Pulp2Repository.objects.filter(
+                pulp2_object_id__in=removed_repo_object_ids):
+            pulp2repo.not_in_plan = True
+            removed_repos.append(pulp2repo)
 
-    Pulp2Repository.objects.bulk_update(objs=removed_repos,
-                                        fields=['not_in_plan'],
-                                        batch_size=1000)
+        Pulp2Repository.objects.bulk_update(objs=removed_repos,
+                                            fields=['not_in_plan'],
+                                            batch_size=1000)
 
 
 async def delete_old_resources(plan):

pulp_2to3_migration/app/serializers.py

@@ -156,6 +156,7 @@ class Pulp2RepositoriesSerializer(ModelSerializer):
     )
     pulp2_object_id = serializers.CharField(max_length=255)
     pulp2_repo_id = serializers.CharField()
+    type = serializers.CharField()
     is_migrated = serializers.BooleanField(default=False)
     not_in_plan = serializers.BooleanField(default=False)
 
@@ -232,6 +233,7 @@ class Meta:
         fields = ModelSerializer.Meta.fields + (
             "pulp2_object_id",
             "pulp2_repo_id",
+            "type",
             "is_migrated",
             "not_in_plan",
             "pulp3_repository_version",