Flexible scheduling

[jpbruinsslot]

Context

The scheduler now references the /random endpoint from octopoes to get ooi's for rescheduling of tasks (ooi * available boefjes = tasks). To keep track of recurring tasks within the scheduler we allow for:

• a more deterministic approach to scheduling and rescheduling of tasks, so we don't rely on randomness from objects returned by octopoes.
• more fine-grained control of the scheduling of specific tasks, potential for scheduling job independent of signals from octopoes, katalogus, and octopoes random endpoint.
• allows for scheduling other tasks such as reports
• scheduling of tasks without an associated ooi, i.e. independent run boefjes
• scheduling of normalizer tasks, e.g. run a specific normalizer task at an interval regardless of a signal from bytes.
• give better insight into what tasks are scheduled and at what time, and how many times the specific task has been executed
• leads into new ranking possibilities

Proposed solution

Changes

Change 1: Schedule model

Create a new model Schedule which contains the necessary information to create a new Task. A Schedule has a 1-to-many relationship with an instanced Task. The updated entity relationship diagram for the scheduler:

erDiagram
task {
    id uuid PK
    scheduler_id str
    schedule_id uuid FK
    hash str
    priority int
    status taskstatus
    data jsonb
    created_at timestamp
    modified_at timestamp
}

schedule {
    id uuid PK
    scheduler_id str
    hash str
    data jsonb
    enabled bool
    schedule str
    tasks list[task]
    deadline_at timestamp
    created_at timestamp
    modified_at timestamp
}

task }o--|| schedule: ""

Loading

With both models looking like follows:

nl-kat-coordination/mula/scheduler/models/schema.py

Lines 22 to 41 in 3942029

	class TaskSchema(BaseModel):
	model_config = ConfigDict(from_attributes=True, validate_assignment=True)
	id: uuid.UUID = Field(default_factory=uuid.uuid4)
	scheduler_id: str
	hash: str | None = Field(None, max_length=32)
	data: dict | None = {}
	enabled: bool = True
	schedule: str | None = None
	tasks: list[Task] = []
	deadline_at: datetime | None = None
	created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
	modified_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

[!NOTE] This model is in the process to be rename to Schedule.

nl-kat-coordination/mula/scheduler/models/task.py

Lines 47 to 66 in 3942029

	class Task(BaseModel):
	model_config = ConfigDict(from_attributes=True)
	id: uuid.UUID = Field(default_factory=uuid.uuid4)
	scheduler_id: str
	schema_id: uuid.UUID | None = None
	# schema: TaskSchema ## FIXME: naming conflict with pydantic .schema()
	priority: int | None = 0
	status: TaskStatus = TaskStatus.PENDING
	hash: str | None = Field(None, max_length=32)
	data: dict | None = {}
	created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
	modified_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

The data field contains the actual data that is needed for a task runner to execute its task, in case of a BoefjeScheduler this is a BoefjeTask:

{
  "id": "",
  "boefje": {},
  "input_ooi":  "",
  "organization": ""
}

Change 2: Removal of the PrioritizedItem table and model

Removal of the PrioritizedItem model and table. By removing this, a substantial decrease in data duplication within the scheduler has been achieved. The queue is now 'materialized' by filtering on the task table based on the status of a Task, i.e. all task with the status QUEUED. In practice this means the following, when a task is created (either by manual entry, enabling of boefjes, rescheduling, etc) this task is added to the task table, and for every task that gets created a Schedule is created as well.

Example of both Task and Schedule table, here a Schedule already scheduled a Task in the past, and QUEUED a Task in the future.

Schedule table:

id	scheduler_id	hash	data	manual	enabled	schedule	deadline_at	created_at	modified_at
6f904	boefje-org0	1234	{}	true	true	0 12 * * 1	2020-01-01	2019-01-01	2020-01-01

[!NOTE] a Schedule can be manually defined by a cron expression

Task table:

