initial commit for multidb

.ci/ansible/start_container.yaml

@@ -76,6 +76,13 @@
           retries: 2
           delay: 5
 
+        - name: "Wait for postgres-satellite"
+          ansible.builtin.wait_for:
+            host: "postgres-satellite"
+            port: 5432
+            timeout: 30
+          when: "multi_db_test | default(false)"
+
         - name: "Wait for Pulp"
           ansible.builtin.uri:
             url: "http://pulp{{ pulp_scenario_settings.api_root | default(pulp_settings.api_root | default('\/pulp\/', True), True) }}api/v3/status/"

.github/workflows/ci.yml

@@ -107,7 +107,7 @@ jobs:
     uses: "./.github/workflows/test.yml"
     with:
       matrix_env: |
-        [{"TEST": "pulp"}, {"TEST": "azure"}, {"TEST": "s3"}, {"TEST": "lowerbounds"}]
+        [{"TEST": "pulp"}, {"TEST": "azure"}, {"TEST": "s3"}, {"TEST": "lowerbounds"}, {"TEST": "multi_db"}]
 
   deprecations:
     runs-on: "ubuntu-latest"

.github/workflows/scripts/before_install.sh

@@ -123,6 +123,20 @@ pulp_scenario_env: {}
 VARSYAML
 fi
 
+if [ "$TEST" = "multi_db" ]; then
+  cat >> .ci/ansible/vars/main.yaml << VARSYAML
+  - name: "postgres-satellite"
+    image: "docker.io/library/postgres:16"
+    env:
+      POSTGRES_USER: "postgres"
+      POSTGRES_PASSWORD: "postgres"
+      POSTGRES_DB: "pulp"
+multi_db_test: true
+pulp_scenario_settings: null
+pulp_scenario_env: {}
+VARSYAML
+fi
+
 cat >> .ci/ansible/vars/main.yaml << VARSYAML
 ...
 VARSYAML

.github/workflows/scripts/script.sh

@@ -129,7 +129,11 @@ cmd_user_prefix bash -c "django-admin makemigrations file --check --dry-run"
 cmd_user_prefix bash -c "django-admin makemigrations certguard --check --dry-run"
 
 # Run unit tests.
-cmd_user_prefix bash -c "PULP_DATABASES__default__USER=postgres pytest -v -r sx --color=yes --suppress-no-test-exit-code -p no:pulpcore --durations=20 --pyargs pulpcore.tests.unit"
+MULTI_DB_ENV=""
+if [[ "$TEST" == "multi_db" ]]; then
+  MULTI_DB_ENV="PULP_DATABASES__data_1__ENGINE=django.db.backends.postgresql PULP_DATABASES__data_1__NAME=pulp PULP_DATABASES__data_1__USER=postgres PULP_DATABASES__data_1__PASSWORD=postgres PULP_DATABASES__data_1__HOST=postgres-satellite PULP_DATABASES__data_1__PORT=5432 PULP_DATABASE_ROUTERS='[\"pulpcore.app.db_router.PulpDomainRouter\"]'"
+fi
+cmd_user_prefix bash -c "PULP_DATABASES__default__USER=postgres $MULTI_DB_ENV pytest -v -r sx --color=yes --suppress-no-test-exit-code -p no:pulpcore --durations=20 --pyargs pulpcore.tests.unit"
 cmd_user_prefix bash -c "PULP_DATABASES__default__USER=postgres pytest -v -r sx --color=yes --suppress-no-test-exit-code -p no:pulpcore --durations=20 --pyargs pulp_file.tests.unit"
 cmd_user_prefix bash -c "PULP_DATABASES__default__USER=postgres pytest -v -r sx --color=yes --suppress-no-test-exit-code -p no:pulpcore --durations=20 --pyargs pulp_certguard.tests.unit"
 # Run functional tests

pulpcore/app/apps.py

@@ -1,3 +1,4 @@
+import logging
 import random
 from collections import defaultdict
 from gettext import gettext as _
@@ -6,8 +7,8 @@
 from django import apps
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
-from django.db import connection, transaction
-from django.db.models.signals import post_migrate, pre_migrate
+from django.db import connection, connections, transaction
+from django.db.models.signals import post_delete, post_migrate, post_save, pre_migrate
 from django.utils.module_loading import module_has_submodule
 
 from pulpcore.exceptions.plugin import MissingPlugin
