Repository: skypilot-org/skypilot
Pull request: Remove tracebacks for exceptions to improve UX
Previously, the raised exceptions will print a large amount of unnecessary traceback information of the internal sky code as well `ssh` error, so that the user's file_mounts / setup / run error will be more close to the end and easier to find. This PR suppresses the errors that occurred in the following yaml. Closes #384 .
```yaml
# workdir: ~/not-exits

# file_mounts:
#  /not-exists: ./examples
#  ~/test: ~/not-exits

resources:
  cloud: aws

setup: |
  echo "running setup"
  # python -c "raise RuntimeError('setup failed')"

run: |
  echo "started"
  python -c "raise RuntimeError('run failed')"
```

The following is an output example of setup failure:
```
Task from YAML spec: ./examples/minimal.yaml
Running task on cluster min...
I 02-24 01:19:14 optimizer.py:213] Defaulting estimated time to 1 hr. Call Task.set_time_estimator() to override.
I 02-24 01:19:15 optimizer.py:322] Optimizer - plan minimizing cost (~$0.4):
I 02-24 01:19:15 optimizer.py:337] 
I 02-24 01:19:15 optimizer.py:337] TASK     BEST_RESOURCE
I 02-24 01:19:15 optimizer.py:337] minimal  AWS(m4.2xlarge)I 02-24 01:19:15 optimizer.py:337] I 02-24 01:19:15 cloud_vm_ray_backend.py:1016] Creating a new cluster: "min" [1x AWS(m4.2xlarge)].
I 02-24 01:19:15 cloud_vm_ray_backend.py:1016] Tip: to reuse an existing cluster, specify --cluster-name (-c) in the CLI or use sky.launch(.., cluster_name=..) in the Python API. Run `sky status` to see existing clusters.
I 02-24 01:19:15 cloud_vm_ray_backend.py:665] To view detailed progress: tail -n100 -f sky_logs/sky-2022-02-24-01-19-14-644420/provision.log
I 02-24 01:19:15 cloud_vm_ray_backend.py:675] I 02-24 01:19:15 cloud_vm_ray_backend.py:675] Launching on AWS us-east-1 (us-east-1a,us-east-1b,us-east-1c,us-east-1d,us-east-1e,us-east-1f)
I 02-24 01:22:30 cloud_vm_ray_backend.py:746] Successfully provisioned or found existing VM.
I 02-24 01:22:34 cloud_vm_ray_backend.py:1287] Running setup...
running setup
Traceback (most recent call last):
  File "<string>", line 1, in <module>
RuntimeError: setup failed
E 02-24 01:22:45 cloud_vm_ray_backend.py:1303] Setup failed with return code 1.
```

Tested:
- [x] `examples/run_smoke_tests.sh`
- [x] `sky cpunode --tmux`

Problem statement:
[Logging/Sky Launch] Highlight Error Message from User Setup/Run Commands
When Sky is provisioning machine and running the user's `setup/run` commands in the yaml file, it is difficult for initial users to find the error message in their setup or run script amidst all the logs. Hence, it is a good idea to highlight the errors.

Hints from issue discussion:

Commit message: Fix #442

Patch to review:
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 1d914a1c800..27e67dc9b2a 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -593,9 +593,9 @@ def run_command_on_ip_via_ssh(
     ssh_private_key: str,
     port_forward: Optional[List[int]] = None,
     # Advanced options.
+    require_outputs: bool = False,
     log_path: str = '/dev/null',
     stream_logs: bool = True,
-    check: bool = False,
     ssh_mode: SshMode = SshMode.NON_INTERACTIVE,
     ssh_control_name: Optional[str] = None,
 ) -> Tuple[subprocess.Popen, str, str]:
