diff --git a/.github/workflows/scripts/script.sh b/.github/workflows/scripts/script.sh index 4575280ae93..05dd1be48cb 100755 --- a/.github/workflows/scripts/script.sh +++ b/.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" +# data_1 is a second database on the same local postgres server used by "default" -- Django's +# test runner creates/tears down its own "test_..." database for it, so multi-db unit tests +# run as part of the normal suite without a dedicated satellite service/CI job. +MULTI_DB_ENV="PULP_DATABASES__data_1__ENGINE=django.db.backends.postgresql PULP_DATABASES__data_1__NAME=pulp_data_1 PULP_DATABASES__data_1__USER=postgres PULP_DATABASE_ROUTERS='[\"pulpcore.app.db_router.PulpDomainRouter\"]'" +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 diff --git a/pulpcore/app/apps.py b/pulpcore/app/apps.py index eb80e452d74..ea32c9baf65 100644 --- a/pulpcore/app/apps.py +++ b/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 Domain rows are reconciled to this alias.", + 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}, diff --git a/pulpcore/app/contexts.py b/pulpcore/app/contexts.py index d86b9ce6f11..a18c30191d1 100644 --- a/pulpcore/app/contexts.py +++ b/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): diff --git a/pulpcore/app/db_router.py b/pulpcore/app/db_router.py new file mode 100644 index 00000000000..e24f9355215 --- /dev/null +++ b/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.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" + + # Use __dict__ / fields_cache, not getattr/hasattr. FK descriptors can + # recurse into this router during instance construction or issue an extra query. + 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) diff --git a/pulpcore/app/domain_sync.py b/pulpcore/app/domain_sync.py new file mode 100644 index 00000000000..25f4756c522 --- /dev/null +++ b/pulpcore/app/domain_sync.py @@ -0,0 +1,171 @@ +import logging +import time + +from django.conf import settings + +logger = logging.getLogger(__name__) + +REPLICATION_RETRY_ATTEMPTS = 3 +REPLICATION_RETRY_BACKOFF = 1 + + +def satellite_aliases(): + return [alias for alias in settings.DATABASES if alias != "default"] + + +def _target_aliases(domain): + if domain.name == "default": + return satellite_aliases() + if domain.database_alias in satellite_aliases(): + return [domain.database_alias] + return [] + + +def domain_field_values(domain): + return {field.attname: getattr(domain, field.attname) for field in domain._meta.concrete_fields} + + +def _comparable_domain_field_values(domain): + values = domain_field_values(domain) + values.pop("pulp_last_updated", None) + return values + + +def replicate_domain_save(domain, using=None, attempts=REPLICATION_RETRY_ATTEMPTS): + values = domain_field_values(domain) + pulp_id = values.pop("pulp_id") + for alias in _target_aliases(domain): + if alias == using: + continue + _replicate_one_save(alias, pulp_id, values, attempts) + + +def replicate_domain_delete(domain, using=None, attempts=REPLICATION_RETRY_ATTEMPTS): + for alias in _target_aliases(domain): + if alias == using: + continue + _replicate_one_delete(alias, domain.pulp_id, attempts) + + +def ensure_domain_on_alias(domain, alias, attempts=REPLICATION_RETRY_ATTEMPTS): + values = domain_field_values(domain) + pulp_id = values.pop("pulp_id") + _replicate_one_save(alias, pulp_id, values, attempts) + + +def reconcile_domains_to_alias(alias, dry_run=False): + from pulpcore.app.models import Domain + + desired_domains = { + domain.pulp_id: domain + for domain in Domain.objects.using("default") + if domain.name == "default" or domain.database_alias == alias + } + desired_ids = set(desired_domains) + + satellite_ids = set(Domain.objects.using(alias).values_list("pulp_id", flat=True)) + + missing = desired_ids - satellite_ids + extra = satellite_ids - desired_ids + stale = set() + for pulp_id in desired_ids & satellite_ids: + satellite_domain = Domain.objects.using(alias).get(pulp_id=pulp_id) + if _comparable_domain_field_values( + desired_domains[pulp_id] + ) != _comparable_domain_field_values(satellite_domain): + stale.add(pulp_id) + + if dry_run: + return {"missing": missing, "extra": extra, "stale": stale} + + for pulp_id in missing | stale: + values = domain_field_values(desired_domains[pulp_id]) + values.pop("pulp_id") + try: + instance = Domain.objects.using(alias).get(pulp_id=pulp_id) + for key, value in values.items(): + setattr(instance, key, value) + except Domain.DoesNotExist: + instance = Domain(pulp_id=pulp_id, **values) + instance.save(using=alias, skip_hooks=True) + for pulp_id in extra: + Domain.objects.using(alias).filter(pulp_id=pulp_id).delete() + + return {"missing": missing, "extra": extra, "stale": stale} + + +def _replicate_one_save(alias, pulp_id, defaults, attempts): + from pulpcore.app.models import Domain + + delay = REPLICATION_RETRY_BACKOFF + for attempt in range(1, attempts + 1): + try: + manager = Domain.objects.using(alias) + try: + instance = manager.get(pulp_id=pulp_id) + for key, value in defaults.items(): + setattr(instance, key, value) + except Domain.DoesNotExist: + instance = Domain(pulp_id=pulp_id, **defaults) + instance.save(using=alias, skip_hooks=True) + return + except Exception: + logger.warning( + "Domain replication to alias '%s' failed (attempt %d/%d) for domain %s.", + alias, + attempt, + attempts, + pulp_id, + exc_info=True, + ) + if attempt < attempts: + time.sleep(delay) + delay *= 2 + logger.error( + "Domain replication to alias '%s' failed after %d attempts for domain %s. " + "Data-plane writes for this domain on that alias will fail until it is reconciled.", + alias, + attempts, + pulp_id, + ) + + +def _replicate_one_delete(alias, pulp_id, attempts): + from pulpcore.app.models import Domain + + delay = REPLICATION_RETRY_BACKOFF + for attempt in range(1, attempts + 1): + try: + Domain.objects.using(alias).filter(pulp_id=pulp_id).delete() + return + except Exception: + logger.warning( + "Domain delete-replication to alias '%s' failed (attempt %d/%d) for domain %s.", + alias, + attempt, + attempts, + pulp_id, + exc_info=True, + ) + if attempt < attempts: + time.sleep(delay) + delay *= 2 + logger.error( + "Domain delete-replication to alias '%s' failed after %d attempts for domain %s. " + "A stale copy of this domain may remain on that alias until it is reconciled.", + alias, + attempts, + pulp_id, + ) + + +def on_domain_post_save(sender, instance, created, using, **kwargs): + if using != "default": + return + replicate_domain_save(instance, using=using) + + +def on_domain_post_delete(sender, instance, using, **kwargs): + if using != "default": + return + replicate_domain_delete(instance, using=using) diff --git a/pulpcore/app/management/commands/analyze-publication.py b/pulpcore/app/management/commands/analyze-publication.py index 07252c96e2d..f796aae91db 100644 --- a/pulpcore/app/management/commands/analyze-publication.py +++ b/pulpcore/app/management/commands/analyze-publication.py @@ -3,7 +3,7 @@ from django.core.management import BaseCommand, CommandError from django.urls import reverse -from pulpcore.app.models import Artifact, Distribution, Publication +from pulpcore.app.models import Artifact, Distribution, Domain, Publication from pulpcore.app.util import get_view_name_for_model @@ -19,6 +19,12 @@ def add_arguments(self, parser): "--distribution-base-path", required=False, help=_("A base_path of a distribution.") ) parser.add_argument("--tabular", action="store_true", help=_("Display as a table")) + parser.add_argument( + "--domain", + default="default", + required=False, + help=_("The pulp domain the publication/distribution belongs to."), + ) def handle(self, *args, **options): """Implement the command.""" @@ -33,17 +39,28 @@ def handle(self, *args, **options): raise CommandError("Must provide either --publication or --distribution-base-path") elif options["publication"] and options["distribution_base_path"]: raise CommandError("Cannot provide both --publication and --distribution-base-path") - elif options["publication"]: - publication = Publication.objects.get(pk=options["publication"]) + + try: + domain = Domain.objects.get(name=options["domain"]) + except Domain.DoesNotExist: + raise CommandError(_("Domain '{name}' does not exist.").format(name=options["domain"])) + alias = domain.database_alias + + if options["publication"]: + publication = Publication.objects.using(alias).get(pk=options["publication"]) else: - distribution = Distribution.objects.get(base_path=options["distribution_base_path"]) + distribution = Distribution.objects.using(alias).get( + base_path=options["distribution_base_path"] + ) if distribution.publication: publication = distribution.publication elif distribution.repository: repository = distribution.repository - publication = Publication.objects.filter( - repository_version__in=repository.versions.all(), complete=True - ).latest("repository_version", "pulp_created") + publication = ( + Publication.objects.using(alias) + .filter(repository_version__in=repository.versions.all(), complete=True) + .latest("repository_version", "pulp_created") + ) published_artifacts = publication.published_artifact.select_related( "content_artifact__artifact" diff --git a/pulpcore/app/management/commands/datarepair-2327.py b/pulpcore/app/management/commands/datarepair-2327.py index 56a59bab969..46af36b57cf 100644 --- a/pulpcore/app/management/commands/datarepair-2327.py +++ b/pulpcore/app/management/commands/datarepair-2327.py @@ -3,11 +3,12 @@ import cryptography from django.conf import settings from django.core.management import BaseCommand -from django.db import connection +from django.db import connections from django.db.models import Q from django.utils.encoding import force_bytes, force_str from pulpcore.app.models import Remote +from pulpcore.app.util import for_each_domain class Command(BaseCommand): @@ -45,64 +46,71 @@ def handle(self, *args, **options): | Q(client_key__isnull=False) ) - number_unencrypted = 0 - number_multi_encrypted = 0 - - for remote_pk in Remote.objects.filter(possibly_affected_remotes).values_list( - "pk", flat=True - ): - try: - remote = Remote.objects.get(pk=remote_pk) - # if we can get the remote successfully, it is either OK or the fields are - # encrypted more than once - except cryptography.fernet.InvalidToken: - # If decryption fails then it probably hasn't been encrypted yet - # get the raw column value, avoiding any Django field handling - with connection.cursor() as cursor: - cursor.execute( - "SELECT username, password, proxy_username, proxy_password, client_key " - "FROM core_remote WHERE pulp_id = %s", - [str(remote_pk)], - ) - row = cursor.fetchone() - - field_values = {} + counts = {"number_unencrypted": 0, "number_multi_encrypted": 0} + + def _repair_for_domain(domain, alias): + for remote_pk in ( + Remote.objects.using(alias) + .filter(possibly_affected_remotes) + .values_list("pk", flat=True) + ): + try: + remote = Remote.objects.using(alias).get(pk=remote_pk) + # if we can get the remote successfully, it is either OK or the fields are + # encrypted more than once + except cryptography.fernet.InvalidToken: + # If decryption fails then it probably hasn't been encrypted yet + # get the raw column value, avoiding any Django field handling + with connections[alias].cursor() as cursor: + cursor.execute( + "SELECT username, password, proxy_username, proxy_password, " + "client_key FROM core_remote WHERE pulp_id = %s", + [str(remote_pk)], + ) + row = cursor.fetchone() + + field_values = {} + + for field, value in zip(fields, row): + field_values[field] = value - for field, value in zip(fields, row): - field_values[field] = value - - if not dry_run: - Remote.objects.filter(pk=remote_pk).update(**field_values) - number_unencrypted += 1 - else: - times_decrypted = 0 - keep_trying = True - needs_update = False - - while keep_trying: - for field in fields: - field_value = getattr(remote, field) # value gets decrypted once on access - if not field_value: - continue - - try: - # try to decrypt it again - field_value = force_str(fernet.decrypt(force_bytes(field_value))) - # it was decrypted successfully again time, so it was probably - # encrypted multiple times over. lets re-set the value with the - # newly decrypted value - setattr(remote, field, field_value) - needs_update = True - except cryptography.fernet.InvalidToken: - # couldn't be decrypted again, stop here - keep_trying = False - - times_decrypted += 1 - - if needs_update: if not dry_run: - remote.save() - number_multi_encrypted += 1 + Remote.objects.using(alias).filter(pk=remote_pk).update(**field_values) + counts["number_unencrypted"] += 1 + else: + times_decrypted = 0 + keep_trying = True + needs_update = False + + while keep_trying: + for field in fields: + # value gets decrypted once on access + field_value = getattr(remote, field) + if not field_value: + continue + + try: + # try to decrypt it again + field_value = force_str(fernet.decrypt(force_bytes(field_value))) + # it was decrypted successfully again time, so it was probably + # encrypted multiple times over. lets re-set the value with the + # newly decrypted value + setattr(remote, field, field_value) + needs_update = True + except cryptography.fernet.InvalidToken: + # couldn't be decrypted again, stop here + keep_trying = False + + times_decrypted += 1 + + if needs_update: + if not dry_run: + remote.save() + counts["number_multi_encrypted"] += 1 + + for_each_domain(_repair_for_domain) + number_unencrypted = counts["number_unencrypted"] + number_multi_encrypted = counts["number_multi_encrypted"] if dry_run: print("Remotes with un-encrypted fields: {}".format(number_unencrypted)) diff --git a/pulpcore/app/management/commands/datarepair.py b/pulpcore/app/management/commands/datarepair.py index ab998f45752..d561723da85 100644 --- a/pulpcore/app/management/commands/datarepair.py +++ b/pulpcore/app/management/commands/datarepair.py @@ -3,11 +3,12 @@ import cryptography from django.conf import settings from django.core.management import BaseCommand, CommandError -from django.db import connection +from django.db import connections from django.db.models import Q from django.utils.encoding import force_bytes, force_str from pulpcore.app import models +from pulpcore.app.util import domain_db, for_each_domain class Command(BaseCommand): @@ -50,57 +51,62 @@ def repair_7272(self, options): for domain in models.Domain.objects.all(): has_printed_domain = False - for repo in models.Repository.objects.filter(pulp_domain=domain): - for rv in models.RepositoryVersion.objects.filter(repository=repo): - needs_fix = False - if rv.content_ids is not None: - cached_id_set = set(rv.content_ids) - repositorycontent_id_set = set( - rv._content_relationships().values_list("content__pk", flat=True) + with domain_db(domain) as alias: + for repo in models.Repository.objects.using(alias).filter(pulp_domain=domain): + for rv in models.RepositoryVersion.objects.using(alias).filter(repository=repo): + needs_fix = False + if rv.content_ids is not None: + cached_id_set = set(rv.content_ids) + repositorycontent_id_set = set( + rv._content_relationships().values_list("content__pk", flat=True) + ) + if cached_id_set != repositorycontent_id_set: + if not has_printed_domain: + self.stdout.write(f'In domain "{domain.name}"') + has_printed_domain = True + + self.stdout.write( + f'\tRepository "{repo.name}" (type "{repo.pulp_type}") ' + f"version {rv.number} has a mismatch between the " + "RepositoryContent and the cached ID set" + ) + needs_fix = True + + repositorycontent_id_count = rv._content_relationships().count() + if repositorycontent_id_count == 0: + continue + rv_count_details = models.RepositoryVersionContentDetails.objects.using( + alias + ).filter( + repository_version=rv, + count_type=models.RepositoryVersionContentDetails.PRESENT, ) - if cached_id_set != repositorycontent_id_set: + + # need to sum across all content types + total_count = sum(rvcd.count for rvcd in rv_count_details) + + if total_count != repositorycontent_id_count: + needs_fix = True if not has_printed_domain: self.stdout.write(f'In domain "{domain.name}"') has_printed_domain = True - self.stdout.write( f'\tRepository "{repo.name}" (type "{repo.pulp_type}") ' f"version {rv.number} has a mismatch between the " - "RepositoryContent and the cached ID set" + "RepositoryContent and RepositoryVersionContentDetails" ) - needs_fix = True - repositorycontent_id_count = rv._content_relationships().count() - if repositorycontent_id_count == 0: - continue - rv_count_details = models.RepositoryVersionContentDetails.objects.filter( - repository_version=rv, - count_type=models.RepositoryVersionContentDetails.PRESENT, - ) - - # need to sum across all content types - total_count = sum(rvcd.count for rvcd in rv_count_details) - - if total_count != repositorycontent_id_count: - needs_fix = True - if not has_printed_domain: - self.stdout.write(f'In domain "{domain.name}"') - has_printed_domain = True - self.stdout.write( - f'\tRepository "{repo.name}" (type "{repo.pulp_type}") ' - f"version {rv.number} has a mismatch between the " - "RepositoryContent and RepositoryVersionContentDetails" - ) - - if needs_fix: - number_broken += 1 + if needs_fix: + number_broken += 1 - if not dry_run: - rv.content_ids = list( - rv._content_relationships().values_list("content__pk", flat=True) - ) - rv.save() - rv._compute_counts() + if not dry_run: + rv.content_ids = list( + rv._content_relationships().values_list( + "content__pk", flat=True + ) + ) + rv.save() + rv._compute_counts() self.stdout.write() @@ -130,64 +136,73 @@ def repair_2327(self, options): | Q(client_key__isnull=False) ) - number_unencrypted = 0 - number_multi_encrypted = 0 - - for remote_pk in models.Remote.objects.filter(possibly_affected_remotes).values_list( - "pk", flat=True - ): - try: - remote = models.Remote.objects.get(pk=remote_pk) - # if we can get the remote successfully, it is either OK or the fields are - # encrypted more than once - except cryptography.fernet.InvalidToken: - # If decryption fails then it probably hasn't been encrypted yet - # get the raw column value, avoiding any Django field handling - with connection.cursor() as cursor: - cursor.execute( - "SELECT username, password, proxy_username, proxy_password, client_key " - "FROM core_remote WHERE pulp_id = %s", - [str(remote_pk)], - ) - row = cursor.fetchone() - - field_values = {} + counts = {"number_unencrypted": 0, "number_multi_encrypted": 0} + + def _repair_2327_for_domain(domain, alias): + for remote_pk in ( + models.Remote.objects.using(alias) + .filter(possibly_affected_remotes) + .values_list("pk", flat=True) + ): + try: + remote = models.Remote.objects.using(alias).get(pk=remote_pk) + # if we can get the remote successfully, it is either OK or the fields are + # encrypted more than once + except cryptography.fernet.InvalidToken: + # If decryption fails then it probably hasn't been encrypted yet + # get the raw column value, avoiding any Django field handling + with connections[alias].cursor() as cursor: + cursor.execute( + "SELECT username, password, proxy_username, proxy_password, " + "client_key FROM core_remote WHERE pulp_id = %s", + [str(remote_pk)], + ) + row = cursor.fetchone() - for field, value in zip(fields, row): - field_values[field] = value + field_values = {} - if not dry_run: - models.Remote.objects.filter(pk=remote_pk).update(**field_values) - number_unencrypted += 1 - else: - times_decrypted = 0 - keep_trying = True - needs_update = False - - while keep_trying: - for field in fields: - field_value = getattr(remote, field) # value gets decrypted once on access - if not field_value: - continue + for field, value in zip(fields, row): + field_values[field] = value - try: - # try to decrypt it again - field_value = force_str(fernet.decrypt(force_bytes(field_value))) - # it was decrypted successfully again time, so it was probably - # encrypted multiple times over. lets re-set the value with the - # newly decrypted value - setattr(remote, field, field_value) - needs_update = True - except cryptography.fernet.InvalidToken: - # couldn't be decrypted again, stop here - keep_trying = False - - times_decrypted += 1 - - if needs_update: if not dry_run: - remote.save() - number_multi_encrypted += 1 + models.Remote.objects.using(alias).filter(pk=remote_pk).update( + **field_values + ) + counts["number_unencrypted"] += 1 + else: + times_decrypted = 0 + keep_trying = True + needs_update = False + + while keep_trying: + for field in fields: + # value gets decrypted once on access + field_value = getattr(remote, field) + if not field_value: + continue + + try: + # try to decrypt it again + field_value = force_str(fernet.decrypt(force_bytes(field_value))) + # it was decrypted successfully again time, so it was probably + # encrypted multiple times over. lets re-set the value with the + # newly decrypted value + setattr(remote, field, field_value) + needs_update = True + except cryptography.fernet.InvalidToken: + # couldn't be decrypted again, stop here + keep_trying = False + + times_decrypted += 1 + + if needs_update: + if not dry_run: + remote.save() + counts["number_multi_encrypted"] += 1 + + for_each_domain(_repair_2327_for_domain) + number_unencrypted = counts["number_unencrypted"] + number_multi_encrypted = counts["number_multi_encrypted"] if dry_run: print("Remotes with un-encrypted fields: {}".format(number_unencrypted)) @@ -210,22 +225,25 @@ def repair_7465(self, options): for domain in models.Domain.objects.all(): has_printed_domain = False - for repo in models.Repository.objects.filter(pulp_domain=domain): - for rv in models.RepositoryVersion.objects.filter(repository=repo): - if rv.content_ids is None: - if not has_printed_domain: - self.stdout.write(f'In domain "{domain.name}"') - has_printed_domain = True - number_missing += 1 - self.stdout.write( - f'\tRepository "{repo.name}" (type "{repo.pulp_type}") ' - f"version {rv.number} has a missing content_ids cache" - ) - if not dry_run: - rv.content_ids = list( - rv._content_relationships().values_list("content__pk", flat=True) + with domain_db(domain) as alias: + for repo in models.Repository.objects.using(alias).filter(pulp_domain=domain): + for rv in models.RepositoryVersion.objects.using(alias).filter(repository=repo): + if rv.content_ids is None: + if not has_printed_domain: + self.stdout.write(f'In domain "{domain.name}"') + has_printed_domain = True + number_missing += 1 + self.stdout.write( + f'\tRepository "{repo.name}" (type "{repo.pulp_type}") ' + f"version {rv.number} has a missing content_ids cache" ) - rv.save() + if not dry_run: + rv.content_ids = list( + rv._content_relationships().values_list( + "content__pk", flat=True + ) + ) + rv.save() if not number_missing: self.stdout.write("Finished. (OK)") diff --git a/pulpcore/app/management/commands/dump-publications-to-fs.py b/pulpcore/app/management/commands/dump-publications-to-fs.py index 299b0c085b3..5146947e839 100644 --- a/pulpcore/app/management/commands/dump-publications-to-fs.py +++ b/pulpcore/app/management/commands/dump-publications-to-fs.py @@ -5,12 +5,13 @@ from django.core.exceptions import ObjectDoesNotExist from django.core.management import BaseCommand, CommandError -from pulpcore.app.models import Distribution, Publication +from pulpcore.app.models import Distribution, Domain, Publication from pulpcore.app.tasks.export import ( UnexportableArtifactException, _export_location_is_clean, _export_publication_to_file_system, ) +from pulpcore.app.util import for_each_domain from pulpcore.app.viewsets.base import NamedModelViewSet from pulpcore.constants import FS_EXPORT_METHODS @@ -23,6 +24,12 @@ class Command(BaseCommand): def add_arguments(self, parser): """Set up arguments.""" parser.add_argument("--publication", required=False, help=_("A publication ID.")) + parser.add_argument( + "--domain", + default="default", + required=False, + help=_("The pulp domain --publication belongs to (ignored otherwise)."), + ) parser.add_argument( "--distribution-path-prefix", required=False, @@ -75,45 +82,53 @@ def handle(self, *args, **options): publication_pk = NamedModelViewSet.extract_pk(options["publication"]) except Exception: publication_pk = options["publication"] - publication = Publication.objects.get(pk=publication_pk) + try: + domain = Domain.objects.get(name=options["domain"]) + except Domain.DoesNotExist: + raise CommandError( + _("Domain '{name}' does not exist.").format(name=options["domain"]) + ) + publication = Publication.objects.using(domain.database_alias).get(pk=publication_pk) to_export.append((options["dest"], publication)) # If no publication was specified go through the distributions and dump them if they # meet the criteria else: - # If a base_path prefix was provided, filter out distributions with a base path - # that doesn't start with the prefix - if options.get("distribution_path_prefix"): - distributions = Distribution.objects.filter( - base_path__startswith=options["distribution_path_prefix"] - ) - else: - distributions = Distribution.objects.all() - - # Filter out distributions that don't match the type specified (if any) - if options["type"]: - distributions = distributions.filter(pulp_type__startswith=options["type"]) - - # For all matching distributions, if they have a publication, dump it in a directory - # matching the original distribution structure - for distribution in distributions: - if distribution.publication: - publication = distribution.publication - elif distribution.repository: - repository = distribution.repository - # Account for distributions serving the latest publication of a given repository - try: - publication = Publication.objects.filter( - repository_version__in=repository.versions.all(), complete=True - ).latest("repository_version", "pulp_created") - repo_path = os.path.join(options["dest"], distribution.base_path) - to_export.append((repo_path, publication)) - except ObjectDoesNotExist: - logging.warning( - "No publication found for the repo published at '{}': skipping".format( - distribution.base_path + + def _collect_for_domain(domain, alias): + if options.get("distribution_path_prefix"): + distributions = Distribution.objects.using(alias).filter( + base_path__startswith=options["distribution_path_prefix"] + ) + else: + distributions = Distribution.objects.using(alias).all() + + if options["type"]: + distributions = distributions.filter(pulp_type__startswith=options["type"]) + + for distribution in distributions: + if distribution.publication: + publication = distribution.publication + elif distribution.repository: + repository = distribution.repository + try: + publication = ( + Publication.objects.using(alias) + .filter( + repository_version__in=repository.versions.all(), + complete=True, + ) + .latest("repository_version", "pulp_created") ) - ) + repo_path = os.path.join(options["dest"], distribution.base_path) + to_export.append((repo_path, publication)) + except ObjectDoesNotExist: + logging.warning( + "No publication found for the repo published at '{}' in " + "domain '{}': skipping".format(distribution.base_path, domain.name) + ) + + for_each_domain(_collect_for_domain) # Go through all the target directories first, if any of them are dirty, print warnings # and exit - unless the user explicitly asked to go through with it anyway. diff --git a/pulpcore/app/management/commands/handle-artifact-checksums.py b/pulpcore/app/management/commands/handle-artifact-checksums.py index 11c3ba72233..d8699561ea0 100644 --- a/pulpcore/app/management/commands/handle-artifact-checksums.py +++ b/pulpcore/app/management/commands/handle-artifact-checksums.py @@ -8,6 +8,7 @@ from pulpcore import constants from pulpcore.app import pulp_hashlib +from pulpcore.app.util import for_each_domain from pulpcore.plugin.models import ( Artifact, Content, @@ -51,19 +52,27 @@ def _print_out_repository_version_hrefs(self, repo_versions): ) ) - def _show_on_demand_content(self, checksums): + def _show_on_demand_content(self, checksums, alias): query = Q(pk__in=[]) for checksum in checksums: query |= Q(**{f"{checksum}__isnull": False}) - remote_artifacts = RemoteArtifact.objects.filter(query).filter( - content_artifact__artifact__isnull=True + remote_artifacts = ( + RemoteArtifact.objects.using(alias) + .filter(query) + .filter(content_artifact__artifact__isnull=True) ) ras_size = remote_artifacts.aggregate(Sum("size"))["size__sum"] - content_artifacts = ContentArtifact.objects.filter(remoteartifact__pk__in=remote_artifacts) - content = Content.objects.filter(contentartifact__pk__in=content_artifacts) - repo_versions = RepositoryVersion.objects.with_content(content).select_related("repository") + content_artifacts = ContentArtifact.objects.using(alias).filter( + remoteartifact__pk__in=remote_artifacts + ) + content = Content.objects.using(alias).filter(contentartifact__pk__in=content_artifacts) + repo_versions = ( + RepositoryVersion.objects.using(alias) + .with_content(content) + .select_related("repository") + ) self.stdout.write( "Found {} on-demand content units with forbidden checksums.".format(content.count()) @@ -77,7 +86,7 @@ def _show_on_demand_content(self, checksums): self.stdout.write(_("\nAffected repository versions with remote content:")) self._print_out_repository_version_hrefs(repo_versions) - def _show_immediate_content(self, forbidden_checksums): + def _show_immediate_content(self, forbidden_checksums, alias): allowed_checksums = set( constants.ALL_KNOWN_CONTENT_CHECKSUMS.symmetric_difference(forbidden_checksums) ) @@ -89,10 +98,14 @@ def _show_immediate_content(self, forbidden_checksums): for allowed_checksum in allowed_checksums: query_required |= Q(**{f"{allowed_checksum}__isnull": True}) - artifacts = Artifact.objects.filter(query_forbidden | query_required) - content_artifacts = ContentArtifact.objects.filter(artifact__in=artifacts) - content = Content.objects.filter(contentartifact__pk__in=content_artifacts) - repo_versions = RepositoryVersion.objects.with_content(content).select_related("repository") + artifacts = Artifact.objects.using(alias).filter(query_forbidden | query_required) + content_artifacts = ContentArtifact.objects.using(alias).filter(artifact__in=artifacts) + content = Content.objects.using(alias).filter(contentartifact__pk__in=content_artifacts) + repo_versions = ( + RepositoryVersion.objects.using(alias) + .with_content(content) + .select_related("repository") + ) self.stdout.write( "Found {} downloaded content units with forbidden or missing checksums.".format( @@ -110,11 +123,11 @@ def _show_immediate_content(self, forbidden_checksums): self.stdout.write(_("\nAffected repository versions with present content:")) self._print_out_repository_version_hrefs(repo_versions) - def _download_artifact(self, artifact, checksum, file_path): + def _download_artifact(self, artifact, checksum, file_path, alias): restored = False - for ca in artifact.content_memberships.all(): + for ca in artifact.content_memberships.using(alias).all(): if not restored: - for ra in ca.remoteartifact_set.all(): + for ra in ca.remoteartifact_set.using(alias).all(): remote = ra.remote.cast() if remote.policy == "immediate": self.stdout.write(_("Restoring missing file {}").format(file_path)) @@ -153,8 +166,12 @@ def _report(self, allowed_checksums): allowed_checksums ) - self._show_on_demand_content(forbidden_checksums) - self._show_immediate_content(forbidden_checksums) + def _report_for_domain(domain, alias): + self.stdout.write(_("\n=== Domain '{name}' ===").format(name=domain.name)) + self._show_on_demand_content(forbidden_checksums, alias) + self._show_immediate_content(forbidden_checksums, alias) + + for_each_domain(_report_for_domain) def handle(self, *args, **options): if options["report"]: @@ -167,30 +184,36 @@ def handle(self, *args, **options): log.setLevel(logging.ERROR) hrefs = set() - for checksum in settings.ALLOWED_CONTENT_CHECKSUMS: - params = {f"{checksum}__isnull": True} - artifacts_qs = Artifact.objects.filter(**params) - artifacts = [] - for a in artifacts_qs.iterator(): - hasher = pulp_hashlib.new(checksum) - try: - with a.file as fp: - for chunk in fp.chunks(CHUNK_SIZE): - hasher.update(chunk) - setattr(a, checksum, hasher.hexdigest()) - except FileNotFoundError: - file_path = os.path.join(settings.MEDIA_ROOT, a.file.name) - restored = self._download_artifact(a, checksum, file_path) - if not restored: - hrefs.add(file_path) - artifacts.append(a) - - if len(artifacts) >= 1000: - Artifact.objects.bulk_update(objs=artifacts, fields=[checksum], batch_size=1000) - artifacts.clear() - - if artifacts: - Artifact.objects.bulk_update(objs=artifacts, fields=[checksum]) + + def _populate_missing_for_domain(domain, alias): + for checksum in settings.ALLOWED_CONTENT_CHECKSUMS: + params = {f"{checksum}__isnull": True} + artifacts_qs = Artifact.objects.using(alias).filter(**params) + artifacts = [] + for a in artifacts_qs.iterator(): + hasher = pulp_hashlib.new(checksum) + try: + with a.file as fp: + for chunk in fp.chunks(CHUNK_SIZE): + hasher.update(chunk) + setattr(a, checksum, hasher.hexdigest()) + except FileNotFoundError: + file_path = os.path.join(settings.MEDIA_ROOT, a.file.name) + restored = self._download_artifact(a, checksum, file_path, alias) + if not restored: + hrefs.add(file_path) + artifacts.append(a) + + if len(artifacts) >= 1000: + Artifact.objects.using(alias).bulk_update( + objs=artifacts, fields=[checksum], batch_size=1000 + ) + artifacts.clear() + + if artifacts: + Artifact.objects.using(alias).bulk_update(objs=artifacts, fields=[checksum]) + + for_each_domain(_populate_missing_for_domain) if hrefs: raise CommandError( @@ -200,12 +223,20 @@ def handle(self, *args, **options): forbidden_checksums = set(constants.ALL_KNOWN_CONTENT_CHECKSUMS).difference( settings.ALLOWED_CONTENT_CHECKSUMS ) - for checksum in forbidden_checksums: - search_params = {f"{checksum}__isnull": False} - update_params = {f"{checksum}": None} - artifacts_qs = Artifact.objects.filter(**search_params) - if artifacts_qs.exists(): - self.stdout.write("Removing forbidden checksum {} from database".format(checksum)) - artifacts_qs.update(**update_params) + + def _remove_forbidden_for_domain(domain, alias): + for checksum in forbidden_checksums: + search_params = {f"{checksum}__isnull": False} + update_params = {f"{checksum}": None} + artifacts_qs = Artifact.objects.using(alias).filter(**search_params) + if artifacts_qs.exists(): + self.stdout.write( + "Removing forbidden checksum {} from database (domain '{}')".format( + checksum, domain.name + ) + ) + artifacts_qs.update(**update_params) + + for_each_domain(_remove_forbidden_for_domain) self.stdout.write(_("Finished aligning checksums with settings.ALLOWED_CONTENT_CHECKSUMS")) diff --git a/pulpcore/app/management/commands/remove-plugin.py b/pulpcore/app/management/commands/remove-plugin.py index 7531670b03e..f7d45367b9c 100644 --- a/pulpcore/app/management/commands/remove-plugin.py +++ b/pulpcore/app/management/commands/remove-plugin.py @@ -5,7 +5,7 @@ from django.conf import settings from django.contrib.contenttypes.models import ContentType from django.core.management import BaseCommand, CommandError, call_command -from django.db import IntegrityError, connection +from django.db import IntegrityError, connections from django.db.migrations.exceptions import IrreversibleError from django.db.models.signals import post_migrate @@ -92,16 +92,21 @@ def _remove_plugin_data(self, app_label): In some cases, the order in which models are removed matters, e.g. FK is a part of uniqueness constraint. Try to remove such problematic models later. """ + for alias in settings.DATABASES: + self._remove_plugin_data_from_alias(app_label, alias) + self._remove_indirect_plugin_data(app_label) + def _remove_plugin_data_from_alias(self, app_label, alias): + self.stdout.write(_("Removing {} plugin data from alias '{}'...").format(app_label, alias)) models_to_delete = set(apps.all_models[app_label].values()) prev_model_count = len(models_to_delete) + 1 while models_to_delete and len(models_to_delete) < prev_model_count: # while there is something to delete and something is being deleted on each iteration removed_models = set() for model in models_to_delete: - self.stdout.write(_("Removing model: {}").format(model)) + self.stdout.write(_("Removing model: {} (alias '{}')").format(model, alias)) try: - model.objects.filter().delete() + model.objects.using(alias).filter().delete() except IntegrityError: continue else: @@ -114,18 +119,16 @@ def _remove_plugin_data(self, app_label): # Never-happen case raise CommandError( ( - "Data for the following models can't be removed: {}. Please contact plugin " - "maintainers." - ).format(list(models_to_delete)) + "Data for the following models can't be removed on alias '{}': {}. Please " + "contact plugin maintainers." + ).format(alias, list(models_to_delete)) ) - self._remove_indirect_plugin_data(app_label) - - def _drop_plugin_tables(self, app_label): + def _drop_plugin_tables(self, app_label, alias): """ Drop plugin table with raw SQL. """ - with connection.cursor() as cursor: + with connections[alias].cursor() as cursor: cursor.execute(DROP_PLUGIN_TABLES_QUERY.format(app_label=app_label)) def _unapply_migrations(self, app_label): @@ -148,13 +151,20 @@ def _unapply_migrations(self, app_label): if app_config.label == "core": post_migrate.disconnect(sender=app_config, dispatch_uid="delete_anon_identifier") - try: - call_command("migrate", app_label=app_label, migration_name="zero") - except (IrreversibleError, Exception): - # a plugin has irreversible migrations or some other problem, drop the tables and fake - # that migrations are unapplied. - self._drop_plugin_tables(app_label) - call_command("migrate", app_label=app_label, migration_name="zero", fake=True) + for alias in settings.DATABASES: + try: + call_command("migrate", app_label=app_label, migration_name="zero", database=alias) + except (IrreversibleError, Exception): + # a plugin has irreversible migrations or some other problem, drop the tables and + # fake that migrations are unapplied. + self._drop_plugin_tables(app_label, alias) + call_command( + "migrate", + app_label=app_label, + migration_name="zero", + fake=True, + database=alias, + ) def handle(self, *args, **options): plugin_name = options["plugin_name"] diff --git a/pulpcore/app/management/commands/repository-size.py b/pulpcore/app/management/commands/repository-size.py index a8ab13f7705..9bb12fc0df3 100644 --- a/pulpcore/app/management/commands/repository-size.py +++ b/pulpcore/app/management/commands/repository-size.py @@ -7,8 +7,8 @@ from django.conf import settings from django.core.management import BaseCommand, CommandError -from pulpcore.app.models import Repository -from pulpcore.app.util import extract_pk, get_url +from pulpcore.app.models import Domain, Repository +from pulpcore.app.util import extract_pk, for_each_domain, get_url def gather_repository_sizes(repositories, include_versions=False, include_on_demand=False): @@ -106,22 +106,49 @@ def add_arguments(self, parser): def handle(self, *args, **options): """Implement the command.""" - domain = options.get("domain") + domain_name = options.get("domain") repository_hrefs = options.get("repositories") - if domain and repository_hrefs: + if domain_name and repository_hrefs: raise CommandError(_("--domain and --repositories are mutually exclusive")) - repositories = Repository.objects.all() + report = [] if repository_hrefs: repos_ids = [extract_pk(r) for r in repository_hrefs] - repositories = repositories.filter(pk__in=repos_ids) - elif domain: - repositories = repositories.filter(pulp_domain__name=domain) - - report = gather_repository_sizes( - repositories, - include_versions=options["include_versions"], - include_on_demand=options["include_on_demand"], - ) + for alias in settings.DATABASES: + repositories = Repository.objects.using(alias).filter(pk__in=repos_ids) + report.extend( + gather_repository_sizes( + repositories, + include_versions=options["include_versions"], + include_on_demand=options["include_on_demand"], + ) + ) + elif domain_name: + try: + domain = Domain.objects.get(name=domain_name) + except Domain.DoesNotExist: + raise CommandError(_("Domain '{name}' does not exist.").format(name=domain_name)) + repositories = Repository.objects.using(domain.database_alias).filter( + pulp_domain=domain + ) + report = gather_repository_sizes( + repositories, + include_versions=options["include_versions"], + include_on_demand=options["include_on_demand"], + ) + else: + + def _gather_for_domain(domain, alias): + repositories = Repository.objects.using(alias).filter(pulp_domain=domain) + report.extend( + gather_repository_sizes( + repositories, + include_versions=options["include_versions"], + include_on_demand=options["include_on_demand"], + ) + ) + + for_each_domain(_gather_for_domain) + json.dump(report, sys.stdout, indent=4) print() diff --git a/pulpcore/app/management/commands/rotate-db-key.py b/pulpcore/app/management/commands/rotate-db-key.py index 6bbec206e10..086206586c9 100644 --- a/pulpcore/app/management/commands/rotate-db-key.py +++ b/pulpcore/app/management/commands/rotate-db-key.py @@ -2,8 +2,9 @@ from gettext import gettext as _ from django.apps import apps +from django.conf import settings from django.core.management import BaseCommand -from django.db import connection, transaction +from django.db import connections, transaction from pulpcore.app.models import MasterModel from pulpcore.app.models.fields import EncryptedJSONField, EncryptedTextField @@ -40,6 +41,11 @@ def add_arguments(self, parser): def handle(self, *args, **options): dry_run = options["dry_run"] + for alias in settings.DATABASES: + self._rotate_alias(alias, dry_run) + + def _rotate_alias(self, alias, dry_run): + print(_("Rotating encrypted fields on database alias '{alias}'.").format(alias=alias)) for model in apps.get_models(): if issubclass(model, MasterModel) and model._meta.master_model is None: # This is a master model, and we will handle all it's descendents. @@ -51,23 +57,23 @@ def handle(self, *args, **options): ] if field_names: print( - _("Updating {fields} on {model}.").format( - model=model.__name__, fields=",".join(field_names) + _("Updating {fields} on {model} (alias '{alias}').").format( + model=model.__name__, fields=",".join(field_names), alias=alias ) ) exclude_filters = {f"{field_name}": None for field_name in field_names} - qs = model.objects.exclude(**exclude_filters).only(*field_names) - with suppress(DryRun), transaction.atomic(): + qs = model.objects.using(alias).exclude(**exclude_filters).only(*field_names) + with suppress(DryRun), transaction.atomic(using=alias): batch = [] for item in qs.iterator(): batch.append(item) if len(batch) >= 1024: - model.objects.bulk_update(batch, field_names) + model.objects.using(alias).bulk_update(batch, field_names) batch = [] if batch: - model.objects.bulk_update(batch, field_names) + model.objects.using(alias).bulk_update(batch, field_names) batch = [] if dry_run: - with connection.cursor() as cursor: + with connections[alias].cursor() as cursor: cursor.execute("SET CONSTRAINTS ALL IMMEDIATE") raise DryRun() diff --git a/pulpcore/app/migrations/0101_add_domain.py b/pulpcore/app/migrations/0101_add_domain.py index ac5363050ca..71cd734c33e 100644 --- a/pulpcore/app/migrations/0101_add_domain.py +++ b/pulpcore/app/migrations/0101_add_domain.py @@ -7,7 +7,6 @@ import pulpcore.app.models.fields import uuid - DEFAULT_DELETE_TRIGGER = """ CREATE OR REPLACE FUNCTION protect_default() RETURNS TRIGGER as $protect_default$ BEGIN @@ -28,42 +27,52 @@ def create_default_domain(apps, schema_editor): - Domain = apps.get_model('core', 'Domain') + if schema_editor.connection.alias != "default": + return + Domain = apps.get_model("core", "Domain") try: - default_domain = Domain.objects.get(name="default") + default_domain = Domain.objects.using("default").get(name="default") except Domain.DoesNotExist: default_domain = Domain( name="default", storage_class=settings.STORAGES["default"]["BACKEND"] ) - default_domain.save(skip_hooks=True) + default_domain.save(using="default", skip_hooks=True) class Migration(migrations.Migration): - dependencies = [ - ('contenttypes', '0002_remove_content_type_name'), + ("contenttypes", "0002_remove_content_type_name"), migrations.swappable_dependency(settings.AUTH_USER_MODEL), - ('core', '0100_upstreampulp'), + ("core", "0100_upstreampulp"), ] operations = [ migrations.CreateModel( - name='Domain', + name="Domain", fields=[ - ('pulp_id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)), - ('pulp_created', models.DateTimeField(auto_now_add=True)), - ('pulp_last_updated', models.DateTimeField(auto_now=True, null=True)), - ('name', models.SlugField(unique=True)), - ('description', models.TextField(null=True)), - ('storage_class', models.TextField()), - ('storage_settings', pulpcore.app.models.fields.EncryptedJSONField(default=dict)), - ('redirect_to_object_storage', models.BooleanField(default=True)), - ('hide_guarded_distributions', models.BooleanField(default=False)), + ( + "pulp_id", + models.UUIDField( + default=uuid.uuid4, editable=False, primary_key=True, serialize=False + ), + ), + ("pulp_created", models.DateTimeField(auto_now_add=True)), + ("pulp_last_updated", models.DateTimeField(auto_now=True, null=True)), + ("name", models.SlugField(unique=True)), + ("description", models.TextField(null=True)), + ("storage_class", models.TextField()), + ("storage_settings", pulpcore.app.models.fields.EncryptedJSONField(default=dict)), + ("redirect_to_object_storage", models.BooleanField(default=True)), + ("hide_guarded_distributions", models.BooleanField(default=False)), ], options={ - 'permissions': [('manage_roles_domain', 'Can manage role assignments on domain')], + "permissions": [("manage_roles_domain", "Can manage role assignments on domain")], }, - bases=(django_lifecycle.mixins.LifecycleModelMixin, models.Model, pulpcore.app.models.access_policy.AutoAddObjPermsMixin), + bases=( + django_lifecycle.mixins.LifecycleModelMixin, + models.Model, + pulpcore.app.models.access_policy.AutoAddObjPermsMixin, + ), ), migrations.RunSQL(DEFAULT_DELETE_TRIGGER, reverse_sql=REMOVE_DEFAULT_DELETE_TRIGGER), migrations.RunPython(code=create_default_domain, reverse_code=migrations.RunPython.noop), diff --git a/pulpcore/app/migrations/0159_domain_database_alias_domain_moving.py b/pulpcore/app/migrations/0159_domain_database_alias_domain_moving.py new file mode 100644 index 00000000000..8ec4ae209b7 --- /dev/null +++ b/pulpcore/app/migrations/0159_domain_database_alias_domain_moving.py @@ -0,0 +1,28 @@ +# Generated by Django 5.2.13 on 2026-07-09 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0158_domain_default_content_guard"), + ] + + operations = [ + migrations.AddField( + model_name="domain", + name="database_alias", + field=models.SlugField( + default="default", + help_text="DATABASES alias where this domain's data-plane objects reside.", + ), + ), + migrations.AddField( + model_name="domain", + name="moving", + field=models.BooleanField( + default=False, + help_text="True while this domain's data is being moved between database aliases.", + ), + ), + ] diff --git a/pulpcore/app/migrations/0160_createdresource_content_object_domain_and_more.py b/pulpcore/app/migrations/0160_createdresource_content_object_domain_and_more.py new file mode 100644 index 00000000000..9ae97b2621a --- /dev/null +++ b/pulpcore/app/migrations/0160_createdresource_content_object_domain_and_more.py @@ -0,0 +1,53 @@ +# Generated by Django 5.2.15 on 2026-07-09 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0159_domain_database_alias_domain_moving"), + ] + + operations = [ + migrations.AddField( + model_name="createdresource", + name="content_object_domain", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="+", + to="core.domain", + ), + ), + migrations.AddField( + model_name="exportedresource", + name="content_object_domain", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="+", + to="core.domain", + ), + ), + migrations.AddField( + model_name="grouprole", + name="content_object_domain", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="+", + to="core.domain", + ), + ), + migrations.AddField( + model_name="userrole", + name="content_object_domain", + field=models.ForeignKey( + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="+", + to="core.domain", + ), + ), + ] diff --git a/pulpcore/app/models/content.py b/pulpcore/app/models/content.py index e62c3f85a6e..465f692b272 100644 --- a/pulpcore/app/models/content.py +++ b/pulpcore/app/models/content.py @@ -27,6 +27,7 @@ from pulpcore.app import pulp_hashlib from pulpcore.app.models import BaseModel, MasterModel, fields, storage from pulpcore.app.models.fields import RelativePathField +from pulpcore.app.queryset import CrossDBQuerySetMixin from pulpcore.app.util import get_domain_pk, gpg_verify from pulpcore.constants import ALL_KNOWN_CONTENT_CHECKSUMS from pulpcore.exceptions import ( @@ -98,7 +99,7 @@ def bulk_get_or_create(self, objs, batch_size=None): return objs -class BulkTouchQuerySet(models.QuerySet): +class BulkTouchQuerySet(CrossDBQuerySetMixin, models.QuerySet): """ A query set that provides `touch()`. """ @@ -696,7 +697,7 @@ def sort_key(ca): return c_key, a_key -class RemoteArtifactQuerySet(models.QuerySet): +class RemoteArtifactQuerySet(CrossDBQuerySetMixin, models.QuerySet): """QuerySet that provides methods for querying RemoteArtifact.""" def acs(self): diff --git a/pulpcore/app/models/domain.py b/pulpcore/app/models/domain.py index e8e22eb76d1..d3240d25ccc 100644 --- a/pulpcore/app/models/domain.py +++ b/pulpcore/app/models/domain.py @@ -1,7 +1,9 @@ +from django.conf import settings from django.contrib.postgres.fields import HStoreField +from django.core.exceptions import ValidationError from django.core.files.storage import default_storage from django.db import models -from django_lifecycle import BEFORE_DELETE, BEFORE_UPDATE, hook +from django_lifecycle import BEFORE_CREATE, BEFORE_DELETE, BEFORE_UPDATE, hook from pulpcore.app.models import AutoAddObjPermsMixin, BaseModel from pulpcore.exceptions import DomainProtectedError @@ -51,6 +53,14 @@ class Domain(BaseModel, AutoAddObjPermsMixin): default_content_guard = models.ForeignKey( "ContentGuard", null=True, on_delete=models.SET_NULL, related_name="+" ) + database_alias = models.SlugField( + default="default", + help_text="DATABASES alias where this domain's data-plane objects reside.", + ) + moving = models.BooleanField( + default=False, + help_text="True while this domain's data is being moved between database aliases.", + ) def get_storage(self): """Returns this domain's instantiated storage class.""" @@ -73,13 +83,29 @@ def get_storage(self): def prevent_default_deletion(self): raise models.ProtectedError("Default domain can not be updated/deleted.", [self]) + @hook(BEFORE_CREATE) + @hook(BEFORE_UPDATE, when="database_alias", has_changed=True) + def _validate_database_alias(self): + if self.database_alias not in settings.DATABASES: + raise ValidationError( + { + "database_alias": ( + f"'{self.database_alias}' is not a configured DATABASES alias." + ) + } + ) + @hook(BEFORE_DELETE, when="name", is_not="default") def _cleanup_orphans_pre_delete(self): - protected_content_set = self.content_set.exclude(version_memberships__isnull=True) + protected_content_set = self.content_set.using(self.database_alias).exclude( + version_memberships__isnull=True + ) if protected_content_set.exists(): raise DomainProtectedError() - self.content_set.filter(version_memberships__isnull=True).delete() - for artifact in self.artifact_set.all().iterator(): + self.content_set.using(self.database_alias).filter( + version_memberships__isnull=True + ).delete() + for artifact in self.artifact_set.using(self.database_alias).all().iterator(): # Delete on by one to properly cleanup the storage. artifact.delete() diff --git a/pulpcore/app/models/generic.py b/pulpcore/app/models/generic.py index 2d8420d4a01..8503ed64d04 100644 --- a/pulpcore/app/models/generic.py +++ b/pulpcore/app/models/generic.py @@ -5,14 +5,110 @@ https://docs.djangoproject.com/en/3.2/ref/contrib/contenttypes/#generic-relations """ +import logging + from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.models import ContentType +from django.core.exceptions import ObjectDoesNotExist from django.db import models from pulpcore.app.models.base import BaseModel +_logger = logging.getLogger(__name__) + + +_UNSET = object() + +_DOMAIN_WALK_MAX_DEPTH = 2 + + +def _resolve_domain_id(value, _depth=0, _seen=None): + domain_id = getattr(value, "pulp_domain_id", None) + if domain_id is not None: + return domain_id + if _depth >= _DOMAIN_WALK_MAX_DEPTH: + return None + if _seen is None: + _seen = set() + if value.pk is not None: + key = (type(value), value.pk) + if key in _seen: + return None + _seen.add(key) + for field in value._meta.get_fields(): + if not (field.many_to_one or field.one_to_one) or not getattr(field, "concrete", False): + continue + try: + related = getattr(value, field.name) + except ObjectDoesNotExist: + continue + if related is None or not hasattr(related, "_meta"): + continue + resolved = _resolve_domain_id(related, _depth + 1, _seen) + if resolved is not None: + return resolved + return None + + +class DomainResolvedGenericRelation: + def __init__(self, *args, **kwargs): + has_content_object = "content_object" in kwargs + content_object = kwargs.pop("content_object", None) + super().__init__(*args, **kwargs) + if has_content_object: + self.content_object = content_object + + @property + def content_object(self): + cached = self.__dict__.get("_content_object_cache", _UNSET) + if cached is not _UNSET: + return cached + if self.content_type_id is None or self.object_id is None: + return None + model_class = self.content_type.model_class() + if self.content_object_domain_id is not None: + alias = self.content_object_domain.database_alias + try: + resolved = model_class.objects.using(alias).get(pk=self.object_id) + except model_class.DoesNotExist: + _logger.warning( + "content_object for %s (pk=%s) not found on alias '%s' " + "(content_type_id=%s, object_id=%s). The referenced object may have been " + "deleted, or this row's domain may not be replicated to that alias.", + self._meta.label, + self.pk, + alias, + self.content_type_id, + self.object_id, + ) + resolved = None + else: + try: + resolved = model_class._base_manager.using(self._state.db or "default").get( + pk=self.object_id + ) + except model_class.DoesNotExist: + resolved = None + self.__dict__["_content_object_cache"] = resolved + return resolved + + @content_object.setter + def content_object(self, value): + self.__dict__["_content_object_cache"] = value + if value is None: + self.content_type = None + self.object_id = None + self.content_object_domain_id = None + return + gfk = type(self)._content_object + self.content_type = ContentType.objects.db_manager("default").get_for_model( + value, for_concrete_model=gfk.for_concrete_model + ) + self.object_id = value.pk + self.content_object_domain_id = _resolve_domain_id(value) + -class GenericRelationModel(BaseModel): +class GenericRelationModel(DomainResolvedGenericRelation, BaseModel): """Base model class for implementing Generic Relations. This class provides the required fields to implement generic relations. Instances of @@ -22,7 +118,10 @@ class GenericRelationModel(BaseModel): content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE) object_id = models.UUIDField() - content_object = GenericForeignKey("content_type", "object_id", for_concrete_model=False) + _content_object = GenericForeignKey("content_type", "object_id", for_concrete_model=False) + content_object_domain = models.ForeignKey( + "core.Domain", null=True, on_delete=models.SET_NULL, related_name="+" + ) class Meta: abstract = True diff --git a/pulpcore/app/models/publication.py b/pulpcore/app/models/publication.py index 5b62b6014f9..72fba3ce735 100644 --- a/pulpcore/app/models/publication.py +++ b/pulpcore/app/models/publication.py @@ -23,6 +23,7 @@ from pulpcore.app.files import PulpTemporaryUploadedFile from pulpcore.app.models import AutoAddObjPermsMixin from pulpcore.app.models.fields import RelativePathField +from pulpcore.app.queryset import CrossDBQuerySetMixin from pulpcore.app.util import cache_key, get_domain_pk, get_url, retain_distributed_pub_enabled from pulpcore.cache import Cache from pulpcore.responses import ArtifactResponse @@ -35,7 +36,7 @@ _logger = logging.getLogger(__name__) -class PublicationQuerySet(models.QuerySet): +class PublicationQuerySet(CrossDBQuerySetMixin, models.QuerySet): """A queryset that provides publication filtering methods.""" def with_content(self, content): diff --git a/pulpcore/app/models/repository.py b/pulpcore/app/models/repository.py index f18ca998f73..20fa6252490 100644 --- a/pulpcore/app/models/repository.py +++ b/pulpcore/app/models/repository.py @@ -18,6 +18,7 @@ from django_lifecycle import AFTER_UPDATE, BEFORE_CREATE, BEFORE_DELETE, hook from rest_framework.exceptions import APIException +from pulpcore.app.queryset import CrossDBQuerySetMixin from pulpcore.app.util import ( batch_qs, cache_key, @@ -882,7 +883,7 @@ class Meta: ) -class RepositoryVersionQuerySet(models.QuerySet): +class RepositoryVersionQuerySet(CrossDBQuerySetMixin, models.QuerySet): """A queryset that provides repository version filtering methods.""" def complete(self): diff --git a/pulpcore/app/models/role.py b/pulpcore/app/models/role.py index c7db1d3fa24..7835b531e0b 100644 --- a/pulpcore/app/models/role.py +++ b/pulpcore/app/models/role.py @@ -5,6 +5,7 @@ from django.db import models from pulpcore.app.models import BaseModel, Group +from pulpcore.app.models.generic import DomainResolvedGenericRelation class Role(BaseModel): @@ -27,7 +28,7 @@ class Role(BaseModel): permissions = models.ManyToManyField(Permission) -class UserRole(BaseModel): +class UserRole(DomainResolvedGenericRelation, BaseModel): """ Join table for user to role associations with optional content object. @@ -46,7 +47,10 @@ class UserRole(BaseModel): role = models.ForeignKey(Role, related_name="object_users", on_delete=models.CASCADE) content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True) object_id = models.TextField(null=True) - content_object = GenericForeignKey("content_type", "object_id", for_concrete_model=False) + _content_object = GenericForeignKey("content_type", "object_id", for_concrete_model=False) + content_object_domain = models.ForeignKey( + "Domain", null=True, on_delete=models.SET_NULL, related_name="+" + ) domain = models.ForeignKey("Domain", null=True, on_delete=models.CASCADE) class Meta: @@ -57,7 +61,7 @@ class Meta: ] -class GroupRole(BaseModel): +class GroupRole(DomainResolvedGenericRelation, BaseModel): """ Join table for group to role associations with optional content object. @@ -74,7 +78,10 @@ class GroupRole(BaseModel): role = models.ForeignKey(Role, related_name="object_groups", on_delete=models.CASCADE) content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, null=True) object_id = models.TextField(null=True) - content_object = GenericForeignKey("content_type", "object_id", for_concrete_model=False) + _content_object = GenericForeignKey("content_type", "object_id", for_concrete_model=False) + content_object_domain = models.ForeignKey( + "Domain", null=True, on_delete=models.SET_NULL, related_name="+" + ) domain = models.ForeignKey("Domain", null=True, on_delete=models.CASCADE) class Meta: diff --git a/pulpcore/app/queryset.py b/pulpcore/app/queryset.py new file mode 100644 index 00000000000..3c066bec370 --- /dev/null +++ b/pulpcore/app/queryset.py @@ -0,0 +1,37 @@ +from django.db import models +from django.db.models import Q + + +class CrossDBQuerySetMixin: + def filter(self, *args, **kwargs): + from pulpcore.app.db_router import is_multi_db_routing_active + + if not is_multi_db_routing_active(): + return super().filter(*args, **kwargs) + args = tuple(self._resolve_cross_db_q(a) if isinstance(a, Q) else a for a in args) + self._resolve_cross_db_kwargs(kwargs) + return super().filter(*args, **kwargs) + + def exclude(self, *args, **kwargs): + from pulpcore.app.db_router import is_multi_db_routing_active + + if not is_multi_db_routing_active(): + return super().exclude(*args, **kwargs) + args = tuple(self._resolve_cross_db_q(a) if isinstance(a, Q) else a for a in args) + self._resolve_cross_db_kwargs(kwargs) + return super().exclude(*args, **kwargs) + + def _resolve_cross_db_kwargs(self, kwargs): + for key, value in list(kwargs.items()): + if isinstance(value, models.QuerySet) and value.db != self.db: + kwargs[key] = list(value) + + def _resolve_cross_db_q(self, q): + for i, child in enumerate(q.children): + if isinstance(child, Q): + self._resolve_cross_db_q(child) + elif isinstance(child, tuple): + key, value = child + if isinstance(value, models.QuerySet) and value.db != self.db: + q.children[i] = (key, list(value)) + return q diff --git a/pulpcore/app/role_util.py b/pulpcore/app/role_util.py index 103c68541e3..a5d49426c9c 100644 --- a/pulpcore/app/role_util.py +++ b/pulpcore/app/role_util.py @@ -132,9 +132,11 @@ def get_objects_for_user_roles( ): return qs - user_role_pks = user.object_roles.filter( - domain__isnull=True, role__permissions=permission - ).values_list("object_id", flat=True) + user_role_pks = list( + user.object_roles.filter(domain__isnull=True, role__permissions=permission).values_list( + "object_id", flat=True + ) + ) final_q = Q(pk_str__in=user_role_pks) if accept_domain_perms and hasattr(qs.model, "pulp_domain"): domains = list( @@ -155,9 +157,11 @@ def get_objects_for_user_roles( ) if use_groups: - group_role_pks = GroupRole.objects.filter( - group__in=user.groups.all(), role__permissions=permission, domain__isnull=True - ).values_list("object_id", flat=True) + group_role_pks = list( + GroupRole.objects.filter( + group__in=user.groups.all(), role__permissions=permission, domain__isnull=True + ).values_list("object_id", flat=True) + ) final_q |= Q(pk_str__in=group_role_pks) return qs.annotate(pk_str=Cast("pk", output_field=CharField())).filter(final_q) @@ -575,3 +579,21 @@ def get_groups_with_perms( for_concrete_model=for_concrete_model, ) return qs.distinct() + + +def cleanup_roles_for_deleted_object(instance): + content_type = ContentType.objects.get_for_model(instance, for_concrete_model=False) + object_id = str(instance.pk) + UserRole.objects.using("default").filter( + content_type=content_type, object_id=object_id + ).delete() + GroupRole.objects.using("default").filter( + content_type=content_type, object_id=object_id + ).delete() + + +def on_any_model_post_delete(sender, instance, **kwargs): + from pulpcore.app.models import BaseModel + + if isinstance(instance, BaseModel): + cleanup_roles_for_deleted_object(instance) diff --git a/pulpcore/app/util.py b/pulpcore/app/util.py index 76f2a5b47fb..944fccb2d52 100644 --- a/pulpcore/app/util.py +++ b/pulpcore/app/util.py @@ -2,7 +2,7 @@ import os import socket import zlib -from contextlib import ExitStack +from contextlib import ExitStack, contextmanager from datetime import timedelta from functools import lru_cache from gettext import gettext as _ @@ -15,14 +15,19 @@ from django.apps import apps from django.conf import settings -from django.db import connection +from django.db import connections from django.db.models import Model, UUIDField from rest_framework.reverse import reverse as drf_reverse from rest_framework.serializers import ValidationError from pulpcore.app import models from pulpcore.app.apps import pulp_plugin_configs -from pulpcore.app.contexts import _current_domain, _current_user_func, current_pulp_api_version +from pulpcore.app.contexts import ( + _current_domain, + _current_user_func, + current_pulp_api_version, + with_domain, +) from pulpcore.app.loggers import deprecation_logger from pulpcore.exceptions.validation import InvalidSignatureError @@ -646,7 +651,7 @@ def get_domain_pk(): if default_domain: return default_domain.pk # If we haven't cached the default_domain then use raw SQL to get its PK - with connection.cursor() as cursor: + with connections["default"].cursor() as cursor: cursor.execute("SELECT pulp_id FROM core_domain WHERE name = 'default'") row = cursor.fetchone() return row[0] @@ -658,6 +663,18 @@ def set_domain(new_domain): return new_domain +@contextmanager +def domain_db(domain): + with with_domain(domain): + yield domain.database_alias + + +def for_each_domain(callback): + for domain in models.Domain.objects.all(): + with domain_db(domain) as alias: + callback(domain, alias) + + def cache_key(base_path): """Returns the base-key(s) used in the Cache for the passed base_path(s).""" if settings.DOMAIN_ENABLED: diff --git a/pulpcore/app/viewsets/task.py b/pulpcore/app/viewsets/task.py index 3dba074e95f..81e45aab62a 100644 --- a/pulpcore/app/viewsets/task.py +++ b/pulpcore/app/viewsets/task.py @@ -12,6 +12,7 @@ from pulpcore.app.models import ( AppStatus, + Artifact, CreatedResource, ProfileArtifact, RepositoryVersion, @@ -290,8 +291,10 @@ def profile_artifacts(self, request, pk, **kwargs): task = self.get_object() data = {} - for pa in ProfileArtifact.objects.select_related("artifact").filter(task=task): - data[pa.name] = get_artifact_url(pa.artifact) + alias = task.pulp_domain.database_alias + for pa in ProfileArtifact.objects.filter(task=task): + artifact = Artifact.objects.using(alias).get(pk=pa.artifact_id) + data[pa.name] = get_artifact_url(artifact) return Response({"urls": data}) diff --git a/pulpcore/tests/functional/api/test_tasking.py b/pulpcore/tests/functional/api/test_tasking.py index 4b8135882d4..62becc737cb 100644 --- a/pulpcore/tests/functional/api/test_tasking.py +++ b/pulpcore/tests/functional/api/test_tasking.py @@ -12,7 +12,7 @@ from pulpcore.client.pulpcore import ApiException from pulpcore.constants import IMMEDIATE_TIMEOUT -from pulpcore.tests.functional.utils import PulpTaskError, download_file +from pulpcore.tests.functional.utils import SLEEP_TIME, PulpTaskError, download_file @pytest.fixture(scope="module") @@ -479,10 +479,24 @@ def test_finalizer_task_runs_after_all_siblings(dispatch_task_group, monitor_tas @pytest.mark.parallel def test_cancel_task_group(pulpcore_bindings, dispatch_task_group, gen_user): """Test that task groups can be canceled.""" + cancel_retry_timeout = 60 + + def _cancel_task_group_retrying(): + deadline = time.monotonic() + cancel_retry_timeout + while True: + try: + return pulpcore_bindings.TaskGroupsApi.task_groups_cancel( + tgroup_href, {"state": "canceled"} + ) + except ApiException as e: + if e.status != 409 or time.monotonic() >= deadline: + raise + time.sleep(SLEEP_TIME) + kwargs = {"inbetween": 1, "intervals": [10, 10, 10, 10, 10]} tgroup_href = dispatch_task_group("pulpcore.app.tasks.test.dummy_group_task", kwargs=kwargs) - tgroup = pulpcore_bindings.TaskGroupsApi.task_groups_cancel(tgroup_href, {"state": "canceled"}) + tgroup = _cancel_task_group_retrying() for task in tgroup.tasks: assert task.state in ["canceled", "canceling"] @@ -500,7 +514,7 @@ def test_cancel_task_group(pulpcore_bindings, dispatch_task_group, gen_user): assert "You do not have permission" in e.value.message with gen_user(model_roles=["core.task_owner"]): - pulpcore_bindings.TaskGroupsApi.task_groups_cancel(tgroup_href, {"state": "canceled"}) + _cancel_task_group_retrying() LT_TIMEOUT = IMMEDIATE_TIMEOUT / 2 diff --git a/pulpcore/tests/unit/content/test_handler.py b/pulpcore/tests/unit/content/test_handler.py index 045efdad6cd..f29f10c2a87 100644 --- a/pulpcore/tests/unit/content/test_handler.py +++ b/pulpcore/tests/unit/content/test_handler.py @@ -8,7 +8,8 @@ from django.db import IntegrityError from django_guid import clear_guid, set_guid -from pulpcore.app.models import AppStatus +from pulpcore.app.contexts import with_domain +from pulpcore.app.models import AppStatus, Domain from pulpcore.constants import TASK_STATES from pulpcore.content.handler import CheckpointListings, Handler, PathNotResolved from pulpcore.plugin.models import ( @@ -22,6 +23,7 @@ Repository, RepositoryVersion, ) +from pulpcore.tests.unit.test_multi_database_routing import SATELLITE_ALIAS, requires_multi_db @pytest.fixture @@ -318,6 +320,49 @@ def test_pull_through_save_single_artifact_content( assert ra is not None +@requires_multi_db +@pytest.mark.django_db(databases=["default", SATELLITE_ALIAS]) +def test_pull_through_save_single_artifact_content_multi_db( + request123, download_result_mock, monkeypatch, tmp_path +): + domain = Domain.objects.create( + name="ki27-pull-through-domain", + storage_class="pulpcore.app.models.storage.FileSystem", + storage_settings={"location": str(tmp_path)}, + database_alias=SATELLITE_ALIAS, + ) + try: + with with_domain(domain): + remote = Remote.objects.create(name="123", url="https://123") + handler = Handler() + remote.get_remote_artifact_content_type = Mock(return_value=Content) + content_init_mock = Mock(return_value=Content()) + monkeypatch.setattr(Content, "init_from_artifact_and_relative_path", content_init_mock) + ca = ContentArtifact(relative_path="c123") + ra = RemoteArtifact(url=f"{remote.url}/c123", remote=remote, content_artifact=ca) + + content_artifacts = handler._save_artifact(download_result_mock, ra, request=request123) + artifact = content_artifacts[ra.content_artifact.relative_path].artifact + + assert Artifact.objects.using(SATELLITE_ALIAS).filter(pk=artifact.pk).exists() + assert not Artifact.objects.using("default").filter(pk=artifact.pk).exists() + saved_ra = ( + RemoteArtifact.objects.using(SATELLITE_ALIAS) + .filter(url=f"{remote.url}/c123", remote=remote) + .first() + ) + assert saved_ra is not None + assert saved_ra.pulp_domain_id == domain.pk + finally: + for alias in {SATELLITE_ALIAS, "default"}: + RemoteArtifact.objects.using(alias).filter(pulp_domain=domain).delete() + ContentArtifact.objects.using(alias).filter(content__pulp_domain=domain).delete() + Content.objects.using(alias).filter(pulp_domain=domain).delete() + Artifact.objects.using(alias).filter(pulp_domain=domain).delete() + Remote.objects.using(alias).filter(pulp_domain=domain).delete() + domain.delete() + + def test_pull_through_save_multi_artifact_content( remote123, request123, download_result_mock, monkeypatch, tmp_path ): diff --git a/pulpcore/tests/unit/models/test_generic.py b/pulpcore/tests/unit/models/test_generic.py new file mode 100644 index 00000000000..402bc248471 --- /dev/null +++ b/pulpcore/tests/unit/models/test_generic.py @@ -0,0 +1,65 @@ +from uuid import uuid4 + +import pytest + +from pulpcore.app.contexts import with_task_context +from pulpcore.app.models import CreatedResource, RepositoryVersion, Task +from pulpcore.app.models.generic import _resolve_domain_id + +from pulp_file.app.models import FileRepository + + +@pytest.fixture +def task(): + t = Task.objects.create(name="test-generic-relation-task") + yield t + t.delete() + + +@pytest.mark.django_db +def test_content_object_returns_none_for_deleted_domain_scoped_target(task): + with with_task_context(task): + repository = FileRepository.objects.create(name=str(uuid4())) + created_resource = CreatedResource.objects.create(content_object=repository) + assert created_resource.content_object_domain_id is not None + + repository.delete() + + created_resource = CreatedResource.objects.get(pk=created_resource.pk) + assert created_resource.content_object is None + + +@pytest.mark.django_db +def test_content_object_resolves_existing_domain_scoped_target(task): + with with_task_context(task): + repository = FileRepository.objects.create(name=str(uuid4())) + created_resource = CreatedResource.objects.create(content_object=repository) + + created_resource = CreatedResource.objects.get(pk=created_resource.pk) + resolved = created_resource.content_object + assert resolved is not None + assert resolved.pk == repository.pk + + +@pytest.mark.django_db +def test_resolve_domain_id_walks_transitive_fk(task): + with with_task_context(task): + repository = FileRepository.objects.create(name=str(uuid4())) + version = RepositoryVersion.objects.create(repository=repository, number=1) + + assert getattr(version, "pulp_domain_id", None) is None + assert _resolve_domain_id(version) == repository.pulp_domain_id + + +@pytest.mark.django_db +def test_content_object_domain_id_set_for_repository_version(task): + with with_task_context(task): + repository = FileRepository.objects.create(name=str(uuid4())) + version = RepositoryVersion.objects.create(repository=repository, number=1) + created_resource = CreatedResource.objects.create(content_object=version) + assert created_resource.content_object_domain_id == repository.pulp_domain_id + + created_resource = CreatedResource.objects.get(pk=created_resource.pk) + resolved = created_resource.content_object + assert resolved is not None + assert resolved.pk == version.pk diff --git a/pulpcore/tests/unit/models/test_remote.py b/pulpcore/tests/unit/models/test_remote.py index bb394816d37..cdf23d0b562 100644 --- a/pulpcore/tests/unit/models/test_remote.py +++ b/pulpcore/tests/unit/models/test_remote.py @@ -1,16 +1,21 @@ +from pathlib import Path from uuid import uuid4 import pytest from cryptography.fernet import InvalidToken +from django.conf import settings from django.core.management import call_command from django.db import connection +from pulpcore.app.contexts import with_domain from pulpcore.app.models import Domain, Remote from pulpcore.app.models.fields import EncryptedTextField, _fernet TEST_KEY1 = b"hPCIFQV/upbvPRsEpgS7W32XdFA2EQgXnMtyNAekebQ=" TEST_KEY2 = b"6Xyv+QezAQ+4R870F5qsgKcngzmm46caDB2gyo9qnpc=" +SATELLITE_ALIAS = "data_1" + @pytest.fixture def fake_fernet(tmp_path, settings): @@ -50,27 +55,61 @@ def test_encrypted_proxy_password(fake_fernet): assert proxy_password == "test" -@pytest.mark.django_db +@pytest.mark.django_db(databases=list(settings.DATABASES)) def test_rotate_db_key(fake_fernet): remote = Remote.objects.create(name=uuid4(), proxy_password="test") domain = Domain.objects.create(name=uuid4(), storage_settings={"base_path": "/foo"}) - next(fake_fernet) # new + old key - - call_command("rotate-db-key") - - next(fake_fernet) # new key - - del remote.proxy_password - assert remote.proxy_password == "test" - del domain.storage_settings - assert domain.storage_settings == {"base_path": "/foo"} - - next(fake_fernet) # old key - - del remote.proxy_password - with pytest.raises(InvalidToken): - remote.proxy_password - del domain.storage_settings - with pytest.raises(InvalidToken): - domain.storage_settings + satellite_remote = None + satellite_domain = None + if SATELLITE_ALIAS in settings.DATABASES: + satellite_domain = Domain.objects.create( + name=uuid4(), + storage_class="pulpcore.app.models.storage.FileSystem", + storage_settings={"base_path": "/satellite"}, + database_alias=SATELLITE_ALIAS, + ) + with with_domain(satellite_domain): + satellite_remote = Remote.objects.create( + name=uuid4(), proxy_password="satellite-secret" + ) + assert not Remote.objects.using("default").filter(pk=satellite_remote.pk).exists() + assert Remote.objects.using(SATELLITE_ALIAS).filter(pk=satellite_remote.pk).exists() + + try: + next(fake_fernet) # new + old key + + call_command("rotate-db-key") + + next(fake_fernet) # new key + + del remote.proxy_password + assert remote.proxy_password == "test" + del domain.storage_settings + assert domain.storage_settings == {"base_path": "/foo"} + + if satellite_remote is not None: + satellite_remote = Remote.objects.using(SATELLITE_ALIAS).get(pk=satellite_remote.pk) + assert satellite_remote.proxy_password == "satellite-secret" + + next(fake_fernet) # old key + + del remote.proxy_password + with pytest.raises(InvalidToken): + remote.proxy_password + del domain.storage_settings + with pytest.raises(InvalidToken): + domain.storage_settings + + if satellite_remote is not None: + with pytest.raises(InvalidToken): + Remote.objects.using(SATELLITE_ALIAS).get(pk=satellite_remote.pk) + finally: + if satellite_remote is not None or satellite_domain is not None: + key_file = Path(settings.DB_ENCRYPTION_KEY) + key_file.write_bytes(TEST_KEY2 + b"\n" + TEST_KEY1) + _fernet.cache_clear() + if satellite_remote is not None: + Remote.objects.using(SATELLITE_ALIAS).filter(pk=satellite_remote.pk).delete() + if satellite_domain is not None: + satellite_domain.delete() diff --git a/pulpcore/tests/unit/test_db_router.py b/pulpcore/tests/unit/test_db_router.py new file mode 100644 index 00000000000..f47a388789e --- /dev/null +++ b/pulpcore/tests/unit/test_db_router.py @@ -0,0 +1,48 @@ +import pytest +from django.db import router as django_router + +from pulpcore.app.db_router import PulpDomainRouter, _database_alias, is_multi_db_routing_active +from pulpcore.app.models import Domain + + +@pytest.mark.django_db +def test_database_alias_reads_loaded_field(): + domain = Domain.objects.get(name="default") + assert _database_alias(domain) == "default" + + +@pytest.mark.django_db +def test_database_alias_does_not_query_when_field_is_deferred(django_assert_num_queries): + domain = Domain.objects.only("pk", "name").get(name="default") + assert "database_alias" not in domain.__dict__, ( + "test setup assumption broken: .only('pk', 'name') should defer 'database_alias'" + ) + with django_assert_num_queries(0): + assert _database_alias(domain) == "default" + + +@pytest.mark.django_db +def test_database_alias_reads_non_default_value_when_loaded(): + domain = Domain.objects.get(name="default") + domain.__dict__["database_alias"] = "data_1" + assert _database_alias(domain) == "data_1" + + +def test_is_multi_db_routing_active_false_by_default(): + original_routers = django_router.routers + try: + django_router.routers = [] + assert is_multi_db_routing_active() is False + finally: + django_router.routers = original_routers + + +def test_is_multi_db_routing_active_true_when_registered_then_false_after(): + original_routers = django_router.routers + try: + django_router.routers = [] + assert is_multi_db_routing_active() is False + django_router.routers = [PulpDomainRouter()] + assert is_multi_db_routing_active() is True + finally: + django_router.routers = original_routers diff --git a/pulpcore/tests/unit/test_multi_database_routing.py b/pulpcore/tests/unit/test_multi_database_routing.py new file mode 100644 index 00000000000..8ddb55e36ac --- /dev/null +++ b/pulpcore/tests/unit/test_multi_database_routing.py @@ -0,0 +1,118 @@ +from contextlib import contextmanager + +import pytest +from django.conf import settings + +from pulpcore.app.contexts import with_domain +from pulpcore.app.db_router import is_multi_db_routing_active +from pulpcore.app.models import ContentArtifact, Domain, Remote, RemoteArtifact, Repository, Task +from pulpcore.constants import TASK_STATES + +SATELLITE_ALIAS = "data_1" + +requires_multi_db = pytest.mark.skipif( + SATELLITE_ALIAS not in settings.DATABASES or not is_multi_db_routing_active(), + reason=( + f"Multi-database routing tests require a '{SATELLITE_ALIAS}' alias in settings.DATABASES " + f"(set PULP_DATABASES__{SATELLITE_ALIAS}__* env vars to a second real Postgres instance) " + "and PulpDomainRouter registered in DATABASE_ROUTERS." + ), +) + +pytestmark = [requires_multi_db, pytest.mark.django_db(databases=["default", SATELLITE_ALIAS])] + + +@contextmanager +def _satellite_domain(**extra_fields): + domain = Domain.objects.create( + name=f"test-satellite-domain-{extra_fields.get('_suffix', '')}".rstrip("-"), + storage_class="pulpcore.app.models.storage.FileSystem", + database_alias=SATELLITE_ALIAS, + **{k: v for k, v in extra_fields.items() if k != "_suffix"}, + ) + try: + yield domain + finally: + domain.delete() + + +class TestPulpDomainRouter: + def test_data_plane_object_routes_to_satellite_alias(self): + with _satellite_domain(_suffix="routing") as domain: + with with_domain(domain): + repo = Repository.objects.create(name=f"{domain.name}-repo", pulp_domain=domain) + try: + assert Repository.objects.using(SATELLITE_ALIAS).filter(pk=repo.pk).exists(), ( + "Repository created under a satellite-domain context should exist on the " + "satellite alias." + ) + assert not Repository.objects.using("default").filter(pk=repo.pk).exists(), ( + "Repository created under a satellite-domain context must NOT exist on " + "'default' -- routing to the wrong alias would silently duplicate/leak data." + ) + finally: + Repository.objects.using(SATELLITE_ALIAS).filter(pk=repo.pk).delete() + + def test_instance_hint_routes_without_contextvar(self): + with _satellite_domain(_suffix="instancehint") as domain: + with with_domain(domain): + repo = Repository.objects.create(name=f"{domain.name}-repo", pulp_domain=domain) + try: + repo_fresh = ( + Repository.objects.using(SATELLITE_ALIAS) + .select_related("pulp_domain") + .get(pk=repo.pk) + ) + repo_fresh.description = "updated via instance hint, no ContextVar" + repo_fresh.save() + assert ( + Repository.objects.using(SATELLITE_ALIAS).get(pk=repo.pk).description + == "updated via instance hint, no ContextVar" + ) + finally: + Repository.objects.using(SATELLITE_ALIAS).filter(pk=repo.pk).delete() + + def test_control_plane_model_always_routes_to_default(self): + with _satellite_domain(_suffix="controlplane") as domain: + with with_domain(domain): + task = Task.objects.create(name="test-task", state=TASK_STATES.WAITING) + try: + assert Task.objects.using("default").filter(pk=task.pk).exists() + assert not Task.objects.using(SATELLITE_ALIAS).filter(pk=task.pk).exists() + finally: + Task.objects.using("default").filter(pk=task.pk).delete() + + +class TestRouterInstanceHintSafety: + def test_remote_artifact_construction_does_not_recurse(self): + with _satellite_domain(_suffix="norecursion") as domain: + with with_domain(domain): + remote = Remote.objects.create(name="ki27-remote", url="https://example.com") + ca = ContentArtifact(relative_path="ki27/path") + try: + ra = RemoteArtifact(remote=remote, url=f"{remote.url}/x", content_artifact=ca) + except RecursionError: + pytest.fail( + "PulpDomainRouter._resolve_db recursed while constructing a " + "RemoteArtifact with a preceding unsaved FK" + ) + try: + assert ra.pulp_domain_id == domain.pk + finally: + Remote.objects.using(SATELLITE_ALIAS).filter(pk=remote.pk).delete() + + def test_relation_access_does_not_issue_extra_domain_query(self, django_assert_num_queries): + from pulp_file.app.models import FileRemote, FileRepository + + remote = FileRemote.objects.create(name="ki27-cast-remote") + repository = FileRepository.objects.create(name="ki27-cast-repo", remote=remote) + try: + with django_assert_num_queries(1): + fetched = Repository.objects.get(pk=repository.pk) + with django_assert_num_queries(1): + fetched = fetched.cast() + with django_assert_num_queries(1): + assert fetched.remote.pk == remote.pk + finally: + repository.delete() + remote.delete()