@@ -255,14 +256,39 @@ def ready(self):
         post_migrate.connect(
             _populate_system_id, sender=self, dispatch_uid="populate_system_id_identifier"
         )
+        post_migrate.connect(
+            _ensure_domains_replicated,
+            sender=self,
+            dispatch_uid="ensure_domains_replicated_identifier",
+        )
         post_migrate.connect(
             _populate_artifact_serving_distribution,
             sender=self,
             dispatch_uid="populate_artifact_serving_distribution_identifier",
         )
+        from pulpcore.app.domain_sync import on_domain_post_delete, on_domain_post_save
+        from pulpcore.app.models import Domain
+
+        post_save.connect(
+            on_domain_post_save, sender=Domain, dispatch_uid="replicate_domain_post_save"
+        )
+        post_delete.connect(
+            on_domain_post_delete, sender=Domain, dispatch_uid="replicate_domain_post_delete"
+        )
+
+        from pulpcore.app.db_router import is_multi_db_routing_active
+
+        if is_multi_db_routing_active():
+            from pulpcore.app.role_util import on_any_model_post_delete
+
+            post_delete.connect(
+                on_any_model_post_delete, dispatch_uid="cleanup_cross_plane_roles_post_delete"
+            )
 
 
 def _clean_app_status(sender, apps, verbosity, **kwargs):
+    if kwargs.get("using", "default") != "default":
+        return
     from django.contrib.postgres.functions import TransactionNow
     from django.db.models import F
 
@@ -276,6 +302,9 @@ def _clean_app_status(sender, apps, verbosity, **kwargs):
 
 
 def _populate_access_policies(sender, apps, verbosity, **kwargs):
+    if kwargs.get("using", "default") != "default":
+        return
+
     from pulpcore.app.util import get_view_urlpattern
     from pulpcore.app.viewsets import LoginViewSet
 
@@ -320,12 +349,16 @@ def _populate_access_policies(sender, apps, verbosity, **kwargs):
 
 
 def _populate_system_id(sender, apps, verbosity, **kwargs):
+    if kwargs.get("using", "default") != "default":
+        return
     SystemID = apps.get_model("core", "SystemID")
     if not SystemID.objects.exists():
         SystemID().save()
 
 
 def _ensure_default_domain(sender, **kwargs):
+    if kwargs.get("using", "default") != "default":
+        return
     table_names = connection.introspection.table_names()
     if "core_domain" in table_names:
         from pulpcore.app.util import get_default_domain
@@ -343,7 +376,29 @@ def _ensure_default_domain(sender, **kwargs):
             default.save(skip_hooks=True)
 
 
+def _ensure_domains_replicated(sender, **kwargs):
+    using = kwargs.get("using", "default")
+    if using == "default":
+        return
+    if "core_domain" not in connections[using].introspection.table_names():
+        return
+    from pulpcore.app.domain_sync import reconcile_domains_to_alias
+
+    try:
+        reconcile_domains_to_alias(using)
+    except Exception:
+        logging.getLogger(__name__).error(
+            "Reconciling Domain rows to alias '%s' failed during migration. Data-plane objects "
+            "created on this alias by later migrations/post_migrate hooks that FK to Domain may "
+            "fail until 'pulpcore-manager sync-domains' is run.",
+            using,
+            exc_info=True,
+        )
+
+
 def _populate_roles(sender, apps, verbosity, **kwargs):
+    if kwargs.get("using", "default") != "default":
+        return
     role_prefix = f"{sender.label}."
     # collect all plugin defined roles
     desired_roles = {}
@@ -403,6 +458,7 @@ def _get_permission(perm):
 
 
 def _populate_artifact_serving_distribution(sender, apps, verbosity, **kwargs):