@@ -631,7 +631,7 @@ def run_command_on_ip_via_ssh(
     if ssh_mode == SshMode.LOGIN:
         assert isinstance(cmd, list), 'cmd must be a list for login mode.'
         command = base_ssh_command + cmd
-        proc = run(command, shell=False, check=check)
+        proc = run(command, shell=False)
         return proc, '', ''
     if isinstance(cmd, list):
         cmd = ' '.join(cmd)
@@ -647,7 +647,10 @@ def run_command_on_ip_via_ssh(
         shlex.quote(f'true && source ~/.bashrc && export OMP_NUM_THREADS=1 '
                     f'PYTHONWARNINGS=ignore && ({cmd})'),
     ]
-    return log_lib.run_with_log(command, log_path, stream_logs, check=check)
+    return log_lib.run_with_log(command,
+                                log_path,
+                                stream_logs,
+                                require_outputs=require_outputs)
 
 
 def run(cmd, **kwargs):
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 26a7f0938d2..2a0019021b2 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -245,7 +245,7 @@ def check_ip():
                                      placement_group_bundle_index=i).remote()
                     for i in range(pg.bundle_count)
                 ])
-                print('SKY INFO: Placement group IPs:', ip_list)
+                print('SKY INFO: Reserved IPs:', ip_list)
                 ip_list_str = '\\n'.join(ip_list)
                 export_sky_env_vars = 'export SKY_NODE_IPS="' + ip_list_str + '"\\n'
                 """),
@@ -343,12 +343,17 @@ def add_epilogue(self) -> None:
 
         self._code += [
             textwrap.dedent(f"""\
-            try:
-                ray.get(futures)
-                job_lib.set_status({self.job_id!r}, job_lib.JobStatus.SUCCEEDED)
-            except ray.exceptions.WorkerCrashedError as e:
+            returncodes = ray.get(futures)
+            if sum(returncodes) != 0:    
                 job_lib.set_status({self.job_id!r}, job_lib.JobStatus.FAILED)
