6 scheduling #36
6 scheduling #36
Conversation
There was a problem hiding this comment.
Ah, nice you were able to come up with something so quickly already!
I looked at it from a high level, and noticed that this is currently implemented for Kubernetes only (that makes sense), and also setup in such a way that it needs refactoring for Docker. I would think of the scheduler as something that could work for both Docker and Kubernetes, especially the scheduling decisions. Also, there is now k8s-specific code in the main file (e.g. the import), and the kubernetes scheduler, this makes the code somewhat spaghetti: there are specific implementation-specific classes where responsibility is meant to be delegated. If you need to access the scheduler in the main file, use a generic scheduler, and make the docker-based parts not implemented. I think that would give a much cleaner design.
Also, I would consider making the launcher responsible for scheduling. And then have the scheduler talk to the launcher to actually start jobs.
I'm not yet sure if we should allow running without the scheduler, or if it would always be active.
|
Hope my feedback was at an angle that helps you at this stage. In any case, well done, keep it going! p.s. the CI error looks like it could be cause by Kubernetes-specific things having entered into the main api code, which wouldn't work when running with Docker. |
|
Working on Docker implementation to be added to this PR |
scrapyd_k8s/k8s_scheduler.py
Outdated
| @@ -0,0 +1,96 @@ | |||
| import logging | |||
There was a problem hiding this comment.
Can we fit this in the directory structure? I wouldn't expect this in the src root.
There was a problem hiding this comment.
For me, this functionality seems related to the kubernetes launcher.
There was a problem hiding this comment.
k8s_scheduler is not the part of launcher because it is 1) a subscriber and should have this functionality to subscribe to the observer, 2) be the part which is initialized optionally when a user wants to limit the number of parallel job (launcher is not optional), 3) contains higher-level logic that uses low-level methods and helper methods from the launcher.
If you don't like that this file is located in the root, I can relocate it to the directory that belongs to this feature, say, limit_jobs or something like this, you can pick any name you like and I will add it.
scrapyd_k8s/launcher/k8s.py
Outdated
| if not jobs.items: | ||
| logger.error(f"No job found with job_id={job_id}") | ||
| return None | ||
| return jobs.items[0].metadata.name |
There was a problem hiding this comment.
when you're listing jobs, would you also get the name already?
|
I have problems because I separated this PR partially and now I have a multiverse which I need to refactor to the only source of truth. Going to spend some uncertain amount of time on that. |
|
The way I would do this:
Of this is much work in many commits, you may consider first doing an interactive rebase of this PR, to simplify it, and reduce the number of commits (that each may need amending). Yes, this is a bit of work, but something I come across now and then, in various projects. |
|
Thank you for the advice! No worries, this is me who messed up merging, complexity is part of the job:D Learning to make more granular commits and cleaner PRs the hard way:D |
vlerkin
force-pushed
the
6-scheduling
branch
2 times, most recently
from
c8b35ad to
6394633
Compare
|
I modified one of the methods in the scheduler (get_next_suspended_job_id) to handle cases if a job does not have a creation_timestamp. It is not expected but if someone used a custom resource and forgot to add this field or made any other error, the job will get the timestamp assigned and will be processed like other jobs in the queue. Also, there are now unit tests that cover different scenarios for the scheduler. If you have any other comments for improvements, let me know! |
|
Could you please resolve conflicts on |
There was a problem hiding this comment.
At first glance, well done! Seems to do the job (though haven't tested it).
An integration test would increase my confidence that it performs well (now I'd have to run it locally to see if it actually works - I think it would, but still I would feel necessitated to do so).
Also, I think the scheduling logic is now implemented in K8s and Docker separately. Would it make sense to have a single piece of code decide when to schedule, and let e.g. the launcher and listener be the interface to K8s/Docker? Haven't thought this fully through, but the question comes up.
Some questions and notes remain, otherwise it's well on the way.
| # Number of attempts to reconnect with k8s API to watch events, default is 5 | ||
| reconnection_attempts = 5 | ||
|
|
||
| # Minimum time in seconds to wait before reconnecting to k8s API to watch events, default is 5 | ||
| backoff_time = 5 | ||
|
|
||
| # Coefficient that is multiplied by backoff_time to provide exponential backoff to prevent k8s API from being overwhelmed | ||
| # default is 2, every reconnection attempt will take backoff_time*backoff_coefficient | ||
| backoff_coefficient = 2 |
There was a problem hiding this comment.
Do we think the average user needs to configure this? If yes, keep. If no, the documentation would be enough.
Good to remove max_proc.
There was a problem hiding this comment.
max_proc is not present by default, yes, so all scheduled jobs will run immediately. But when provided, then scheduler is activated.
| @@ -44,7 +44,7 @@ def test_listversions_ok(): | |||
| assert_response_ok(response) | |||
|
|
|||
| json = response.json() | |||
| assert json['versions'] == AVAIL_VERSIONS | |||
| assert set(AVAIL_VERSIONS).issubset(set(json['versions'])) | |||
There was a problem hiding this comment.
That is useful. But it does not let us test the precise response anymore, and the order is not tested anymore.
I think I'd rather keep the old code, and make sure we set AVAIL_VERSIONS better if it would fail?
There was a problem hiding this comment.
Since tests modify the state, the idempotency is violated, so it's hard to test the exact response and the test was failing because the expected response does not match reality any more.
There was a problem hiding this comment.
Does the test modify the available versions? I don't really see that.
I think it talks to the image repository, and obtains the versions. One solution would be to get the real versions in the script invoking the test job, another to implement our own test image repository. Since this is not really related to this feature, I suggest we update the hardcoded versions, and fix this issue later. Or leave it failing, to fix later.
| assert job_id == 'job1' | ||
| mock_logger.warning.assert_called_with( | ||
| f"Job {job} missing 'metadata.creation_timestamp'; assigned max timestamp." | ||
| ) |
There was a problem hiding this comment.
Very nice you're testing this!
Yet ... can we think of an integration test? It's much more reliable if we can test this against Kubernetes (and Docker).
With max_proc = 0 no job should ever run.
With max_proc = 1 scheduling two jobs, where one has a sleep, we want to see the last job suspended, and started after the first job has ended.
(etc?)
I think this is not that hard either, except we need to test with different configs, something we don't really do yet.
The current test seems tightly coupled to the implementation. An integration test would allow us to refactor the code, and have the test still tell us if it works or not.
There was a problem hiding this comment.
Anyway, you this may be for later :)
There was a problem hiding this comment.
Normally unit tests that rely on a method signature are enough to confirm the way a method works, integration tests are extras, we can add a ticket to create integration tests that would test different config files.
…logic for observer in a RecourceWatcher class; added method to stop a thread gracefully
…that handles the logic to unsuspend jobs and get the next in order according to the creation timestamp; modify schedule endpoint to start jobs suspended if there is already enogh jobs running; modify corresponding function in k8s launcher; add to k8s launcher methods to unsuspend job, to get current number of running jobs, to list suspended jobs and a private method to get job name to be used for unsuspend function
…source watcher instance to enable_joblogs to subscribe to the event watcher if the log feature is configured; delete logic about event watcher from main; pass container for list objects function instead of container name; remove start methon from log handler class; modify joblogs init to subscribe to event watcher
…rs and run more from the queue of created jobs when capacity is available; add backgroung thread that sleeps for 5 sec and triggers the function that starts additional containers up to capacity; add a method to gracefully stop the background thread that might be used in the future to stop the thread when app stops; encapsulate k8s and docker related schedule functionality in corresponding launchers and keep api.py launcher agnostic; add max_proc to config for docker
…nd patch is sufficient
…nnect loop for event watcher; make number of reconnect attempts, backoff time and a coefficient for exponential growth configurable via config; add backoff_time, reconnection_attempts and backoff_coefficient as attributes to the resource watcher init; add resource_version as a param to w.stream so a failed stream can read from the last resource it was able to catch; add urllib3.exceptions.ProtocolError and handle reconnection after some exponential backoff time to avoid api flooding; add config as a param for init for resource watcher; modify config in kubernetes.yaml and k8s config to contain add backoff_time, reconnection_attempts and backoff_coefficient
…and a label selector to make the code in listjobs, get_running_jobs and list_suspended_jobs DRY; refactor listjobs to use the helper function with the existing _parse_job as a filter_func parameter
…unction because list jobs uses a different logic
…nnect loop for event watcher; make number of reconnect attempts, backoff time and a coefficient for exponential growth configurable via config; add backoff_time, reconnection_attempts and backoff_coefficient as attributes to the resource watcher init; add resource_version as a param to w.stream so a failed stream can read from the last resource it was able to catch; add urllib3.exceptions.ProtocolError and handle reconnection after some exponential backoff time to avoid api flooding; add config as a param for init for resource watcher; modify config in kubernetes.yaml and k8s config to contain add backoff_time, reconnection_attempts and backoff_coefficient
… connection to the k8s was achieved so only sequential failures detected; add exception handling to watch_pods to handle failure in urllib3, when source version is old and not available anymore, and when stream is ended; remove k8s resource watcher initialization from run function in api.py and move it to k8s.py launcher as _init_resource_watcher; refactor existing logic from joblogs/__init__.py to keep it in _init_resource_watcher and enable_joblogs in k8s launcher
…ed to re-establish connection to the Kubernetes wather
…k to the CONFIG.md in the README.md; remove variables for reconnection_attempts, backoff_time and backoff_coefficient fron the sample config since default values are provided in the code.
… a package that has an enable function in launcher/k8s.py which also part of resource watcher initialization; initialize the scheduler if max_proc was provided in the scrapyd section of the config file; refactor related methods in the launcher to use extra functionality for job number limiting only if max_proc is provided
…; remove max_proc from config file since by default we want to runn all scheduled jobs in parallel; add a section about max_proc to the CONFIG.md
… propagate the exception
…o handle and propagate the critical ones
…on_timestamps to the jobs that do not have them, so they are proccessed at the end of the queue; add unit tests for k8s_scheduler class
…for docker implementation of the scheduler
|
The launcher is already sort of an interface, since every implementation uses its own launcher. Listener cannot be an interface because event watcher is a native Kubernetes thing, if I understand your comment correctly. So for k8s we still use the watcher to start suspended jobs when at least one job is done or deleted, and for Docker we have a background thread that checks the state of existing containers, we have discussed it previously. |
What happens in the PR:
The big picture:
Event watcher connects to the k8s api and receives the stream of events, it then notifies the subscribers if a new event is received and passes it to the provided callback. The subscriber - KubernetesScheduler - receives event in a handle_pod_event method, this method reacts to the changes in job statuses, and if job completed running or failed it calls another method - check_and_unsuspend_jobs - that checks capacity and unsuspends jobs until the number of allowed parallel jobs is reached, while doing this it relies on another method - get_next_suspended_job_id - to unsuspend the most recent job, to keep the order in which jobs were initially scheduled.
When the job is scheduled, based on the number of currently active jobs and max_proc provided in the config (default is 4), the job runs or goes to the queue of suspended jobs (native k8s queue). Then events that change the number of active jobs trigger the logic of KubernetesScheduler class that unsuspend suspended jobs until the desired state (num of parallel jobs) is achieved.