habsi/authentik
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

lib/sync/outgoing: sync in parallel (#14697)

Browse Source 
Co-authored-by: Jens Langhammer <jens@goauthentik.io>
This commit is contained in:
Marc 'risson' Schmitt
2025-05-27 15:26:43 +02:00
committed by GitHub
parent 245126a1c3
commit e905dd52d8
2 changed files with 45 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
authentik/lib/sync/outgoing/tasks.py
View File

@ -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:

2
authentik/providers/scim/tests/test_user.py
View File

@ -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"])