[ManagedJobs] Autoscaling worker pool for job submission - #6260
Conversation
|
/smoke-test --serve |
|
/smoke-test --serve |
|
/smoke-test --serve |
|
/smoke-test --serve |
|
/smoke-test --serve |
|
/quicktest-core |
|
/quicktest-core |
|
/quicktest-core |
|
/quicktest-core |
| handle=local_handle, | ||
| all_file_mounts=controller_task.file_mounts, | ||
| storage_mounts=controller_task.storage_mounts) | ||
| with sky_logging.silent(): |
There was a problem hiding this comment.
To suppress the (mostly) meaningless file mount logs. Those file mount are the one we internally setup in jobs controller (local user yaml -> remote (jobs controller) user yaml). I dont think exposing it to user helps
| path = pathlib.Path(serve_constants.SKYSERVE_METADATA_DIR) | ||
| if pool is not None: | ||
| path = path / pool | ||
| path = path / 'pm.lock' |
There was a problem hiding this comment.
Would still like to rename this file if possible.
There was a problem hiding this comment.
Good point! Renamed to pool.lock
|
|
||
|
|
||
| def _get_pool_filelock_path(pool: Optional[str]) -> str: | ||
| path = pathlib.Path(serve_constants.SKYSERVE_METADATA_DIR) |
There was a problem hiding this comment.
Should we just set this to something for jobs? Weird that it's using the skyserve metadata dir but it's in sky/jobs
There was a problem hiding this comment.
The problem is that we want to cleanup the lock for a specific pool after termination of a pool. Currently it is put in ~/.sky/serve/<service-name>, which will be cleanup after the service is down. If we put this lock in jobs path, we need to cleanup corresponding jobs path in serve code, which is also very weird.
Edit: I moved the code (and filelock) to serve_utils, cuz we are already using managed_jobs_state in the get_next_cluster_name function (which is required to track resources), hopefully this is less weird. We need somewhere to call both serve DB and jobs DB anyway..
| def apply( | ||
| task: 'sky.Task', | ||
| service_name: str, | ||
| mode: serve_utils.UpdateMode = serve_utils.DEFAULT_UPDATE_MODE, | ||
| pool: bool = False, | ||
| ) -> None: | ||
| """Applies the config to the service or pool.""" | ||
| try: | ||
| handle = backend_utils.is_controller_accessible( | ||
| controller=controller_utils.Controllers.SKY_SERVE_CONTROLLER, | ||
| stopped_message='') | ||
| backend = backend_utils.get_backend_from_handle(handle) | ||
| assert isinstance(backend, backends.CloudVmRayBackend) | ||
| service_record = _get_service_record(service_name, pool, handle, | ||
| backend) | ||
| if service_record is not None: | ||
| return update(task, service_name, mode, pool) | ||
| except exceptions.ClusterNotUpError: | ||
| pass | ||
| up(task, service_name, pool) |
There was a problem hiding this comment.
is a race condition possible here btw?
There was a problem hiding this comment.
Good point! Added a filelock per service name.
|
/quicktest-core |
|
/smoke-test --managed-jobs |
|
/smoke-test |
f56aece to
01aa96c
Compare
2 participants
TODO:
sky jobs pool up/sky jobs create-pool)--batch-size/--jobs-replica/ ...Tested (run the relevant ones):
bash format.sh/smoke-test(CI) orpytest tests/test_smoke.py(local)/smoke-test -k test_name(CI) orpytest tests/test_smoke.py::test_name(local)/quicktest-core(CI) orpytest tests/smoke_tests/test_backward_compat.py(local)