From 2d9f12364d6a28fd22c47477186b2d88e98678ad Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Wed, 9 Sep 2026 20:56:45 -0700 Subject: [PATCH 1/2] fix(ml): connect new projects to the pull-mode workers that process their images A new project was only connected to the push-mode service named by DEFAULT_PROCESSING_SERVICE_ENDPOINT. Asynchronous jobs find a worker through the project-to-service relation, and asynchronous is the dispatch mode new projects use, so a project created on a deployment served by polling workers had no way to process anything. Registering pipelines also fetches the service over HTTP from inside project creation, so a service that had moved or lost its certificate turned every new project into a server error, after the project row had already been written. - Connect a new project to every pull-mode processing service, with a config row per pipeline. No network call: those workers register their own pipelines. - Treat pipeline registration from the push-mode default as best-effort. The project is created without those pipelines and the failure is logged. - Extract configure_pipeline_for_project, shared by create_pipelines and the new path. This also fixes create_pipelines reporting the wrong pipelines as newly created, because the inner loop overwrote the flag the outer loop set. - Comment the endpoint out of the production env example, since a deployment on polling workers should not set it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CnUB2nqxi92uGqisKdJKkX --- .envs/.production/.django-example | 9 ++- ami/main/models.py | 9 ++- ami/main/tests.py | 83 ++++++++++++++++++++++ ami/ml/models/processing_service.py | 106 +++++++++++++++++++++------- 4 files changed, 178 insertions(+), 29 deletions(-) diff --git a/.envs/.production/.django-example b/.envs/.production/.django-example index 1b8b6b0b8..a672ba866 100644 --- a/.envs/.production/.django-example +++ b/.envs/.production/.django-example @@ -141,8 +141,13 @@ DJANGO_ACCOUNT_ALLOW_REGISTRATION=True WEB_CONCURRENCY=4 # Default processing service -DEFAULT_PROCESSING_SERVICE_NAME="AMI Data Companion" -DEFAULT_PROCESSING_SERVICE_ENDPOINT=https://ml.antenna.insectai.org/ +# Only needed when a single processing service is reached over HTTP at a fixed address. +# Workers that poll for tasks register themselves and need no endpoint here; new projects +# are connected to those automatically. Leave this unset unless you run a push-mode service, +# and point it somewhere reachable if you do — an address that no longer answers means new +# projects are created without its pipelines. +# DEFAULT_PROCESSING_SERVICE_NAME="AMI Data Companion" +# DEFAULT_PROCESSING_SERVICE_ENDPOINT=https://ml.example.org/ DEFAULT_PIPELINES_ENABLED=global_moths_2024,quebec_vermont_moths_2023,panama_moths_2023,uk_denmark_moths_2023 # NATS diff --git a/ami/main/models.py b/ami/main/models.py index 78e67e275..88323d543 100644 --- a/ami/main/models.py +++ b/ami/main/models.py @@ -263,8 +263,15 @@ def create_related_defaults(self, project: "Project"): if not project.sourceimage_collections.exists(): get_or_create_default_collection(project=project) if not project.processing_services.exists(): - from ami.ml.models.processing_service import get_or_create_default_processing_service + from ami.ml.models.processing_service import ( + attach_async_processing_services, + get_or_create_default_processing_service, + ) + # Pull-mode workers run the platform's jobs and are how a new project processes + # anything, since new jobs dispatch asynchronously by default. A push-mode + # service is added on top only where one is configured, such as local development. + attach_async_processing_services(project) get_or_create_default_processing_service(project=project) diff --git a/ami/main/tests.py b/ami/main/tests.py index e8f56c485..4b3f1ac0b 100644 --- a/ami/main/tests.py +++ b/ami/main/tests.py @@ -5,6 +5,7 @@ from io import BytesIO from unittest import mock +import requests from django.conf import settings from django.contrib.auth.models import AnonymousUser from django.core.files.uploadedfile import SimpleUploadedFile @@ -135,6 +136,88 @@ def test_processing_service_if_not_configured(self): service, "Default processing service should not be created if environment variables are not set." ) + def _create_pull_mode_service(self, name="Pull-mode Worker", pipeline_slugs=("pipeline_a", "pipeline_b")): + """Register a worker that polls for tasks, with the pipelines it reports it can run.""" + service = ProcessingService.objects.create(name=name, endpoint_url=None) + for slug in pipeline_slugs: + pipeline = Pipeline.objects.create(slug=slug, name=slug, version=1) + service.pipelines.add(pipeline) + return service + + @override_settings( + DEFAULT_PROCESSING_SERVICE_NAME=None, + DEFAULT_PROCESSING_SERVICE_ENDPOINT=None, + DEFAULT_PIPELINES_ENABLED=["pipeline_a"], + ) + def test_new_project_is_connected_to_pull_mode_services(self): + """ + A new project can run a job on the platform's pull-mode workers. + + An asynchronous job looks up workers through the project-to-service relation, and + asynchronous is the dispatch mode new projects use, so a project created without that + relation has no way to process anything. + """ + service = self._create_pull_mode_service() + + project = Project.objects.create(name="Project on pull-mode workers", create_defaults=True) + + self.assertIn(service, project.processing_services.all()) + configs = ProjectPipelineConfig.objects.filter(project=project) + self.assertEqual({config.pipeline.slug for config in configs}, {"pipeline_a", "pipeline_b"}) + self.assertEqual( + {config.pipeline.slug for config in configs if config.enabled}, + {"pipeline_a"}, + "Only the pipelines named in DEFAULT_PIPELINES_ENABLED should start enabled.", + ) + + @override_settings( + DEFAULT_PROCESSING_SERVICE_NAME="Default Processing Service", + DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://ml_backend:2000/", + DEFAULT_PIPELINES_ENABLED=None, + ) + def test_push_mode_service_is_not_attached_unless_it_is_the_configured_default(self): + """ + Only the configured default push-mode service is attached, not every push-mode service. + + Push-mode services are called at a specific endpoint, so attaching a project to one + that was registered for somebody else would send that project's work to it. + """ + # The manager health-checks a new service on creation; this one is never meant to be + # reached, and letting the check run costs the suite a DNS timeout. + with mock.patch.object(ProcessingService, "get_status"): + other = ProcessingService.objects.create( + name="Someone else's service", endpoint_url="http://elsewhere:2000/" + ) + + project = Project.objects.create(name="Project with an unrelated service", create_defaults=True) + + self.assertNotIn(other, project.processing_services.all()) + + @override_settings( + DEFAULT_PROCESSING_SERVICE_NAME="Unreachable Service", + DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://unreachable.invalid:2000/", + DEFAULT_PIPELINES_ENABLED=None, + ) + def test_project_is_still_created_when_the_default_service_cannot_be_reached(self): + """ + Creating a project survives a default processing service that cannot be reached. + + Registering pipelines fetches the service over the network. That request failing used + to propagate out of the create call, so the request that created the project returned + a server error and the caller could not tell that the project had been written. + """ + with mock.patch.object( + ProcessingService, + "create_pipelines", + side_effect=requests.exceptions.SSLError("certificate verify failed"), + ): + project = Project.objects.create(name="Project with an unreachable service", create_defaults=True) + + self.assertGreaterEqual(project.deployments.count(), 1) + self.assertGreaterEqual(project.sourceimage_collections.count(), 1) + self.assertGreaterEqual(project.processing_services.count(), 1) + self.assertEqual(ProjectPipelineConfig.objects.filter(project=project).count(), 0) + @override_settings( DEFAULT_PROCESSING_SERVICE_NAME="Default Processing Service", DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://ml_backend:2000/", diff --git a/ami/ml/models/processing_service.py b/ami/ml/models/processing_service.py index fce1aefc5..dd16c68e7 100644 --- a/ami/ml/models/processing_service.py +++ b/ami/ml/models/processing_service.py @@ -90,6 +90,18 @@ class Meta: verbose_name = "Processing Service" verbose_name_plural = "Processing Services" + def add_project(self, project: "Project", enable_only: list[str] | None = None) -> None: + """ + Connect a project to this service and configure the pipelines the service already has. + + No request is made to the service. Use this when the pipelines are registered + already and only the project needs connecting, which is the case for pull-mode + workers that register themselves. + """ + self.projects.add(project) + for pipeline in self.pipelines.all(): + configure_pipeline_for_project(project, pipeline, enable_only) + def create_pipelines( self, enable_only: list[str] | None = None, @@ -122,22 +134,7 @@ def create_pipelines( created = True for project in projects: - if enable_only is not None and pipeline.slug not in enable_only: - enabled = False - else: - enabled = True - project_pipeline_config, created = ProjectPipelineConfig.objects.get_or_create( - pipeline=pipeline, - project=project, - defaults={"enabled": enabled, "config": {}}, - ) - if created: - logger.debug( - f"Created project pipeline config for {project.name} and {pipeline.name} (enabled: {enabled})." - ) - project_pipeline_config.save() - else: - logger.debug(f"Using existing project pipeline config for {project.name} and {pipeline.name}.") + configure_pipeline_for_project(project, pipeline, enable_only) self.pipelines.add(pipeline) @@ -301,15 +298,62 @@ def get_pipeline_configs(self, timeout=6): return info_data.pipelines +def configure_pipeline_for_project( + project: "Project", + pipeline: "Pipeline", + enable_only: list[str] | None = None, +) -> "ProjectPipelineConfig": + """ + Give a project its configuration row for a pipeline, so the pipeline can be selected. + + ``enable_only`` of None starts every pipeline enabled. A list of slugs starts only those + enabled, which keeps a new project from offering every pipeline the platform has. + """ + enabled = True if enable_only is None else pipeline.slug in enable_only + config, created = ProjectPipelineConfig.objects.get_or_create( + pipeline=pipeline, + project=project, + defaults={"enabled": enabled, "config": {}}, + ) + if created: + logger.debug(f"Created project pipeline config for {project.name} and {pipeline.name} (enabled: {enabled}).") + else: + logger.debug(f"Using existing project pipeline config for {project.name} and {pipeline.name}.") + return config + + +def attach_async_processing_services(project: "Project") -> list["ProcessingService"]: + """ + Connect a project to every pull-mode processing service registered on the platform. + + Pull-mode workers poll for tasks instead of exposing an endpoint, and an async job only + reaches a worker through this project-to-service link. A project with none attached has + no way to run a job in the dispatch mode that new projects use by default, so a new + project is connected to the pull-mode fleet the same way existing projects are. + + Nothing is requested over the network: these services register their own pipelines when + they check in. + """ + services = list(ProcessingService.objects.async_services()) + for service in services: + service.add_project(project, enable_only=settings.DEFAULT_PIPELINES_ENABLED) + if services: + logger.info(f"Connected project {project} to {len(services)} pull-mode processing services") + else: + logger.info(f"No pull-mode processing services are registered; project {project} was connected to none") + return services + + def get_or_create_default_processing_service( project: "Project", register_pipelines: bool = True, ) -> "ProcessingService | None": """ - Create a default processing service for a project. + Create the push-mode default processing service for a project, if one is configured. - If configured, will use the global default processing service - for the current environment. Otherwise, it return None. + This covers deployments that talk to a single processing service over HTTP, such as a + local development stack. Returns None when no endpoint is configured, which is the + normal case for a deployment served by pull-mode workers. Set the "DEFAULT_PROCESSING_SERVICE_ENDPOINT" and "DEFAULT_PROCESSING_SERVICE_NAME" environment variables to configure & enable the default processing service. @@ -318,9 +362,9 @@ def get_or_create_default_processing_service( name = settings.DEFAULT_PROCESSING_SERVICE_NAME or "Default Processing Service" endpoint_url = settings.DEFAULT_PROCESSING_SERVICE_ENDPOINT if not endpoint_url: - logger.warning( - "Default processing service is not configured. " - "Set the 'DEFAULT_PROCESSING_SERVICE_ENDPOINT' environment variable." + logger.info( + "No push-mode default processing service is configured. " + "Set the 'DEFAULT_PROCESSING_SERVICE_ENDPOINT' environment variable to add one." ) return None @@ -331,8 +375,18 @@ def get_or_create_default_processing_service( service.projects.add(project) logger.info(f"Created default processing service for project {project}") if register_pipelines: - service.create_pipelines( - enable_only=settings.DEFAULT_PIPELINES_ENABLED, - projects=Project.objects.filter(pk=project.pk), - ) + # Registering pipelines fetches the service's /info endpoint. Creating a project is + # otherwise a local operation, so an unreachable or misconfigured service must not + # take it down with it: the project is created without those pipelines and the + # service can be registered again later from the admin. + try: + service.create_pipelines( + enable_only=settings.DEFAULT_PIPELINES_ENABLED, + projects=Project.objects.filter(pk=project.pk), + ) + except Exception: + logger.exception( + f"Could not register pipelines from the default processing service '{name}'. " + f"Project {project} was created without them." + ) return service From 3d36d1adcebee1d2bb3989918fec79f9b4c1755b Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Thu, 10 Sep 2026 17:09:46 -0700 Subject: [PATCH 2/2] fix(ml): only connect new projects to processing services marked public A project manager can register a pull-mode worker for their own project, and attaching every pull-mode service to each new project would send other projects' images to it. Add ProcessingService.is_public (admin-only, not in the project API) and attach only services marked public. Each shared pipeline is now configured once rather than once per worker. Also from review: - Run the default service's pipeline registration inside a savepoint, so a database error there no longer aborts the project-creation transaction. - Report newly registered algorithms in algorithms_created instead of pipelines_created. - Pin that project creation does not run the default service's health check (get_or_create bypasses the manager's create override). Co-Authored-By: Claude Opus 5 --- .envs/.production/.django-example | 4 +- ami/main/models.py | 10 +-- ami/main/tests.py | 68 +++++++++++------- ami/ml/admin.py | 2 + .../0029_processing_service_is_public.py | 20 ++++++ ami/ml/models/processing_service.py | 70 +++++++++---------- 6 files changed, 104 insertions(+), 70 deletions(-) create mode 100644 ami/ml/migrations/0029_processing_service_is_public.py diff --git a/.envs/.production/.django-example b/.envs/.production/.django-example index a672ba866..c2addc931 100644 --- a/.envs/.production/.django-example +++ b/.envs/.production/.django-example @@ -142,8 +142,8 @@ WEB_CONCURRENCY=4 # Default processing service # Only needed when a single processing service is reached over HTTP at a fixed address. -# Workers that poll for tasks register themselves and need no endpoint here; new projects -# are connected to those automatically. Leave this unset unless you run a push-mode service, +# Workers that poll for tasks register themselves and need no endpoint here; new projects are +# connected to the ones marked "public" in the admin. Leave this unset unless you run a push-mode service, # and point it somewhere reachable if you do — an address that no longer answers means new # projects are created without its pipelines. # DEFAULT_PROCESSING_SERVICE_NAME="AMI Data Companion" diff --git a/ami/main/models.py b/ami/main/models.py index 88323d543..48e2bac5a 100644 --- a/ami/main/models.py +++ b/ami/main/models.py @@ -264,14 +264,14 @@ def create_related_defaults(self, project: "Project"): get_or_create_default_collection(project=project) if not project.processing_services.exists(): from ami.ml.models.processing_service import ( - attach_async_processing_services, + attach_public_processing_services, get_or_create_default_processing_service, ) - # Pull-mode workers run the platform's jobs and are how a new project processes - # anything, since new jobs dispatch asynchronously by default. A push-mode - # service is added on top only where one is configured, such as local development. - attach_async_processing_services(project) + # The platform's own workers, marked public by an admin, are how a new project + # processes anything. A push-mode service is added on top only where one is + # configured, such as local development. + attach_public_processing_services(project) get_or_create_default_processing_service(project=project) diff --git a/ami/main/tests.py b/ami/main/tests.py index 4b3f1ac0b..72d537938 100644 --- a/ami/main/tests.py +++ b/ami/main/tests.py @@ -136,11 +136,11 @@ def test_processing_service_if_not_configured(self): service, "Default processing service should not be created if environment variables are not set." ) - def _create_pull_mode_service(self, name="Pull-mode Worker", pipeline_slugs=("pipeline_a", "pipeline_b")): + def _create_pull_mode_service(self, name, pipeline_slugs=("pipeline_a", "pipeline_b"), is_public=False): """Register a worker that polls for tasks, with the pipelines it reports it can run.""" - service = ProcessingService.objects.create(name=name, endpoint_url=None) + service = ProcessingService.objects.create(name=name, endpoint_url=None, is_public=is_public) for slug in pipeline_slugs: - pipeline = Pipeline.objects.create(slug=slug, name=slug, version=1) + pipeline, _ = Pipeline.objects.get_or_create(slug=slug, defaults={"name": slug, "version": 1}) service.pipelines.add(pipeline) return service @@ -149,19 +149,21 @@ def _create_pull_mode_service(self, name="Pull-mode Worker", pipeline_slugs=("pi DEFAULT_PROCESSING_SERVICE_ENDPOINT=None, DEFAULT_PIPELINES_ENABLED=["pipeline_a"], ) - def test_new_project_is_connected_to_pull_mode_services(self): + def test_new_project_is_connected_to_public_processing_services_only(self): """ - A new project can run a job on the platform's pull-mode workers. + A new project is connected to the workers marked public, and to no other worker. - An asynchronous job looks up workers through the project-to-service relation, and - asynchronous is the dispatch mode new projects use, so a project created without that - relation has no way to process anything. + A job only reaches a worker through the project-to-service relation, so without the + public workers a new project cannot process anything. A worker that a project owner + registered for their own project must not be attached, or it would receive other + projects' images. """ - service = self._create_pull_mode_service() + fleet = {self._create_pull_mode_service(f"Platform worker {i}", is_public=True) for i in (1, 2)} + self._create_pull_mode_service("A lab's own worker", pipeline_slugs=("pipeline_c",)) - project = Project.objects.create(name="Project on pull-mode workers", create_defaults=True) + project = Project.objects.create(name="Project on the platform workers", create_defaults=True) - self.assertIn(service, project.processing_services.all()) + self.assertEqual(set(project.processing_services.all()), fleet) configs = ProjectPipelineConfig.objects.filter(project=project) self.assertEqual({config.pipeline.slug for config in configs}, {"pipeline_a", "pipeline_b"}) self.assertEqual( @@ -198,25 +200,39 @@ def test_push_mode_service_is_not_attached_unless_it_is_the_configured_default(s DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://unreachable.invalid:2000/", DEFAULT_PIPELINES_ENABLED=None, ) - def test_project_is_still_created_when_the_default_service_cannot_be_reached(self): + def test_project_is_still_created_when_the_default_service_cannot_be_registered(self): """ - Creating a project survives a default processing service that cannot be reached. + A default processing service that fails to register does not stop a project being created. - Registering pipelines fetches the service over the network. That request failing used - to propagate out of the create call, so the request that created the project returned - a server error and the caller could not tell that the project had been written. + Registration fetches the service over the network, and a failed request must not turn + project creation into a server error. A database error during registration must be + contained too, or it aborts the transaction the project is written in. Creation also + must not wait on a health check of the service, which retries for minutes when the + host does not answer. """ - with mock.patch.object( - ProcessingService, - "create_pipelines", - side_effect=requests.exceptions.SSLError("certificate verify failed"), - ): - project = Project.objects.create(name="Project with an unreachable service", create_defaults=True) - self.assertGreaterEqual(project.deployments.count(), 1) - self.assertGreaterEqual(project.sourceimage_collections.count(), 1) - self.assertGreaterEqual(project.processing_services.count(), 1) - self.assertEqual(ProjectPipelineConfig.objects.filter(project=project).count(), 0) + def fail_in_the_database(*args, **kwargs): + with connection.cursor() as cursor: + cursor.execute("SELECT 1/0") + + failures = { + "network": requests.exceptions.SSLError("certificate verify failed"), + "database": fail_in_the_database, + } + for label, failure in failures.items(): + with ( + self.subTest(failure=label), + mock.patch.object(ProcessingService, "create_pipelines", side_effect=failure), + mock.patch.object(ProcessingService, "get_status") as get_status, + ): + project = Project.objects.create(name=f"Unregistered service ({label})", create_defaults=True) + + get_status.assert_not_called() + self.assertTrue(Project.objects.filter(pk=project.pk).exists()) + self.assertGreaterEqual(project.deployments.count(), 1) + self.assertGreaterEqual(project.sourceimage_collections.count(), 1) + self.assertGreaterEqual(project.processing_services.count(), 1) + self.assertEqual(ProjectPipelineConfig.objects.filter(project=project).count(), 0) @override_settings( DEFAULT_PROCESSING_SERVICE_NAME="Default Processing Service", diff --git a/ami/ml/admin.py b/ami/ml/admin.py index 008b20e84..1a8a78a89 100644 --- a/ami/ml/admin.py +++ b/ami/ml/admin.py @@ -70,8 +70,10 @@ class ProcessingServiceAdmin(AdminBase): "id", "name", "endpoint_url", + "is_public", "created_at", ] + list_filter = ["is_public"] @admin.register(AlgorithmCategoryMap) diff --git a/ami/ml/migrations/0029_processing_service_is_public.py b/ami/ml/migrations/0029_processing_service_is_public.py new file mode 100644 index 000000000..2fc01e2ec --- /dev/null +++ b/ami/ml/migrations/0029_processing_service_is_public.py @@ -0,0 +1,20 @@ +# Generated by Django 4.2.10 on 2026-09-10 20:02 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("ml", "0028_normalize_empty_endpoint_url_to_null"), + ] + + operations = [ + migrations.AddField( + model_name="processingservice", + name="is_public", + field=models.BooleanField( + default=False, + help_text="Connect new projects to this service automatically. Mark only services run for the whole platform: a project's images are sent to every service it is connected to.", + ), + ), + ] diff --git a/ami/ml/models/processing_service.py b/ami/ml/models/processing_service.py index dd16c68e7..0975978ac 100644 --- a/ami/ml/models/processing_service.py +++ b/ami/ml/models/processing_service.py @@ -6,7 +6,7 @@ import requests from django.conf import settings -from django.db import models +from django.db import models, transaction from ami.base.models import BaseQuerySet from ami.main.models import BaseModel, Project @@ -70,6 +70,13 @@ class ProcessingService(BaseModel): last_seen = models.DateTimeField(null=True) last_seen_live = models.BooleanField(null=True) last_seen_latency = models.FloatField(null=True) + is_public = models.BooleanField( + default=False, + help_text=( + "Connect new projects to this service automatically. Mark only services run for the " + "whole platform: a project's images are sent to every service it is connected to." + ), + ) objects = ProcessingServiceManager() @@ -90,18 +97,6 @@ class Meta: verbose_name = "Processing Service" verbose_name_plural = "Processing Services" - def add_project(self, project: "Project", enable_only: list[str] | None = None) -> None: - """ - Connect a project to this service and configure the pipelines the service already has. - - No request is made to the service. Use this when the pipelines are registered - already and only the project needs connecting, which is the case for pull-mode - workers that register themselves. - """ - self.projects.add(project) - for pipeline in self.pipelines.all(): - configure_pipeline_for_project(project, pipeline, enable_only) - def create_pipelines( self, enable_only: list[str] | None = None, @@ -150,7 +145,7 @@ def create_pipelines( if algorithm not in existing_algorithms: logger.debug(f"Registered new algorithm {algorithm.name} to pipeline {pipeline.name}.") pipeline.algorithms.add(algorithm) - pipelines_created.append(algorithm.key) + algorithms_created.append(algorithm.key) else: logger.debug(f"Using existing algorithm {algorithm.name}.") @@ -322,25 +317,26 @@ def configure_pipeline_for_project( return config -def attach_async_processing_services(project: "Project") -> list["ProcessingService"]: +def attach_public_processing_services(project: "Project") -> list["ProcessingService"]: """ - Connect a project to every pull-mode processing service registered on the platform. + Connect a project to the processing services marked public, so it can run jobs from the start. - Pull-mode workers poll for tasks instead of exposing an endpoint, and an async job only - reaches a worker through this project-to-service link. A project with none attached has - no way to run a job in the dispatch mode that new projects use by default, so a new - project is connected to the pull-mode fleet the same way existing projects are. + A job only reaches a service through this project-to-service link, so a new project with + none attached cannot process anything. Only services an admin has marked ``is_public`` are + attached, never a worker that a project owner registered for their own project. - Nothing is requested over the network: these services register their own pipelines when - they check in. + Nothing is requested over the network: the pipelines these services have already + registered are configured for the project. """ - services = list(ProcessingService.objects.async_services()) - for service in services: - service.add_project(project, enable_only=settings.DEFAULT_PIPELINES_ENABLED) - if services: - logger.info(f"Connected project {project} to {len(services)} pull-mode processing services") - else: - logger.info(f"No pull-mode processing services are registered; project {project} was connected to none") + services = list(ProcessingService.objects.filter(is_public=True)) + if not services: + logger.info(f"No processing services are marked public; project {project} was connected to none") + return [] + project.processing_services.add(*services) + # Workers in a fleet usually offer the same pipelines, so configure each pipeline once. + for pipeline in Pipeline.objects.filter(processing_services__in=services).distinct(): + configure_pipeline_for_project(project, pipeline, settings.DEFAULT_PIPELINES_ENABLED) + logger.info(f"Connected project {project} to {len(services)} public processing services") return services @@ -375,15 +371,15 @@ def get_or_create_default_processing_service( service.projects.add(project) logger.info(f"Created default processing service for project {project}") if register_pipelines: - # Registering pipelines fetches the service's /info endpoint. Creating a project is - # otherwise a local operation, so an unreachable or misconfigured service must not - # take it down with it: the project is created without those pipelines and the - # service can be registered again later from the admin. + # Registering pipelines fetches the service's /info endpoint, and an unreachable or + # misconfigured service must not take project creation down with it. The savepoint + # discards a partial registration without breaking the caller's transaction. try: - service.create_pipelines( - enable_only=settings.DEFAULT_PIPELINES_ENABLED, - projects=Project.objects.filter(pk=project.pk), - ) + with transaction.atomic(): + service.create_pipelines( + enable_only=settings.DEFAULT_PIPELINES_ENABLED, + projects=Project.objects.filter(pk=project.pk), + ) except Exception: logger.exception( f"Could not register pipelines from the default processing service '{name}'. "