DM-48838: Create Gafaelfawr tokens in parallel

changelog.d/20250207_162856_rra_DM_48838.md

@@ -0,0 +1,3 @@
+### Bug fixes
+
+- When starting a flock, create user tokens simultaneously (up to the limit of the httpx connection pool size of 100) rather than serially.

src/mobu/factory.py

@@ -51,7 +51,6 @@ def __init__(self, http_client: AsyncClient, events: Events) -> None:
             logger=self.logger,
             events=self.events,
         )
-        self.events = events
 
     async def aclose(self) -> None:
         """Clean up a process context.

src/mobu/services/business/base.py

@@ -10,7 +10,6 @@
 from enum import Enum
 from typing import Generic, TypedDict, TypeVar
 
-from httpx import AsyncClient
 from safir.datetime import current_datetime
 from structlog.stdlib import BoundLogger
 
@@ -62,8 +61,6 @@ class Business(Generic[T], metaclass=ABCMeta):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client.
     events
         Event publishers.
     logger
@@ -77,8 +74,6 @@ class Business(Generic[T], metaclass=ABCMeta):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client.
     events
         Event publishers.
     logger
@@ -101,14 +96,12 @@ def __init__(
         *,
         options: T,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         self.options = options
         self.user = user
-        self.http_client = http_client
         self.events = events
         self.logger = logger
         self.success_count = 0

src/mobu/services/business/gitlfs.py

@@ -7,7 +7,6 @@
 from pathlib import Path
 from urllib.parse import urlparse
 
-from httpx import AsyncClient
 from safir.sentry import duration
 from structlog.stdlib import BoundLogger
 
@@ -30,15 +29,13 @@ def __init__(
         *,
         options: GitLFSBusinessOptions,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         super().__init__(
             options=options,
             user=user,
-            http_client=http_client,
             events=events,
             logger=logger,
             flock=flock,

src/mobu/services/business/notebookrunner.py

@@ -19,7 +19,6 @@
 
 import sentry_sdk
 import yaml
-from httpx import AsyncClient
 from rubin.nublado.client import JupyterLabSession
 from rubin.nublado.client.exceptions import CodeExecutionError
 from rubin.nublado.client.models import CodeContext
@@ -71,8 +70,6 @@ class NotebookRunner(NubladoBusiness):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client for general web access.
     events
         Event publishers.
     logger
@@ -86,15 +83,13 @@ def __init__(
         *,
         options: NotebookRunnerOptions | ListNotebookRunnerOptions,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         super().__init__(
             options=options,
             user=user,
-            http_client=http_client,
             events=events,
             logger=logger,
             flock=flock,

src/mobu/services/business/nublado.py

@@ -12,7 +12,6 @@
 from typing import Generic, TypeVar
 
 import sentry_sdk
-from httpx import AsyncClient
 from rubin.nublado.client import JupyterLabSession, NubladoClient
 from safir.datetime import current_datetime, format_datetime_for_logging
 from safir.sentry import duration
@@ -101,8 +100,6 @@ class NubladoBusiness(Business, Generic[T], metaclass=ABCMeta):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client for general web access.
     events
         Event publishers.
     logger
@@ -114,15 +111,13 @@ def __init__(
         *,
         options: T,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         super().__init__(
             options=options,
             user=user,
-            http_client=http_client,
             events=events,
             logger=logger,
             flock=flock,

src/mobu/services/business/nubladopythonloop.py

@@ -9,7 +9,6 @@
 from datetime import timedelta
 
 import sentry_sdk
-from httpx import AsyncClient
 from rubin.nublado.client import JupyterLabSession
 from rubin.nublado.client.exceptions import CodeExecutionError
 from safir.sentry import duration
@@ -33,8 +32,6 @@ class NubladoPythonLoop(NubladoBusiness):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client for general web access.
     logger
         Logger to use to report the results of business.
     flock
@@ -46,15 +43,13 @@ def __init__(
         *,
         options: NubladoPythonLoopOptions,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         super().__init__(
             options=options,
             user=user,
-            http_client=http_client,
             events=events,
             logger=logger,
             flock=flock,

src/mobu/services/business/tap.py

@@ -9,7 +9,6 @@
 
 import pyvo
 import requests
-from httpx import AsyncClient
 from safir.sentry import duration
 from sentry_sdk import set_context
 from structlog.stdlib import BoundLogger
@@ -41,8 +40,6 @@ class TAPBusiness(Business, Generic[T], metaclass=ABCMeta):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client for general web access.
     events
         Event publishers.
     logger
@@ -54,15 +51,13 @@ def __init__(
         *,
         options: T,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         super().__init__(
             options=options,
             user=user,
-            http_client=http_client,
             events=events,
             logger=logger,
             flock=flock,

src/mobu/services/business/tapqueryrunner.py

@@ -4,7 +4,6 @@
 
 from random import SystemRandom
 
-from httpx import AsyncClient
 from structlog.stdlib import BoundLogger
 
 from ...events import Events
@@ -24,8 +23,6 @@ class TAPQueryRunner(TAPBusiness):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client for general web access.
     events
         Event publishers.
     logger
@@ -39,15 +36,13 @@ def __init__(
         *,
         options: TAPQueryRunnerOptions,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         super().__init__(
             options=options,
             user=user,
-            http_client=http_client,
             events=events,
             logger=logger,
             flock=flock,

src/mobu/services/business/tapquerysetrunner.py

@@ -9,7 +9,6 @@
 import jinja2
 import shortuuid
 import yaml
-from httpx import AsyncClient
 from structlog.stdlib import BoundLogger
 
 from ...events import Events
@@ -29,8 +28,6 @@ class TAPQuerySetRunner(TAPBusiness):
         Configuration options for the business.
     user
         User with their authentication token to use to run the business.
-    http_client
-        Shared HTTP client for general web access.
     events
         Event publishers.
     logger
@@ -44,15 +41,13 @@ def __init__(
         *,
         options: TAPQuerySetRunnerOptions,
         user: AuthenticatedUser,
-        http_client: AsyncClient,
         events: Events,
         logger: BoundLogger,
         flock: str | None,
     ) -> None:
         super().__init__(
             options=options,
             user=user,
-            http_client=http_client,
             events=events,
             logger=logger,
             flock=flock,

src/mobu/services/flock.py

@@ -179,10 +179,10 @@ async def _create_users(self) -> list[AuthenticatedUser]:
             count = self._config.count
             users = self._users_from_spec(self._config.user_spec, count)
         scopes = self._config.scopes
-        return [
-            await self._gafaelfawr.create_service_token(u, scopes)
-            for u in users
+        coros = [
+            self._gafaelfawr.create_service_token(u, scopes) for u in users
         ]
+        return await asyncio.gather(*coros)
 
     def _users_from_spec(self, spec: UserSpec, count: int) -> list[User]:
         """Generate count Users from the provided spec."""

src/mobu/services/monkey.py

@@ -96,7 +96,6 @@ def __init__(
             self.business = EmptyLoop(
                 options=business_config.options,
                 user=user,
-                http_client=self._http_client,
                 events=self._events,
                 logger=self._logger,
                 flock=self._flock,
@@ -105,7 +104,6 @@ def __init__(
             self.business = GitLFSBusiness(
                 options=business_config.options,
                 user=user,
-                http_client=self._http_client,
                 events=self._events,
                 logger=self._logger,
                 flock=self._flock,
@@ -114,7 +112,6 @@ def __init__(
             self.business = NubladoPythonLoop(
                 options=business_config.options,
                 user=user,
-                http_client=self._http_client,
                 events=self._events,
                 logger=self._logger,
                 flock=self._flock,
@@ -123,7 +120,6 @@ def __init__(
             self.business = NotebookRunner(
                 options=business_config.options,
                 user=user,
-                http_client=self._http_client,
                 events=self._events,
                 logger=self._logger,
                 flock=self._flock,
@@ -132,7 +128,6 @@ def __init__(
             self.business = TAPQueryRunner(
                 options=business_config.options,
                 user=user,
-                http_client=self._http_client,
                 events=self._events,
                 logger=self._logger,
                 flock=self._flock,
@@ -141,7 +136,6 @@ def __init__(
             self.business = TAPQuerySetRunner(
                 options=business_config.options,
                 user=user,
-                http_client=self._http_client,
                 events=self._events,
                 logger=self._logger,
                 flock=self._flock,

tests/business/tapquerysetrunner_test.py

@@ -13,7 +13,6 @@
 import yaml
 from anys import ANY_AWARE_DATETIME_STR, AnyContains, AnySearch, AnyWithEntries
 from httpx import AsyncClient
-from safir.dependencies.http_client import http_client_dependency
 from safir.metrics import NOT_NONE, MockEventPublisher
 from safir.testing.sentry import Captured
 
@@ -224,12 +223,10 @@ async def test_random_object(events: Events) -> None:
         )
         logger = structlog.get_logger(__file__)
         options = TAPQuerySetRunnerOptions(query_set=query_set)
-        http_client = await http_client_dependency()
         with patch.object(pyvo.dal, "TAPService"):
             runner = TAPQuerySetRunner(
                 options=options,
                 user=user,
-                http_client=http_client,
                 events=events,
                 logger=logger,
                 flock=None,