Repository: skypilot-org/skypilot
Pull request: [Spot] Show FAILED_CONTROLLER when controller exit abnormally
Fixes #1141

Previously, if the controller failed abnormally, the spot status of the job will only be updated if the user run `sky spot cancel <job_id>`. 

Now we move the status update to the skylet, so that the spot status will be updated automatically.

Tested:
- [x] `sky spot launch -n test-status 'echo hi; sleep 100000'`; `ssh sky-spot-controller-<hash>`; `ray job stop --address http://127.0.0.1:8265 <job_id>-ubuntu`; Check the `sky spot status` and the `~/.sky/skylet.log` after a while.

Problem statement:
[Spot] Populate the FAILED job status to the spot job table
When the spot controller failed due to some unexpected error, though the job status in `jobs.db` will be set to FAILED by skylet, but the status is not populated to the spot job table.

Hints from issue discussion:

Commit message: add spot status update to skylet

Patch to review:
diff --git a/sky/skylet/events.py b/sky/skylet/events.py
index 5d4cd79643c..7f5fbaba17a 100644
--- a/sky/skylet/events.py
+++ b/sky/skylet/events.py
@@ -13,6 +13,7 @@
 from sky import sky_logging
 from sky.backends import backend_utils, cloud_vm_ray_backend
 from sky.skylet import autostop_lib, job_lib
+from sky.spot import spot_utils
 from sky.utils import common_utils
 
 # Seconds of sleep between the processing of skylet events.
@@ -67,6 +68,14 @@ def _run(self):
                               submitted_gap_sec=self._SUBMITTED_GAP_SECONDS)
 
 
+class SpotJobUpdateEvent(SkyletEvent):
+    """Skylet event for updating spot job status."""
+    EVENT_INTERVAL_SECONDS = 300
+
+    def _run(self):
+        spot_utils.update_spot_job_status()
+
+
 class AutostopEvent(SkyletEvent):
     """Skylet event for autostop."""
     EVENT_INTERVAL_SECONDS = 60
diff --git a/sky/skylet/skylet.py b/sky/skylet/skylet.py
index a9c6894b398..30264f368fe 100644
--- a/sky/skylet/skylet.py
+++ b/sky/skylet/skylet.py
@@ -11,6 +11,7 @@
 EVENTS = [
     events.JobUpdateEvent(),
     events.AutostopEvent(),
+    events.SpotJobUpdateEvent(),
 ]
 
 while True:
diff --git a/sky/spot/spot_utils.py b/sky/spot/spot_utils.py
index ec9b6fff8c7..26ede6eac44 100644
--- a/sky/spot/spot_utils.py
+++ b/sky/spot/spot_utils.py
@@ -68,6 +68,41 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     return status
 
 
+def update_spot_job_status(job_id: Optional[int] = None):
+    """Update spot job status if the controller failed abnormally.
+
+    Check the status of the controller. If it is not running, it must be
+    exited abnormally, and we should set the job status to FAILED.
+    """
+    if job_id is None:
+        job_ids = spot_state.get_nonterminal_job_ids_by_name(None)
+    else:
+        job_ids = [job_id]
+    for job in job_ids:
+        controller_status = job_lib.get_status(job)
+        if controller_status.is_terminal():
+            logger.error(f'Controller for job {job} have exited abnormally. '
+                         'Setting the job status to FAILED_CONTROLLER.')
+            task_name = spot_state.get_task_name_by_job_id(job)
+
+            # Tear down the abnormal spot cluster to avoid resource leakage.
+            cluster_name = generate_spot_cluster_name(task_name, job)
+            handle = global_user_state.get_handle_from_cluster_name(
+                cluster_name)
+            if handle is not None:
+                backend = backend_utils.get_backend_from_handle(handle)
+                try:
+                    backend.teardown(handle, terminate=True)
+                except RuntimeError:
+                    logger.error('Failed to tear down the spot cluster '
+                                 f'{cluster_name!r}.')
+
+            # The controller process for this job is not running: it must
+            # have exited abnormally, and we should set the job status to
+            # FAILED_CONTROLLER.
+            spot_state.set_failed(job, spot_state.SpotStatus.FAILED_CONTROLLER)
+
+
 def get_job_timestamp(backend: 'backends.CloudVmRayBackend', cluster_name: str,
                       get_end_time: bool) -> float:
     """Get the started/ended time of the job."""
