Dynamic workers

[haraldsvik]

Uses a class to store state about workers.

The class stores:

• default_max_workers - Max workers we allow
• max_gb_all_workers - Max total size(disk size of tar) of the jobs in the pipeline
• workers - List of workers (updated Worker to have job_size as property)

How it works:

1. When we want to start a new job we first check the size of the .tar file
2. Then we check if we are allowed to spawn a new job:
   2.1 If alive workers >= default_max_workers and current_total_size + new_job_size >= max size in pipeline
   2.2 we allow new job!
3. if we are allowed to spawn the new worker, we then register the job in the state.
4. when the job finished we unregister the job.

Note:

This can cause large jobs to be prioritized later than small jobs:
e.g.
default_max_workers=4
max_gb_all_workers=100GB
JOB_1=20GB --> len(workers) is 0 && current_size is 0GB--> Can spawn!
JOB_2=90GB --> len(workers) is 1 && current_size is 20GB --> Not allowed!
JOB_3=20GB --> len(workers) is 1 && current_size is 20GB --> Can spawn!
JOB_4=20GB --> len(workers) is 2 && current_size is 40GB --> Can spawn!

If we start these 4 jobs. Job_2 won't run until the others are done. (exect if it get picked up first, then all the other jobs will wait.)

[DanielElisenberg]

I really like the state manager class 👍🏻 Easy to read solution to our issue. Good stuff 💯

[DanielElisenberg]

job would be none if it is completed here. I think we should refactor this solution to iterate through each dead worker and use the dead worker first and foremost. Something akin to :

def clean_up_after_dead_workers(
    dead_workers: List[Worker], manager_state
) -> None:
    if len(dead_workers) == 0:
        return
    for dead_worker in dead_workers:
            # only query for the job in question
            # the amount of requests here are maximum 4, so of no consequence
            job = job_service.get_job(dead_worker.job_id)
            if job and job.status not in ["queued", "built"]:
                logger.info(f"Worker died and did not finish job {job.job_id}")
                fix_interrupted_job(job)
            manager_state.unregister_job(dead_worker.job_id)

[DanielElisenberg]

If we do it like this the local storage module itself might as well raise the error so we don't need to do the check here.

Another option would be to fail the job:

        job_service.update_job_status(
            job_id, "failed", log=f"No such dataset available for import"
        )

This is preferable to crashing the whole executor if one dataset should be tampered with in some unexpected way.

[pawbu]

Nice to refactor this, but I think the can_spawn_new_worker should be purely "get" method without extra "setting" things, so refactoring this out will improve readability

[DanielElisenberg]

Just food for thought: Could it be beneficial to also keep the active workers list inside the ManagerState? That is also part of the statefulness of the manager process so it might be more readable for as much of it as possible to exist in this new state class?

[sonarqubecloud]

Quality Gate passed

Issues
1 New issue
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[DanielElisenberg]

Let's send it to integration tests and QA 💯