[jobs] revamp scheduling for managed jobs

[cg505]

Detaches the job controller from ray worker and the ray driver program, and uses our own scheduling and parallelism control mechanism.

See the commands in sky/jobs/scheduler.py for more info.

Tested (run the relevant ones):

• Code formatting: bash format.sh
• Any manual or new tests for this PR (please specify below)
• All smoke tests: pytest tests/test_smoke.py
• Relevant individual smoke tests: pytest tests/test_smoke.py::test_fill_in_the_name
• Backward compatibility tests: conda deactivate; bash -i tests/backward_compatibility_tests.sh

[cg505]

if scheduler is killed before this line (e.g. when running as part of a controller job), we will get stuck since the job will be submitted but the controller will never start. Todo figure out how to recover from this case

[Michaelvll]

We can have a skylet event to monitor managed job table, like we do for normal unmanaged jobs.

[cg505]

We are already using the exiting managed job skylet event for that, but the problem is that if it dies right here, there's no way to know if the scheduler is just about to start the process or if it already died. We need a way to check if the scheduler died or maybe a timestamp for the WAITING -> LAUNCHING transition.

[Michaelvll]

Thanks @cg505 for making this significant change! This is awesome! I glanced the code, and it mostly looks good. The main concern is the complexity and granularity we have for limiting the number of launches. Please see the comments below.

[Michaelvll]

Can we store the logs in either ~/sky_logs/jobs_controller or ~/.sky?

[Michaelvll]

Do we need to redirect the logging?

[cg505]

These were debug lines I left in accidentally. I am now using logger.debug. I guess that's probably fine?

[Michaelvll]

We can have a skylet event to monitor managed job table, like we do for normal unmanaged jobs.

[Michaelvll]

How about calling it schedule_new_job()?

[cg505]

Changed to maybe_start_waiting_jobs(). Open to other names - it might be a bit of a misnomer if it schedules an existing job that wants to launch something.

[Michaelvll]

Suggested change

	"""Determine if any jobs can be launched, and if so, launch them.
	"""Determine if any jobs can be scheduled, and if so, schedule them.

[Michaelvll]

I found this function may overcomplicated the problem here. For launches, we do not need to preserve a FIFO order.
Instead, we can use the schedule_step above to preserve the order for the order of the jobs to be scheduled, while using a semaphore to limit the total number of launches.
Reason: I think we should limit the actual launches in a finer granularity:

skypilot/sky/jobs/recovery_strategy.py

Lines 311 to 321 in 83b2325

	sky.launch(
	self.dag,
	cluster_name=self.cluster_name,
	# We expect to tear down the cluster as soon as the job is
	# finished. However, in case the controller dies, set
	# autodown to try and avoid a resource leak.
	idle_minutes_to_autostop=_AUTODOWN_MINUTES,
	down=True,
	detach_setup=True,
	detach_run=True,
	_is_launched_by_jobs_controller=True)

Otherwise, when there is bad resource capacity, and the job went into an hour-long waiting loop, it will block other jobs from being able to launch their resources.

[Michaelvll]

With this the schedule_step only needs to check the threshold for the total number of jobs that can run in parallel, but don't need to check the total launches.

[cg505]

Updated how this works significantly. New version will transition to ALIVE (does not count towards launching limit) while in backoff.

[cg505]

/quicktest-core