Changes for BoefjeScheduler to support deduplication

mula/pyproject.toml

@@ -51,6 +51,10 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [tool.coverage.run]
 relative_files = true
+data_file = "/tmp/.coverage"
+
+[tool.coverage.xml]
+output = "/tmp/coverage.xml"
 
 omit = [
     "scheduler/alembic/*",

mula/scheduler/clients/http/external/katalogus.py

@@ -4,7 +4,7 @@
 
 from scheduler.clients.errors import exception_handler
 from scheduler.clients.http import HTTPService
-from scheduler.models import Boefje, Organisation, Plugin
+from scheduler.models import Boefje, BoefjeConfig, Organisation, Plugin
 
 
 class Katalogus(HTTPService):
@@ -138,3 +138,29 @@ def get_new_boefjes_by_org_id(self, organisation_id: str) -> list[Plugin]:
             )
 
             return new_boefjes
+
+    def get_configs(
+        self,
+        organisation_id: str,
+        boefje_id: str | None = None,
+        enabled: bool | None = None,
+        offset: int | None = None,
+        limit: int | None = None,
+    ) -> list[BoefjeConfig]:
+        url = f"{self.host}/v1/configs"
+        try:
+            response = self.get(
+                url,
+                params={
+                    "organisation_id": organisation_id,
+                    "boefje_id": boefje_id,
+                    "enabled": enabled,
+                    "offset": offset,
+                    "limit": limit,
+                },
+            )
+            return [Plugin(**plugin) for plugin in response.json()]
+        except httpx.HTTPStatusError as e:
+            if e.response.status_code == httpx.codes.NOT_FOUND:
+                return []
+            raise

mula/scheduler/clients/http/external/octopoes.py

@@ -98,6 +98,10 @@ def get_object(self, organisation_id: str, reference: str) -> OOI | None:
                 return None
             raise
 
+    @exception_handler
+    def get_organisations_by_ooi(self, reference: str) -> list[Organisation]:
+        raise NotImplementedError("Not implemented")
+
     def is_healthy(self) -> bool:
         healthy = True
         for org in self.orgs:

mula/scheduler/models/__init__.py

@@ -1,5 +1,5 @@
 from .base import Base
-from .boefje import Boefje, BoefjeMeta
+from .boefje import Boefje, BoefjeConfig, BoefjeMeta
 from .events import RawData, RawDataReceivedEvent
 from .health import ServiceHealth
 from .normalizer import Normalizer

mula/scheduler/models/boefje.py

@@ -25,3 +25,14 @@ class BoefjeMeta(BaseModel):
 
     started_at: datetime.datetime | None
     ended_at: datetime.datetime | None
+
+
+class BoefjeConfig(BaseModel):
+    """BoefjeConfig is the configuration object for a Boefje"""
+
+    id: int
+    boefje_id: str
+    enabled: bool
+    organisation_id: str
+    env_hash: str
+    settings: dict

mula/scheduler/models/plugin.py

@@ -24,3 +24,4 @@ class Plugin(BaseModel):
     run_on: list[RunOn] | None = None
     oci_image: str | None = None
     oci_arguments: list[str] | None = None
+    env_hash: str | None = None

mula/scheduler/models/task.py

@@ -110,6 +110,7 @@ class BoefjeTask(BaseModel):
     boefje: Boefje
     input_ooi: str | None = None
     organization: str
+    env_hash: str  # TODO: will this be none?
 
     dispatches: list[Normalizer] = Field(default_factory=list)
 

mula/scheduler/schedulers/schedulers/boefje.py

@@ -162,6 +162,7 @@ def process_mutations(self, body: bytes) -> None:
                     boefje=models.Boefje.model_validate(boefje.model_dump()),
                     input_ooi=ooi.primary_key if ooi else None,
                     organization=mutation.client_id,
+                    env_hash=boefje.env_hash,
                 )
             )
 
