From e905dd52d879e7944eb4448901bb773b062ec73f Mon Sep 17 00:00:00 2001 From: Marc 'risson' Schmitt Date: Tue, 27 May 2025 15:26:43 +0200 Subject: [PATCH] lib/sync/outgoing: sync in parallel (#14697) Co-authored-by: Jens Langhammer --- authentik/lib/sync/outgoing/tasks.py | 58 ++++++++++++++++----- authentik/providers/scim/tests/test_user.py | 2 +- 2 files changed, 45 insertions(+), 15 deletions(-) diff --git a/authentik/lib/sync/outgoing/tasks.py b/authentik/lib/sync/outgoing/tasks.py index 48cc4a6f31..a6c8dd1175 100644 --- a/authentik/lib/sync/outgoing/tasks.py +++ b/authentik/lib/sync/outgoing/tasks.py @@ -1,6 +1,7 @@ from collections.abc import Callable from dataclasses import asdict +from celery import group from celery.exceptions import Retry from celery.result import allow_join_result from django.core.paginator import Paginator @@ -82,21 +83,41 @@ class SyncTasks: self.logger.debug("Failed to acquire sync lock, skipping", provider=provider.name) return try: - for page in users_paginator.page_range: - messages.append(_("Syncing page {page} of users".format(page=page))) - for msg in sync_objects.apply_async( - args=(class_to_path(User), page, provider_pk), - time_limit=PAGE_TIMEOUT, - soft_time_limit=PAGE_TIMEOUT, - ).get(): + messages.append(_("Syncing users")) + user_results = ( + group( + [ + sync_objects.signature( + args=(class_to_path(User), page, provider_pk), + time_limit=PAGE_TIMEOUT, + soft_time_limit=PAGE_TIMEOUT, + ) + for page in users_paginator.page_range + ] + ) + .apply_async() + .get() + ) + for result in user_results: + for msg in result: messages.append(LogEvent(**msg)) - for page in groups_paginator.page_range: - messages.append(_("Syncing page {page} of groups".format(page=page))) - for msg in sync_objects.apply_async( - args=(class_to_path(Group), page, provider_pk), - time_limit=PAGE_TIMEOUT, - soft_time_limit=PAGE_TIMEOUT, - ).get(): + messages.append(_("Syncing groups")) + group_results = ( + group( + [ + sync_objects.signature( + args=(class_to_path(Group), page, provider_pk), + time_limit=PAGE_TIMEOUT, + soft_time_limit=PAGE_TIMEOUT, + ) + for page in groups_paginator.page_range + ] + ) + .apply_async() + .get() + ) + for result in group_results: + for msg in result: messages.append(LogEvent(**msg)) except TransientSyncException as exc: self.logger.warning("transient sync exception", exc=exc) @@ -132,6 +153,15 @@ class SyncTasks: self.logger.debug("starting discover") client.discover() self.logger.debug("starting sync for page", page=page) + messages.append( + asdict( + LogEvent( + _("Syncing page {page} of groups".format(page=page)), + log_level="info", + logger=f"{provider._meta.verbose_name}@{object_type}", + ) + ) + ) for obj in paginator.page(page).object_list: obj: Model try: diff --git a/authentik/providers/scim/tests/test_user.py b/authentik/providers/scim/tests/test_user.py index c7fc2c7c40..582ba03814 100644 --- a/authentik/providers/scim/tests/test_user.py +++ b/authentik/providers/scim/tests/test_user.py @@ -384,7 +384,7 @@ class SCIMUserTests(TestCase): self.assertIn(request.method, SAFE_METHODS) task = SystemTask.objects.filter(uid=slugify(self.provider.name)).first() self.assertIsNotNone(task) - drop_msg = task.messages[2] + drop_msg = task.messages[3] self.assertEqual(drop_msg["event"], "Dropping mutating request due to dry run") self.assertIsNotNone(drop_msg["attributes"]["url"]) self.assertIsNotNone(drop_msg["attributes"]["body"])