Combined schedulers

[jpbruinsslot]

Warning

Ready for review, not for merging

Changes

• app.py: Removal of creation of organisational schedulers. We don't gather all available organisations from the katalogus and create individual boefje, report, and normalizer schedulers.
• app.py: Removal of continuous checking of added or removed organisations. We removed the monitor_organisations method that was running in a thread to check if we added or removed organisation and create schedulers for it.
• models/ : Task and Schedule get a specific organisation field. Since we don't differentiate tasks that are pushed on an organisation queue anymore, we need to be able to determine for what organisation an item on a queue is from.¹
• schedulers/: Since we've consolidated the organisational schedulers to 3 types of schedulers, and we assume that we get task creation from consolidated message queues, we need to update the schedulers to handle that.
  • BoefjeScheduler
    • retrieve scan profile mutations from one message queue
    • iterate over organisation to check for enables boefjes²
    • refactor of method names and exception handling
• server/
  • Merger of queues/ and schedulers/ endpoint. The notion of a queue and scheduler are used interchangeably. Therefore it makes more sense at the moment to merge those endpoints to avoid confusion and simplify the code.
  • Support to pop off more than one Task from a scheduler. Task runner will now be able to batch pop tasks from a scheduler using filters.

Next steps and impact

• Pop endpoint has changed from /queues to /schedulers/{id}/pop , additionally it will return a paginated result instead of a single Task , this is because the pop endpoint now supports filtering with multiple tasks returns. Services that rely on the scheduler pop endpoint need to update their interfaces (Update services that rely on /pop endpoint of scheduler #3961)
• Push endpoint changed from /queues to schedulers/{id}/push, services that interface with the push endpoints (rocky) need to update their interfaces. (Update services that rely on /push endpoint of scheduler #3962)
• scan profile mutation message queue, for every organisation a message queue is created for scan profile mutations, this needs to be updated and all scan profile mutations for every organisation needs to be relayed on a single scan profile mutations message queue (Combine organisational scan profile mutation message queues #3963)
• raw data file received message queue (the as with scan profile mutations) (Combine raw data recieved message queues #3964)
• Model definitions updates: organisation fields are added to Task , Schedule , services using these models need to update their specifications. (Model definition updates: addition of organisation field to Task and Schedule models #3965)
• Batch status updates. Several places in the scheduler we can consider batch updating status field of task. Potentially exposing this as a endpoint. This because the task runner will be able to pop off multiple task from a scheduler, it might therefore be beneficial to batch update the tasks. (
• Deleting of organisations needs to be addressed, and what the protocol needs to be with regards of queued tasks, and schedules. (Discussion: how do we handle the deletion of organisations? #3966)

Issue link

#3838

QA notes

The process of scanning tasks, rescheduling of tasks should remain the same from a users standpoint of view. Testing this would be the main focus point. Additionally since all organisational schedulers have been combined to one, particular emphasis should be made to check if users will only see, or are able to interact with tasks that are for the organisation that they are allowed to.

---

Code Checklist

• All the commits in this PR are properly PGP-signed and verified.
• This PR only contains functionality relevant to the issue.
• I have written unit tests for the changes or fixes I made.
• I have checked the documentation and made changes where necessary.
• I have performed a self-review of my code and refactored it to the best of my abilities.

• Tickets have been created for newly discovered issues.
• For any non-trivial functionality, I have added integration and/or end-to-end tests.
• I have informed others of any required .env changes files if required and changed the .env-dist accordingly.
• I have included comments in the code to elaborate on what is not self-evident from the code itself, including references to issues and discussions online, or implicit behavior of an interface.

---

Checklist for code reviewers:

Copy-paste the checklist from the docs/source/templates folder into your comment.

---

Checklist for QA:

Copy-paste the checklist from the docs/source/templates folder into your comment.

Footnotes

1. The attentive reader will know that the organisation information is present in the data specification in a Task , and can be filtered on by using a FilterRequest. However in practice the scheduler is mainly used in the context of multiple organisations and filtering on this id is deemed developer friendly when interfacing with the scheduler. Hence the choice explicitly defining the organisation field on a top level. ↩

2. as mentioned in the code this iterating over all organisations in order to check if there are enabled plugins is very inefficient, please refer to issue Katalogus caching in the scheduler #3357 ↩

[stephanie0x00]

Current state (16/12) of this PR.

• Tasks don't seem to run. I did the onboarding and no tasks for boefjes or normalizers show up on the Tasks page. However, some findings are shown on the Findings page (which remain pending).

No tasks on the tasks page:

Findings page for this state:

boefje-1  | HTTPStatusError: Client error '404 Not Found' for url 'http://scheduler:8000/queues'
boefje-1  | For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
boefje-1  | Traceback (most recent call last):
boefje-1  |   File "/app/boefjes/boefjes/app.py", line 89, in _fill_queue
boefje-1  |     queues = self.scheduler_client.get_queues()
boefje-1  |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
boefje-1  |   File "/app/boefjes/boefjes/clients/scheduler_client.py", line 71, in get_queues
boefje-1  |     self._verify_response(response)
boefje-1  |   File "/app/boefjes/boefjes/clients/scheduler_client.py", line 67, in _verify_response
boefje-1  |     response.raise_for_status()
boefje-1  |   File "/usr/local/lib/python3.11/site-packages/httpx/_models.py", line 763, in raise_for_status
boefje-1  |     raise HTTPStatusError(message, request=request, response=self)
boefje-1  | httpx.HTTPStatusError: Client error '404 Not Found' for url 'http://scheduler:8000/queues'
boefje-1  | For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404
boefje-1  | HTTP Request: GET http://scheduler:8000/queues "HTTP/1.1 404 Not Found"
boefje-1  | 2024-12-16T11:47:16.426262 [error] Getting the queues from the scheduler failed
boefje-1  | ╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
boefje-1  | │ /app/boefjes/boefjes/app.py:89 in _fill_queue                                                    │
boefje-1  | │                                                                                                  │
boefje-1  | │    86 │   │   │   return                                                                         │
boefje-1  | │    87 │   │                                                                                      │
boefje-1  | │    88 │   │   try:                                                                               │
boefje-1  | │ ❱  89 │   │   │   queues = self.scheduler_client.get_queues()                                    │
boefje-1  | │    90 │   │   except HTTPError:                                                                  │
boefje-1  | │    91 │   │   │   # Scheduler is having issues, so make note of it and try again                 │
boefje-1  | │    92 │   │   │   logger.exception("Getting the queues from the scheduler failed")               │
boefje-1  | │                                                                                                  │
boefje-1  | │ ╭────────────────────────────────── locals ──────────────────────────────────╮                   │
boefje-1  | │ │ queue_type = <Queue.BOEFJES: 'boefje'>                                     │                   │
boefje-1  | │ │       self = <boefjes.app.SchedulerWorkerManager object at 0x74d6402b2e10> │                   │
boefje-1  | │ │ task_queue = <AutoProxy[Queue] object, typeid 'Queue' at 0x74d6406533d0>   │                   │
boefje-1  | │ ╰────────────────────────────────────────────────────────────────────────────╯                   │
boefje-1  | │                                                                                                  │
boefje-1  | │ /app/boefjes/boefjes/clients/scheduler_client.py:71 in get_queues                                │
boefje-1  | │                                                                                                  │
boefje-1  | │   68 │                                                                                           │
boefje-1  | │   69 │   def get_queues(self) -> list[Queue]:                                                    │
boefje-1  | │   70 │   │   response = self._session.get("/queues")                                             │
boefje-1  | │ ❱ 71 │   │   self._verify_response(response)                                                     │
boefje-1  | │   72 │   │                                                                                       │
boefje-1  | │   73 │   │   return TypeAdapter(list[Queue]).validate_json(response.content)                     │
boefje-1  | │   74                                                                                             │
boefje-1  | │                                                                                                  │
boefje-1  | │ ╭───────────────────────────────────────── locals ──────────────────────────────────────────╮    │
boefje-1  | │ │ response = <Response [404 Not Found]>                                                     │    │
boefje-1  | │ │     self = <boefjes.clients.scheduler_client.SchedulerAPIClient object at 0x74d6402b2890> │    │
boefje-1  | │ ╰───────────────────────────────────────────────────────────────────────────────────────────╯    │
boefje-1  | │                                                                                                  │
boefje-1  | │ /app/boefjes/boefjes/clients/scheduler_client.py:67 in _verify_response                          │
boefje-1  | │                                                                                                  │
boefje-1  | │   64 │                                                                                           │
boefje-1  | │   65 │   @staticmethod                                                                           │
boefje-1  | │   66 │   def _verify_response(response: Response) -> None:                                       │
boefje-1  | │ ❱ 67 │   │   response.raise_for_status()                                                         │
boefje-1  | │   68 │                                                                                           │
boefje-1  | │   69 │   def get_queues(self) -> list[Queue]:                                                    │
boefje-1  | │   70 │   │   response = self._session.get("/queues")                                             │
boefje-1  | │                                                                                                  │
boefje-1  | │ ╭─────────────── locals ────────────────╮                                                        │
boefje-1  | │ │ response = <Response [404 Not Found]> │                                                        │
boefje-1  | │ ╰───────────────────────────────────────╯                                                        │
boefje-1  | │                                                                                                  │
boefje-1  | │ /usr/local/lib/python3.11/site-packages/httpx/_models.py:763 in raise_for_status                 │
boefje-1  | │                                                                                                  │
boefje-1  | │    760 │   │   }                                                                                 │
boefje-1  | │    761 │   │   error_type = error_types.get(status_class, "Invalid status code")                 │
boefje-1  | │    762 │   │   message = message.format(self, error_type=error_type)                             │
boefje-1  | │ ❱  763 │   │   raise HTTPStatusError(message, request=request, response=self)                    │
boefje-1  | │    764 │                                                                                         │
boefje-1  | │    765 │   def json(self, **kwargs: typing.Any) -> typing.Any:                                   │
boefje-1  | │    766 │   │   return jsonlib.loads(self.content, **kwargs)                                      │
boefje-1  | │                                                                                                  │
boefje-1  | │ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
boefje-1  | │ │   error_type = 'Client error'                                                                │ │
boefje-1  | │ │  error_types = {                                                                             │ │
boefje-1  | │ │                │   1: 'Informational response',                                              │ │
boefje-1  | │ │                │   3: 'Redirect response',                                                   │ │
boefje-1  | │ │                │   4: 'Client error',                                                        │ │
boefje-1  | │ │                │   5: 'Server error'                                                         │ │
boefje-1  | │ │                }                                                                             │ │
boefje-1  | │ │      message = "Client error '404 Not Found' for url 'http://scheduler:8000/queues'\nFor     │ │
boefje-1  | │ │                more inf"+76                                                                  │ │
boefje-1  | │ │      request = <Request('GET', 'http://scheduler:8000/queues')>                              │ │
boefje-1  | │ │         self = <Response [404 Not Found]>                                                    │ │
boefje-1  | │ │ status_class = 4                                                                             │ │
boefje-1  | │ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
boefje-1  | ╰──────────────────────────────────────────────────────────────────────────────────────────────────╯

[ammar92]

Looks good to me now. There are still some test errors, I think some of the test still use the older create_item utility function.

And about the QA test results: I'm not sure (yet) how the 404 for the /queues endpoint is related to this branch. If there was an error on that list endpoint, it shouldn't be a 404

[jpbruinsslot]

Looks good to me now. There are still some test errors, I think some of the test still use the older create_item utility function.

And about the QA test results: I'm not sure (yet) how the 404 for the /queues endpoint is related to this branch. If there was an error on that list endpoint, it shouldn't be a 404

The PR is at its current state not able to be QA'ed. In the description I've added the next steps that need to be taken in order to integrate this PR into openkat. There are substantial changes to both the API and the message queues, these changes need to be integrated in the systems that rely and use those.

In the next steps section I also created the issues related to those integrations and services that need to be picked up and developed.

[sonarqubecloud]

Quality Gate failed

Failed conditions
33.8% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube Cloud