-                raise RuntimeError('Command failed, please check the logs.') from e
+                time.sleep(1)
+                print('SKY INFO: {colorama.Fore.RED}Job {self.job_id} failed with '
+                      'return code list:{colorama.Style.RESET_ALL}',
+                      returncodes,
+                      file=sys.stderr,
+                      flush=True)
+            else:
+                job_lib.set_status({self.job_id!r}, job_lib.JobStatus.SUCCEEDED)
             """)
         ]
 
@@ -701,11 +706,11 @@ def _retry_region_zones(self, to_provision: Resources, num_nodes: int,
                                                     cluster_handle=handle,
                                                     ready=False)
 
-            gang_failed, proc, stdout, stderr = self._gang_schedule_ray_up(
+            gang_failed, rc, stdout, stderr = self._gang_schedule_ray_up(
                 to_provision.cloud, num_nodes, cluster_config_file,
                 log_abs_path, stream_logs)
 
-            if gang_failed or proc.returncode != 0:
+            if gang_failed or rc != 0:
                 if gang_failed:
                     # There exist partial nodes (e.g., head node) so we must
                     # down before moving on to other regions.
@@ -773,7 +778,7 @@ def ray_up(start_streaming_at):
             # different order from directly running in the console. The
             # `--log-style` and `--log-color` flags do not work. To reproduce,
             # `ray up --log-style pretty --log-color true | tee tmp.out`.
-            proc, stdout, stderr = log_lib.run_with_log(
+            returncode, stdout, stderr = log_lib.run_with_log(
                 # NOTE: --no-restart solves the following bug.  Without it, if
                 # 'ray up' (sky launch) twice on a cluster with >1 node, the
                 # worker node gets disconnected/killed by ray autoscaler; the
@@ -789,8 +794,9 @@ def ray_up(start_streaming_at):
                 start_streaming_at=start_streaming_at,
                 # Reduce BOTO_MAX_RETRIES from 12 to 5 to avoid long hanging
                 # time during 'ray up' if insufficient capacity occurs.
-                env=dict(os.environ, BOTO_MAX_RETRIES='5'))
-            return proc, stdout, stderr
+                env=dict(os.environ, BOTO_MAX_RETRIES='5'),
+                require_outputs=True)
+            return returncode, stdout, stderr
 
         config = backend_utils.read_yaml(cluster_config_file)
         file_mounts = config['file_mounts']
@@ -801,12 +807,12 @@ def ray_up(start_streaming_at):
 
         with console.status('[bold cyan]Starting cluster...'):
             # ray up.
-            proc, stdout, stderr = ray_up(
+            returncode, stdout, stderr = ray_up(
                 start_streaming_at='Shared connection to')
 
         # Only 1 node or head node provisioning failure.
-        if num_nodes == 1 or proc.returncode != 0:
-            return False, proc, stdout, stderr
+        if num_nodes == 1 or returncode != 0:
+            return False, returncode, stdout, stderr
 
         logger.info(f'{style.BRIGHT}Successfully provisioned or found'
                     f' existing head VM. Waiting for workers.{style.RESET_ALL}')
@@ -821,9 +827,9 @@ def ray_up(start_streaming_at):
         gang_failed = not cluster_ready
         if gang_failed:
             # Head OK; gang scheduling failure.
-            return gang_failed, proc, stdout, stderr
+            return gang_failed, returncode, stdout, stderr
 
-        return False, proc, stdout, stderr
+        return False, returncode, stdout, stderr
 
     def _ensure_cluster_ray_started(self,
                                     handle: 'CloudVmRayBackend.ResourceHandle',
@@ -839,14 +845,14 @@ def _ensure_cluster_ray_started(self,
             #   - ray up --restart-only
             return
         backend = CloudVmRayBackend()
-        proc, _, _ = backend.run_on_head(
+        returncode = backend.run_on_head(
             handle,
             'ray status',
             # At this state, an erroneous cluster may not have cached
             # handle.head_ip (global_user_state.add_or_update_cluster(...,
             # ready=True)).
             use_cached_head_ip=False)
-        if proc.returncode == 0:
+        if returncode == 0:
             return
         backend.run_on_head(handle, 'ray stop', use_cached_head_ip=False)
         log_lib.run_with_log(
@@ -1026,12 +1032,11 @@ def _set_tpu_name(self, cluster_config_file: str, num_nodes: int,
         for ip in ip_list:
             cmd = (f'[[ -z $TPU_NAME ]] && echo "export TPU_NAME={tpu_name}" '
                    '>> ~/.bashrc || echo "TPU_NAME already set"')
-            backend_utils.run_command_on_ip_via_ssh(
-                ip,
-                cmd,
-                ssh_user=ssh_user,
-                ssh_private_key=ssh_private_key,
-                check=True)
+            returncode = backend_utils.run_command_on_ip_via_ssh(
+                ip, cmd, ssh_user=ssh_user, ssh_private_key=ssh_private_key)
+            if returncode != 0:
+                raise RuntimeError(f'{colorama.Fore.RED}Failed to set TPU_NAME '
+                                   f'on node.{colorama.Style.RESET_ALL}')
 
     def provision(self,
                   task: Task,
@@ -1062,13 +1067,15 @@ def provision(self,
             # Do not remove the stopped cluster from the global state if failed
             # to start.
             if e.no_retry:
-                raise e
+                logger.error(e)
+                sys.exit(1)
             # Clean up the cluster's entry in `sky status`.
             global_user_state.remove_cluster(cluster_name, terminate=True)
-            raise exceptions.ResourcesUnavailableError(
+            logger.error(
                 'Failed to provision all possible launchable resources. '
                 f'Relax the task\'s resource requirements:\n {launched_nodes}x '
-                f'{task.resources}') from e
+                f'{task.resources}')
+            sys.exit(1)
         if dryrun:
             return
         cluster_config_file = config_dict['ray']
@@ -1116,7 +1123,7 @@ def sync_workdir(self, handle: ResourceHandle, workdir: Path) -> None:
                     f'Symlink contents are not uploaded.{style.RESET_ALL}')
             self._rsync_up(handle,
                            ip=ip,
-                           source=f'{workdir}',
+                           source=f'{workdir}/',
                            target=SKY_REMOTE_WORKDIR,
                            with_outputs=True)
 
@@ -1161,14 +1168,18 @@ def sync_to_all_nodes(src: str,
                         f'Symlink contents are not uploaded.{style.RESET_ALL}')
 
                 if command is not None:
-                    backend_utils.run_command_on_ip_via_ssh(
+                    returncode = backend_utils.run_command_on_ip_via_ssh(
                         ip,
                         command,
                         ssh_user=ssh_user,
                         ssh_private_key=ssh_private_key,
                         log_path=os.path.join(self.log_dir, 'file_mounts.log'),
-                        check=True,
                         ssh_control_name=self._ssh_control_name(handle))
+                    if returncode != 0:
+                        logger.error(f'{fore.RED}Failed to run command:\n'
+                                     f'{style.RESET_ALL} {command}')
+                        sys.exit(returncode)
+
                 if run_rsync:
                     # TODO(zhwu): Logging to 'file_mounts.log'
                     # TODO(zhwu): Optimize for large amount of files.
@@ -1238,14 +1249,16 @@ def sync_to_all_nodes(src: str,
         # (2) Run the commands to create symlinks on all the nodes.
         symlink_command = ' && '.join(symlink_commands)
         for ip in ip_list:
-            backend_utils.run_command_on_ip_via_ssh(
+            returncode = backend_utils.run_command_on_ip_via_ssh(
                 ip,
                 symlink_command,
                 ssh_user=ssh_user,
                 ssh_private_key=ssh_private_key,
                 log_path=os.path.join(self.log_dir, 'file_mounts.log'),
-                check=True,
                 ssh_control_name=self._ssh_control_name(handle))
+            if returncode != 0:
+                raise RuntimeError(f'{fore.RED}Failed create symlink for: {ip}'
+                                   f'{style.RESET_ALL}')
 
     def setup(self, handle: ResourceHandle, task: Task) -> None:
         style = colorama.Style
@@ -1278,29 +1291,30 @@ def setup(self, handle: ResourceHandle, task: Task) -> None:
                                source=setup_sh_path,
                                target=f'/tmp/{setup_file}',
                                with_outputs=False)
-                try:
-                    # Need this `-i` option to make sure `source ~/.bashrc` work
-                    backend_utils.run_command_on_ip_via_ssh(
-                        ip,
-                        f'/bin/bash -i /tmp/{setup_file}',
-                        ssh_user=ssh_user,
-                        ssh_private_key=ssh_private_key,
-                        log_path=os.path.join(self.log_dir, 'setup.log'),
-                        check=True,
-                        ssh_control_name=self._ssh_control_name(handle))
-                except subprocess.CalledProcessError as e:
+                # Need this `-i` option to make sure `source ~/.bashrc` work
+                returncode = backend_utils.run_command_on_ip_via_ssh(
+                    ip,
+                    f'/bin/bash -i /tmp/{setup_file}',
+                    ssh_user=ssh_user,
+                    ssh_private_key=ssh_private_key,
+                    log_path=os.path.join(self.log_dir, 'setup.log'),
+                    ssh_control_name=self._ssh_control_name(handle))
+                if returncode != 0:
                     logger.error(f'{fore.RED}Setup failed with return code'
-                                 f' {e.returncode}.{style.RESET_ALL}')
+                                 f' {returncode}.{style.RESET_ALL}')
                     # Suppress the error traceback. Fail as soon as
                     # possible (head node).
-                    sys.exit(e.returncode)
+                    sys.exit(returncode)
         logger.info(f'{fore.GREEN}Setup completed.{style.RESET_ALL}')
 
     def sync_down_logs(self, handle: ResourceHandle, job_id: int) -> None:
         codegen = backend_utils.JobLibCodeGen()
         codegen.get_log_path(job_id)
         code = codegen.build()
-        log_dir = self.run_on_head(handle, code, stream_logs=False)[1].strip()
+        log_dir = self.run_on_head(handle,
+                                   code,
+                                   stream_logs=False,
+                                   require_outputs=True)[1].strip()
 
         local_log_dir = log_dir
         remote_log_dir = os.path.join(SKY_REMOTE_LOGS_ROOT, log_dir)
@@ -1350,6 +1364,9 @@ def _exec_code_on_head(
         detach_run: bool = False,
     ) -> None:
         """Executes generated code on the head node."""
+        colorama.init()
+        style = colorama.Style
+        fore = colorama.Fore
         with tempfile.NamedTemporaryFile('w', prefix='sky_app_') as fp:
             fp.write(codegen)
             fp.flush()
@@ -1374,15 +1391,15 @@ def _exec_code_on_head(
             f'--address=127.0.0.1:8265 --job-id {job_id} --no-wait '
             f'-- "{executable} -u {script_path} > {remote_log_path} 2>&1"')
 
-        self.run_on_head(handle,
-                         f'{cd} && {job_submit_cmd}',
-                         log_path=job_log_path,
-                         stream_logs=False,
-                         check=True)
+        returncode = self.run_on_head(handle,
+                                      f'{cd} && {job_submit_cmd}',
+                                      log_path=job_log_path,
+                                      stream_logs=False)
+        if returncode != 0:
+            logger.error(f'{fore.RED}Fail to submit job {job_id}.'
+                         f'{style.RESET_ALL}')
+            sys.exit(returncode)
 
-        colorama.init()
-        style = colorama.Style
-        fore = colorama.Fore
         logger.info('Job submitted with Job ID: '
                     f'{style.BRIGHT}{job_id}{style.RESET_ALL}')
 
@@ -1413,7 +1430,7 @@ def tail_logs(self, handle: ResourceHandle, job_id: int) -> None:
         code = codegen.build()
         click.secho('Start streaming logs...', fg='yellow')
         try:
-            self.run_on_head(handle, code, stream_logs=True, check=False)
+            self.run_on_head(handle, code, stream_logs=True)
         except KeyboardInterrupt:
             # Do nothing. When receiving ctrl-c.
             pass
@@ -1423,7 +1440,10 @@ def _add_job(self, handle: ResourceHandle, job_name: str) -> int:
         username = getpass.getuser()
         codegen.add_job(job_name, username, self.run_timestamp)
         code = codegen.build()
-        job_id_str = self.run_on_head(handle, code, stream_logs=False)[1]
+        job_id_str = self.run_on_head(handle,
+                                      code,
+                                      stream_logs=False,
+                                      require_outputs=True)[1]
         try:
             job_id = int(job_id_str)
         except ValueError as e:
@@ -1597,11 +1617,12 @@ def teardown(self, handle: ResourceHandle, terminate: bool) -> None:
                     f'--instance-ids $({query_cmd})')
                 with console.status(f'[bold cyan]Terminating '
                                     f'[green]{cluster_name}'):
-                    proc, stdout, stderr = log_lib.run_with_log(
+                    returncode, stdout, stderr = log_lib.run_with_log(
                         terminate_cmd,
                         log_abs_path,
                         shell=True,
-                        stream_logs=False)
+                        stream_logs=False,
+                        require_outputs=True)
             elif isinstance(cloud, clouds.GCP):
                 zone = config['provider']['availability_zone']
                 query_cmd = (
@@ -1613,11 +1634,12 @@ def teardown(self, handle: ResourceHandle, terminate: bool) -> None:
                     f'$({query_cmd})')
                 with console.status(f'[bold cyan]Terminating '
                                     f'[green]{cluster_name}'):
-                    proc, stdout, stderr = log_lib.run_with_log(
+                    returncode, stdout, stderr = log_lib.run_with_log(
                         terminate_cmd,
                         log_abs_path,
                         shell=True,
-                        stream_logs=False)
+                        stream_logs=False,
+                        require_outputs=True)
             elif isinstance(cloud, clouds.Azure):
                 resource_group = config['provider']['resource_group']
                 query_cmd = (f'az vm list -g {resource_group} '
@@ -1625,11 +1647,12 @@ def teardown(self, handle: ResourceHandle, terminate: bool) -> None:
                 terminate_cmd = f'az vm delete --yes --ids $({query_cmd})'
                 with console.status(f'[bold cyan]Terminating '
                                     f'[green]{cluster_name}'):
-                    proc, stdout, stderr = log_lib.run_with_log(
+                    returncode, stdout, stderr = log_lib.run_with_log(
                         terminate_cmd,
                         log_abs_path,
                         shell=True,
-                        stream_logs=False)
+                        stream_logs=False,
+                        require_outputs=True)
             else:
                 raise ValueError(f'Unsupported cloud {cloud} for stopped '
                                  f'cluster {cluster_name!r}.')
@@ -1645,25 +1668,27 @@ def teardown(self, handle: ResourceHandle, terminate: bool) -> None:
                 teardown_verb = 'Terminating' if terminate else 'Stopping'
                 with console.status(f'[bold cyan]{teardown_verb} '
                                     f'[green]{cluster_name}'):
-                    proc, stdout, stderr = log_lib.run_with_log(
+                    returncode, stdout, stderr = log_lib.run_with_log(
                         ['ray', 'down', '-y', f.name],
                         log_abs_path,
-                        stream_logs=False)
+                        stream_logs=False,
+                        require_outputs=True)
 
             if handle.tpu_delete_script is not None:
                 with console.status('[bold cyan]Terminating TPU...'):
-                    tpu_proc, tpu_stdout, tpu_stderr = log_lib.run_with_log(
+                    tpu_rc, tpu_stdout, tpu_stderr = log_lib.run_with_log(
                         ['bash', handle.tpu_delete_script],
                         log_abs_path,
-                        stream_logs=False)
-                if tpu_proc.returncode != 0:
+                        stream_logs=False,
+                        require_outputs=True)
+                if tpu_rc != 0:
                     logger.error(f'{colorama.Fore.RED}Failed to delete TPU.\n'
                                  f'**** STDOUT ****\n'
                                  f'{tpu_stdout}\n'
                                  f'**** STDERR ****\n'
                                  f'{tpu_stderr}{colorama.Style.RESET_ALL}')
 
-        if proc.returncode != 0:
+        if returncode != 0:
             logger.error(
                 f'{colorama.Fore.RED}Failed to terminate {cluster_name}.\n'
                 f'**** STDOUT ****\n'
@@ -1757,9 +1782,13 @@ def _rsync_up(
         if with_outputs:
             # TODO(zhwu): Test this if the command will be still running in the
             # background, after the current program is killed.
-            backend_utils.run(command)
+            proc = backend_utils.run(command, check=False)
         else:
-            backend_utils.run_no_outputs(command)
+            proc = backend_utils.run_no_outputs(command, check=False)
+        if proc.returncode != 0:
+            logger.error(f'{colorama.Fore.RED}Failed to rsync up {source} '
+                         f'-> {target}.{colorama.Style.RESET_ALL}')
+            sys.exit(proc.returncode)
 
     def _ssh_control_name(self, handle: ResourceHandle) -> str:
         return f'{hashlib.md5(handle.cluster_yaml.encode()).hexdigest()[:10]}'
@@ -1802,9 +1831,9 @@ def run_on_head(
         log_path: str = '/dev/null',
         stream_logs: bool = False,
         use_cached_head_ip: bool = True,
-        check: bool = False,
         ssh_mode: backend_utils.SshMode = backend_utils.SshMode.NON_INTERACTIVE,
         under_remote_workdir: bool = False,
+        require_outputs: bool = False,
     ) -> Tuple[subprocess.Popen, str, str]:
         """Runs 'cmd' on the cluster's head node."""
         head_ip = self._get_head_ip(handle, use_cached_head_ip)
