diff --git a/.envs/.production/.django-example b/.envs/.production/.django-example index 1b8b6b0b8..c2addc931 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 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" +# 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..48e2bac5a 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_public_processing_services, + get_or_create_default_processing_service, + ) + # 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 e8f56c485..72d537938 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,104 @@ 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, 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, is_public=is_public) + for slug in pipeline_slugs: + pipeline, _ = Pipeline.objects.get_or_create(slug=slug, defaults={"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_public_processing_services_only(self): + """ + A new project is connected to the workers marked public, and to no other worker. + + 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. + """ + 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 the platform workers", create_defaults=True) + + 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( + {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_registered(self): + """ + A default processing service that fails to register does not stop a project being created. + + 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. + """ + + 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", DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://ml_backend:2000/", 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 fce1aefc5..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() @@ -122,22 +129,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) @@ -153,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}.") @@ -301,15 +293,63 @@ 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_public_processing_services(project: "Project") -> list["ProcessingService"]: + """ + Connect a project to the processing services marked public, so it can run jobs from the start. + + 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: the pipelines these services have already + registered are configured for the project. + """ + 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 + + 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 +358,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 +371,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, 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: + 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}'. " + f"Project {project} was created without them." + ) return service