@@ -204,6 +205,7 @@ def process_new_boefjes(self) -> None:
                             boefje=models.Boefje.model_validate(boefje.dict()),
                             input_ooi=ooi.primary_key,
                             organization=org.id,
+                            env_hash=boefje.env_hash,
                         )
 
                         boefje_tasks.append((boefje_task, org.id))
@@ -339,6 +341,7 @@ def process_rescheduling(self):
                         boefje=models.Boefje.model_validate(plugin.dict()),
                         input_ooi=ooi.primary_key if ooi else None,
                         organization=schedule.organisation,
+                        env_hash=plugin.env_hash,
                     )
                 except (StorageError, ValidationError, ExternalServiceError):
                     self.logger.exception(
@@ -402,6 +405,14 @@ def push_boefje_task(
             task_db.status = models.TaskStatus.FAILED
             self.ctx.datastores.task_store.update_task(task_db)
 
+        # We check on input_ooi because we allow for boefje tasks without an
+        # ooi, something we can't deduplicate. Additionally, because we
+        # potentially call the method itself from `is_ooi_in_other_organisations`,
+        # we need to make sure that we don't endlessly call it, by checking the
+        # caller.
+        if boefje_task.input_ooi is not None and caller != self.is_ooi_in_other_organisations.__name__:
+            self.is_ooi_in_other_organisations(boefje_task)
+
         task = models.Task(
             id=boefje_task.id,
             scheduler_id=self.scheduler_id,
@@ -685,6 +696,49 @@ def get_oois_for_boefje(self, boefje: models.Plugin, organisation: str) -> list[
 
         return oois
 
+    # TODO:: exception handling
+    def is_ooi_in_other_organisations(self, boefje_task: models.BoefjeTask):
+        """Is the ooi in other organisations? When the ooi is in other
+        organisations we don't want to scan those as well, we still create
+        tasks for those organisations but they will be batched together.
+        """
+        # From the boefje id we can get the same boefje in other organisations.
+        configs = self.ctx.services.katalogus.get_configs(boefje_id=boefje_task.boefje.id, enabled=True)
+        for config in configs:
+            if config.organisation_id == boefje_task.organization:
+                # Skip the organisation we are currently in
+                continue
+
+            ooi = self.ctx.services.octopoes.get_object(config.organisation_id, boefje_task.input_ooi)
+
+            boefje = self.ctx.services.katalogus.get_plugin_by_id_and_org_id(
+                boefje_task.boefje.id, config.organisation_id
+            )
+            # TODO: check if we need this, or if this is already handled
+            if not self.has_boefje_permission_to_run(boefje, ooi):
+                self.logger.debug(
+                    "Boefje not allowed to run on ooi",
+                    boefje_id=boefje_task.boefje.id,
+                    ooi_primary_key=ooi.primary_key,
+                    organisation_id=config.organisation_id,
+                    scheduler_id=self.scheduler_id,
+                )
+                continue
+
+            new_boefje_task = models.BoefjeTask(
+                boefje=models.Boefje.model_validate(boefje.model_dump()),
+                input_ooi=ooi.primary_key,
+                organization=config.organisation_id,
+                env_hash=boefje.env_hash,  # FIXME: get this from config?
+            )
+
+            self.push_boefje_task(
+                boefje_task=new_boefje_task,
+                organisation_id=config.organisation_id,
+                create_schedule=self.create_schedule,
+                caller=self.is_ooi_in_other_organisations.__name__,
+            )
+
     def calculate_deadline(self, schedule: models.Schedule) -> models.Schedule:
         """Override Scheduler.calculate_deadline() to calculate the deadline
         for a task and based on the boefje interval."""

mula/tests/factories/plugin.py

@@ -1,3 +1,5 @@
+import uuid
+
 from factory import Factory, LazyFunction, Sequence, fuzzy
 from scheduler.models import Plugin
 from scheduler.models.ooi import RunOn
@@ -15,3 +17,4 @@ class Meta:
     cron: str | None = None
     interval: int | None = None
     run_on: RunOn | None = None
+    env_hash: str = uuid.uuid4().hex