@@ -1821,7 +1850,7 @@ def run_on_head(
             port_forward=port_forward,
             log_path=log_path,
             stream_logs=stream_logs,
-            check=check,
             ssh_mode=ssh_mode,
             ssh_control_name=self._ssh_control_name(handle),
+            require_outputs=require_outputs,
         )
diff --git a/sky/cli.py b/sky/cli.py
index 3df3a0207e5..f416dc1280a 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -331,10 +331,10 @@ def _create_and_ssh_into_node(
         commands += ['screen', '-D', '-R']
     elif session_manager == 'tmux':
         commands += ['tmux', 'attach', '||', 'tmux', 'new']
-    commands = backend.run_on_head(handle,
-                                   commands,
-                                   port_forward=port_forward,
-                                   ssh_mode=backend_utils.SshMode.LOGIN)
+    backend.run_on_head(handle,
+                        commands,
+                        port_forward=port_forward,
+                        ssh_mode=backend_utils.SshMode.LOGIN)
     cluster_name = global_user_state.get_cluster_name_from_handle(handle)
 
     click.echo('To attach to it again:  ', nl=False)
@@ -796,7 +796,7 @@ def _show_job_queue_on_cluster(cluster: str, handle: Optional[Any],
             'Please re-launch it with `sky launch` to view the job queue.')
         return
 
-    job_table = backend.run_on_head(handle, code)[1]
+    job_table = backend.run_on_head(handle, code, require_outputs=True)[1]
     click.echo(f'{job_table}')
 
 
@@ -865,7 +865,9 @@ def cancel(cluster: str, all: bool, jobs: List[int]):  # pylint: disable=redefin
 
     # FIXME: Assumes a specific backend.
     backend = backends.CloudVmRayBackend()
-    backend.run_on_head(handle, code, stream_logs=False, check=True)
+    returncode = backend.run_on_head(handle, code, stream_logs=False)
+    if returncode != 0:
+        click.secho(f'Failed to cancel jobs on cluster {cluster}.', fg='red')
 
 
 @cli.command(cls=_DocumentedCodeCommand)
diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py
index 5e0d5dffe9d..a46bbb4039c 100644
--- a/sky/skylet/log_lib.py
+++ b/sky/skylet/log_lib.py
@@ -85,7 +85,7 @@ def run_with_log(
     log_path: str,
     stream_logs: bool = False,
     start_streaming_at: str = '',
-    return_none: bool = False,
+    require_outputs: bool = False,
     check: bool = False,
     shell: bool = False,
     with_ray: bool = False,
@@ -144,12 +144,10 @@ def run_with_log(
         )
         proc.wait()
         if proc.returncode and check:
-            if stderr:
-                print(stderr, file=sys.stderr)
             raise subprocess.CalledProcessError(proc.returncode, cmd)
-        if return_none:
-            return None
-        return proc, stdout, stderr
+        if require_outputs:
+            return proc.returncode, stdout, stderr
+        return proc.returncode
 
 
 def make_task_bash_script(codegen: str) -> str:
@@ -182,7 +180,7 @@ def run_bash_command_with_log(bash_command: str,
         fp.write(bash_command)
         fp.flush()
         script_path = fp.name
-        run_with_log(
+        return run_with_log(
             # Need this `-i` option to make sure `source ~/.bashrc` work.
             # Do not use shell=True because it will cause the environment
             # set in this task visible to other tasks. shell=False requires
@@ -190,8 +188,6 @@ def run_bash_command_with_log(bash_command: str,
             ['/bin/bash', '-i', script_path],
             log_path,
             stream_logs=stream_logs,
-            return_none=True,
-            check=True,
             with_ray=with_ray,
         )
 


Code review comments:
diff_hunk: @@ -1116,7 +1123,7 @@ def sync_workdir(self, handle: ResourceHandle, workdir: Path) -> None:
                     f'Symlink contents are not uploaded.{style.RESET_ALL}')
             self._rsync_up(handle,
                            ip=ip,
-                           source=f'{workdir}',
+                           source=f'{workdir}/',
path: sky/backends/cloud_vm_ray_backend.py
text: @author:
This is to fix #442.

@user1:
Workdir will still upload symlinked contents if `/` is removed. 

I'll add a PR that will add `/` if workdir is not symlink. Otherwise, no `/`