+    alias = kwargs.get("using", "default")
     if (
         settings.STORAGES["default"]["BACKEND"] == "pulpcore.app.models.storage.FileSystem"
         or not settings.REDIRECT_TO_OBJECT_STORAGE
@@ -415,15 +471,17 @@ def _populate_artifact_serving_distribution(sender, apps, verbosity, **kwargs):
                 print(_("ArtifactDistribution model does not exist. Skipping initialization."))
             return
         try:
-            ArtifactDistribution.objects.get()
+            ArtifactDistribution.objects.using(alias).get()
         except ArtifactDistribution.DoesNotExist:
             name = f"{random.getrandbits(256):x}"
-            with transaction.atomic():
-                content_guard, _created = ContentRedirectContentGuard.objects.get_or_create(
+            with transaction.atomic(using=alias):
+                content_guard, _created = ContentRedirectContentGuard.objects.using(
+                    alias
+                ).get_or_create(
                     name=name,
                     pulp_type="core.content_redirect",
                 )
-                _dist, _created = ArtifactDistribution.objects.get_or_create(
+                _dist, _created = ArtifactDistribution.objects.using(alias).get_or_create(
                     name=name,
                     pulp_type="core.artifact",
                     defaults={"base_path": name, "content_guard": content_guard},

pulpcore/app/contexts.py

@@ -12,6 +12,7 @@
 current_pulp_api_version = ContextVar(
     "current_pulp_api_version", default=settings.REST_FRAMEWORK.get("DEFAULT_VERSION", "v3")
 )
+_current_migration_alias = ContextVar("current_migration_alias", default=None)
 
 
 @contextmanager
@@ -45,6 +46,15 @@ def with_domain(domain):
         _current_domain.reset(token)
 
 
+@contextmanager
+def with_migration_alias(alias):
+    token = _current_migration_alias.set(alias)
+    try:
+        yield
+    finally:
+        _current_migration_alias.reset(token)
+
+
 @contextmanager
 def with_task_context(task):
     with with_domain(task.pulp_domain), with_guid(task.logging_cid), with_user(task.user):

pulpcore/app/db_router.py

@@ -0,0 +1,86 @@
+import logging
+
+from django.apps import apps as django_apps
+from django.db import router as django_router
+
+from pulpcore.app.contexts import _current_migration_alias
+from pulpcore.app.util import get_domain
+
+logger = logging.getLogger(__name__)
+
+CONTROL_PLANE_LABELS = frozenset(
+    {
+        "core.domain",
+        "core.task",
+        "core.taskgroup",
+        "core.taskschedule",
+        "core.createdresource",
+        "core.appstatus",
+        "core.systemid",
+        "core.accesspolicy",
+        "core.role",
+        "core.userrole",
+        "core.grouprole",
+        "core.progressreport",
+        "core.groupprogressreport",
+        "core.migrationstatus",
+        "core.domainmove",
+        "core.profileartifact",
+        "core.signingservice",
+        "core.asciiarmoreddetachedsigningservice",
+        "container.manifestsigningservice",
+        "rpm.rpmpackagesigningservice",
+    }
+)
+
+CONTROL_PLANE_APPS = frozenset({"auth", "contenttypes", "admin", "sessions"})
+
+
+def _database_alias(domain):
+    if "database_alias" in domain.__dict__:
+        return domain.__dict__["database_alias"]
+    return "default"
+
+
+class PulpDomainRouter:
+    def _is_control_plane(self, model):
+        label = f"{model._meta.app_label}.{model._meta.model_name}"
+        return label in CONTROL_PLANE_LABELS or model._meta.app_label in CONTROL_PLANE_APPS
+
+    def _resolve_db(self, model, **hints):
+        if model._meta.apps is not django_apps:
+            migration_alias = _current_migration_alias.get()
+            if migration_alias is not None:
+                return migration_alias
+
+        if self._is_control_plane(model):
+            return "default"
+
+        instance = hints.get("instance")
+        if instance is not None:
+            if "pulp_domain_id" in instance.__dict__:
+                domain = instance._state.fields_cache.get("pulp_domain")
+                if domain is not None:
+                    return _database_alias(domain)
+
+        domain = get_domain()
+        if domain is not None:
+            return _database_alias(domain)
+
+        return "default"
+
+    def db_for_read(self, model, **hints):
+        return self._resolve_db(model, **hints)
+
+    def db_for_write(self, model, **hints):
+        return self._resolve_db(model, **hints)
+
+    def allow_relation(self, obj1, obj2, **hints):
+        return True
+
+    def allow_migrate(self, db, app_label, model_name=None, **hints):
+        return True
+
+
+def is_multi_db_routing_active():
+    return any(isinstance(r, PulpDomainRouter) for r in django_router.routers)