@@ -80,6 +115,8 @@ def get_job_timestamp(backend: 'backends.CloudVmRayBackend', cluster_name: str,
     subprocess_utils.handle_returncode(returncode, code,
                                        'Failed to get job time.',
                                        stdout + stderr)
+    if 'None' in stdout:
+        return -1
     return float(stdout)
 
 
@@ -112,34 +149,7 @@ def cancel_jobs_by_id(job_ids: Optional[List[int]]) -> str:
                         f'{job_status.value}. Skipped.')
             continue
 
-        # Check the status of the controller. If it is not running, it must be
-        # exited abnormally, and we should set the job status to FAILED.
-        # TODO(zhwu): instead of having the liveness check here, we may need
-        # to make it as a event in skylet.
-        controller_status = job_lib.get_status(job_id)
-        if controller_status.is_terminal():
-            logger.error(f'Controller for job {job_id} have exited abnormally. '
-                         'Setting the job status to FAILED_CONTROLLER.')
-            task_name = spot_state.get_task_name_by_job_id(job_id)
-
-            # Tear down the abnormal spot cluster to avoid resource leakage.
-            cluster_name = generate_spot_cluster_name(task_name, job_id)
-            handle = global_user_state.get_handle_from_cluster_name(
-                cluster_name)
-            if handle is not None:
-                backend = backend_utils.get_backend_from_handle(handle)
-                try:
-                    backend.teardown(handle, terminate=True)
-                except RuntimeError:
-                    logger.error('Failed to tear down the spot cluster '
-                                 f'{cluster_name!r}.')
-
-            # The controller process for this job is not running: it must
-            # have exited abnormally, and we should set the job status to
-            # FAILED_CONTROLLER.
-            spot_state.set_failed(job_id,
-                                  spot_state.SpotStatus.FAILED_CONTROLLER)
-            continue
+        update_spot_job_status(job_id)
 
         # Send the signal to the spot job controller.
         signal_file = pathlib.Path(SIGNAL_FILE_PREFIX.format(job_id))
@@ -274,11 +284,13 @@ def dump_spot_job_queue() -> str:
         if job['status'] == spot_state.SpotStatus.RECOVERING:
             # When job is recovering, the duration is exact job['job_duration']
             job_duration = job['job_duration']
-        elif job_start_at > 0:
+        elif job_start_at > 0 and end_at > 0:
             job_duration = end_at - job_start_at
         else:
             # When job_start_at <= 0, that means the last_recovered_at is not
             # set yet, i.e. the job is not started.
+            # When end_at <= 0, that means there is some problem fetching the
+            # end_at time, so we use 0.
             job_duration = 0
         job['job_duration'] = job_duration
         job['status'] = job['status'].value