id	scheduler_id	schedule_id	hash	priority	status	data	created_at	updated_at
59cb0	boefje-org0	6f904	1234	1	COMPLETED	{}	2019-01-01	2019-01-01
122e2	boefje-org0	6f904	1234	1	QUEUED	{}	2020-01-01	2020-01-01

When a task with the highest priority needs to be removed from the queue the following code is then being executed:

nl-kat-coordination/mula/scheduler/storage/pq_store.py

Lines 16 to 34 in 3942029

	def pop(self, scheduler_id: str, filters: FilterRequest | None = None) -> models.Task | None:
	with self.dbconn.session.begin() as session:
	query = (
	session.query(models.TaskDB)
	.filter(models.TaskDB.status == models.TaskStatus.QUEUED)
	.order_by(models.TaskDB.priority.asc())
	.order_by(models.TaskDB.created_at.asc())
	.filter(models.TaskDB.scheduler_id == scheduler_id)
	)
	if filters is not None:
	query = apply_filter(models.TaskDB, query, filters)
	item_orm = query.first()
	if item_orm is None:
	return None
	return models.Task.model_validate(item_orm)

The rescheduling of tasks can be imagined as follows: the BoefjeScheduler queries the Schedule table where the deadline has passed (meaning a task needs to be scheduled now, and pushed onto the queue). A Task will be created and inserted into the Task table. A Task Runner can then retrieve a task through the server API, run the task and update the status when it finished.

Tasklist

• Flexible scheduling: refactor feature branch to be up-to-date to main branch #2295
• Flexible scheduling: fix tests for current implementation #2296
• Flexible scheduling: write boefje scheduler tests #2285
• Flexible scheduling: write tests for 'no ooi' boefjes #2287
• Flexible scheduling: write job_store tests #2286
• Flexible scheduling: solve job creation problem for 'one-off' and tasks that should not be rescheduled #2300
• Flexible scheduling: refactor Job.p_item attribute #2313
• Flexible scheduling: create jobs api endpoints #2288
• Flexible scheduling: create tests for jobs endpoints #2306
• Flexible scheduling: create basic deadline calculation algorithm #2289
• Flexible scheduling: create cron like functionality (optional) #2290
• Flexible scheduling: refactor naming of entities #2372
• Flexible scheduling: fix conflicts with main branch #2591
• Flexible scheduling: p_item data deduplication #2611
• Flexible scheduling: p_item data deduplication > data model definitions #2650
• Flexible scheduling: p_item data deduplication > refactor migrations #2651
• Flexible scheduling: p_item data deduplication > refactor schedulers #2652
• Flexible scheduling: refactor tests #2654
• Flexible scheduling: refactor server #2653
• Flexible scheduling: write migration for data field migration #3120
• Flexible scheduling: refactor TaskSchema to Schedule #2989
• Flexible scheduling: solve merge conflicts #2947
• Flexible scheduling: refactor boefje scheduler for pushing tasks #3156
• Flexible scheduling: refactor normalizer scheduler for pushing tasks #3159
• Flexible scheduling: fix merge conflicts #3160
• Flexible scheduling: fix tests for server api #3166
• Flexible Scheduling: refactor API endpoints to more correctly represent models #2631
• Flexible scheduling: Write tests for test_scheduler.py #3276
• Flexible scheduling: make pr workable for interface development #3198
• Create rate limiting functionality #2291
• Flexible scheduling: review and fix, unittest.skip, TODO's and FIXME's #2292
• Flexible scheduling: finalize testing (stress test) #2293
• Flexible scheduling: update rocky scheduler client to support the new models and new endpoints #2656
• Flexible scheduling: update documentation #2294
• Flexible scheduling: update documentation #3121
• Flexible scheduling: transfer atomic changes to separate pr's #3278

Pull Request

• Flexible scheduling #2786

Associated issues

• Get all ooi's from an organisation that a boefje can run on #199
• Implement additional Katalogus filter options #191
• [Feature] Flexible Scheduling of Boefjes #205
• [mula] Backoff timer multiplier support #1318

[jpbruinsslot]

The design has been agreed upon, with the advice to make sure that database normalization would be possible in the future