Code review comments:
diff_hunk: @@ -68,6 +68,41 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     return status
 
 
+def update_spot_job_status(job_id: Optional[int] = None):
+    """Update spot job status if the controller failed abnormally.
+
+    Check the status of the controller. If it is not running, it must be
path: sky/spot/spot_utils.py
text: @user1:
nit - be -> have

@user1:
nit: FAILED -> FAILED_CONTROLLER

@author:
Done. Thanks!

diff_hunk: @@ -68,6 +68,41 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     return status
 
 
+def update_spot_job_status(job_id: Optional[int] = None):
+    """Update spot job status if the controller failed abnormally.
+
+    Check the status of the controller. If it is not running, it must be
+    exited abnormally, and we should set the job status to FAILED.
+    """
+    if job_id is None:
+        job_ids = spot_state.get_nonterminal_job_ids_by_name(None)
+    else:
+        job_ids = [job_id]
+    for job in job_ids:
+        controller_status = job_lib.get_status(job)
+        if controller_status.is_terminal():
+            logger.error(f'Controller for job {job} have exited abnormally. '
+                         'Setting the job status to FAILED_CONTROLLER.')
+            task_name = spot_state.get_task_name_by_job_id(job)
+
+            # Tear down the abnormal spot cluster to avoid resource leakage.
+            cluster_name = generate_spot_cluster_name(task_name, job)
+            handle = global_user_state.get_handle_from_cluster_name(
+                cluster_name)
+            if handle is not None:
+                backend = backend_utils.get_backend_from_handle(handle)
+                try:
+                    backend.teardown(handle, terminate=True)
+                except RuntimeError:
+                    logger.error('Failed to tear down the spot cluster '
path: sky/spot/spot_utils.py
text: @user1:
Should we avoid setting L103 if we hit this termination error? So it's easier to spot potential leakage.

@author:
I feel like the FAILED_CONTROLLER should be a more strong alert sign for the user to check what is going on for the job than the non-changed nonterminal status, such as RUNNING. An alternative can be having a new spot job status, e.g. FAILED_TERMINATION.

@user1:
My understanding is 
```
  - FAILED_CONTROLLER: The job failed due to an unexpected error in the spot
      controller.
```
suggests the spot job failed, but it could've leaked if termination fails. Should we treat non-terminal statuses = cluster potentially alive, and terminal = cluster definitely down?

diff_hunk: @@ -68,6 +68,41 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     return status
 
 
+def update_spot_job_status(job_id: Optional[int] = None):
+    """Update spot job status if the controller failed abnormally.
+
+    Check the status of the controller. If it is not running, it must be
+    exited abnormally, and we should set the job status to FAILED.
+    """
+    if job_id is None:
+        job_ids = spot_state.get_nonterminal_job_ids_by_name(None)
+    else:
+        job_ids = [job_id]
+    for job in job_ids:
+        controller_status = job_lib.get_status(job)
+        if controller_status.is_terminal():
+            logger.error(f'Controller for job {job} have exited abnormally. '
+                         'Setting the job status to FAILED_CONTROLLER.')
+            task_name = spot_state.get_task_name_by_job_id(job)
+
+            # Tear down the abnormal spot cluster to avoid resource leakage.
+            cluster_name = generate_spot_cluster_name(task_name, job)
+            handle = global_user_state.get_handle_from_cluster_name(
+                cluster_name)
+            if handle is not None:
+                backend = backend_utils.get_backend_from_handle(handle)
+                try:
+                    backend.teardown(handle, terminate=True)
path: sky/spot/spot_utils.py
text: @user1:
Should we add retry (or something to be refactored out with the normal spot cluster termination)? Ok to do it later too.

@author:
Good point! Added a retry. Will try to refactor it later. :)

diff_hunk: @@ -68,6 +68,41 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     return status
 
 
+def update_spot_job_status(job_id: Optional[int] = None):
+    """Update spot job status if the controller failed abnormally.
+
+    Check the status of the controller. If it is not running, it must be
+    exited abnormally, and we should set the job status to FAILED.
+    """
+    if job_id is None:
+        job_ids = spot_state.get_nonterminal_job_ids_by_name(None)
+    else:
+        job_ids = [job_id]
+    for job in job_ids:
+        controller_status = job_lib.get_status(job)
+        if controller_status.is_terminal():
+            logger.error(f'Controller for job {job} have exited abnormally. '
path: sky/spot/spot_utils.py
text: @user1:
nit: have -> has

@author:
Done. Thanks!

diff_hunk: @@ -68,6 +68,41 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     return status
 
 
+def update_spot_job_status(job_id: Optional[int] = None):
+    """Update spot job status if the controller failed abnormally.
+
+    Check the status of the controller. If it is not running, it must be
+    exited abnormally, and we should set the job status to FAILED.
+    """
+    if job_id is None:
+        job_ids = spot_state.get_nonterminal_job_ids_by_name(None)
+    else:
+        job_ids = [job_id]
+    for job in job_ids:
path: sky/spot/spot_utils.py
text: @user1:
`job_id_`?

@author:
Done. Thanks!

diff_hunk: @@ -11,6 +11,7 @@
 EVENTS = [
     events.JobUpdateEvent(),
     events.AutostopEvent(),
+    events.SpotJobUpdateEvent(),
path: sky/skylet/skylet.py
text: @user1:
nit: order these?

@author:
Done. And added a comment for the order of SpotJobUpdateEvent vs JobUpdateEvent.

diff_hunk: @@ -68,6 +68,41 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     return status
 
 
+def update_spot_job_status(job_id: Optional[int] = None):
+    """Update spot job status if the controller failed abnormally.
path: sky/spot/spot_utils.py
text: @user1:
nit: if the controller [process]

We should distinguish controller (VM) from controller processes when possible. Same on L74?

@author:
Good point! Changed to controller job.