Repository: skypilot-org/skypilot
Pull request: [Core] Optimize kubernetes cmd executions with kubernetes command runner
<!-- Describe the changes in this PR -->

Fixes #3154, by refactoring the command runner and avoid using ssh in the backend for kubernetes pods. This is helpful for the future support of Slurm and other services that do not have ssh enabled.

This is also a prototype to reduce the launch and exec time on kubernetes cluster.

We will separate the optimization in this PR to multiple separate PRs:
1. #3394: lazy import and avoid rsync for exec **(1.3x faster for exec)**
2. #3484: avoid source bashrc for internal remote command execution **(1.3x faster for exec)**
3. This PR: kubernetes command runner

With the three optimization we can get **3x** speed up for exec

<details>

<summary>Exec speed (1.8x speed up)</summary>

```
multitime -n 5 sky exec test-speed echo hi -d
```
Master:
```
            Mean        Std.Dev.    Min         Median      Max
real        10.377      0.379       9.948       10.372      10.821      
user        1.373       0.025       1.327       1.381       1.401       
sys         0.178       0.021       0.150       0.171       0.209 
```
This PR
```
            Mean        Std.Dev.    Min         Median      Max
real        5.715       0.146       5.499       5.668       5.925       
user        1.582       0.132       1.475       1.512       1.835       
sys         0.209       0.027       0.170       0.202       0.253  
```

</details>

<details>

<summary>Launch on existing cluster (1.7x speed up)</summary>

`sky launch -c test-existing`
``multitime -n 5 sky launch -y -c test-existing echo hi``
master: 
```
            Mean        Std.Dev.    Min         Median      Max
real        40.467      2.363       37.563      39.865      44.683      
user        3.246       0.573       2.918       2.986       4.389       
sys         0.359       0.065       0.313       0.322       0.487 
```
This PR:
```
            Mean        Std.Dev.    Min         Median      Max
real        23.459      0.304       23.020      23.426      23.918      
user        3.220       0.041       3.155       3.223       3.266       
sys         0.433       0.028       0.389       0.438       0.463 
```

</details>

<details>

<summary>New clusters without setup/run (1.3x speed up)</summary>

``multitime -n 5 sky launch -y --cloud kubernetes --cpus 2``
Master: 
```
            Mean        Std.Dev.    Min         Median      Max
real        55.359      2.039       53.175      54.318      58.499      
user        3.180       0.046       3.112       3.177       3.236       
sys         0.377       0.014       0.356       0.385       0.391  
```
This PR:
```
            Mean        Std.Dev.    Min         Median      Max
real        41.574      1.206       39.681      41.475      43.460      
user        3.809       0.506       3.466       3.577       4.816       
sys         0.553       0.081       0.462       0.540       0.699 
```

</details>



<details>

<summary>New clusters with setup/run (1.4x speed up)</summary>

``multitime -n 5 sky launch -y --cloud kubernetes --cpus 2 task.yaml``
```
setup: echo Some "setup commands"
run: echo Some "run commands"
```
Master:
```
            Mean        Std.Dev.    Min         Median      Max
real        74.712      2.938       71.890      73.015      79.771      
user        3.598       0.465       3.318       3.373       4.526       
sys         0.463       0.104       0.376       0.435       0.661 
```
This PR:
```
            Mean        Std.Dev.    Min         Median      Max
real        54.095      1.973       51.659      54.739      56.846      
user        4.221       0.068       4.174       4.181       4.354       
sys         0.647       0.026       0.599       0.652       0.672 
```

</details>

<details>

<summary>New clusters with file mounts(1.6x speed up)</summary>

`task.yaml`
```yaml
resources:
  cpus: 1
  # image_id: docker:michaelvll/no-local-bin-test:v3
  # image_id: docker:michaelvll/k8s-dep-test:v2

file_mounts:
  /sky/examples: ./examples
  ~/.viminfo : ~/.viminfo
  ~/task.yaml : ./task.yaml
  ~/.config/gcloud: ~/.config/gcloud

run: |
  echo hi
```
``multitime -n 5 sky launch -y --cloud kubernetes --cpus 1 task.yaml``
Master: 
```
            Mean        Std.Dev.    Min         Median      Max
real        99.381      2.770       95.061      101.003     102.193     
user        4.529       0.458       4.255       4.319       5.443       
sys         0.903       0.115       0.821       0.840       1.126  
```
This PR:
```
            Mean        Std.Dev.    Min         Median      Max
real        62.851      2.610       59.383      62.714      67.189      
user        5.895       0.590       5.473       5.558       7.043       
sys         1.366       0.236       1.044       1.333       1.778 
```

</details>

~~Blocked by #3037~~.



<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [ ] Code formatting: `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
  - [x] `sky launch -c test-ssh --cloud gcp --cpus 2 --num-nodes 2 task.yaml`
  - [x] `sky launch -c test-k8s --cloud kubernetes --cpus 2 --num-nodes 2 task.yaml`
  - [x] `sky launch --cloud aws --cpus 2 task.yaml` with internal ips and jump server.
  - [x] `sky launch --cloud kubernetes --cpus 2 task.yaml`
  - [x]  `sky launch --cloud gcp --cpus 2 task.yaml`
  - [x] `sky launch -c test-docker --cloud gcp --cpus 2 --image-id docker:ubuntu:18.04 task.yaml`
  - [x] `sky launch -c test-docker --cloud aws --cpus 2 --image-id docker:ubuntu:18.04 task.yaml`
    ```yaml
    resources:
      cpus: 2
    
    
    file_mounts:
      /my-examples: ./examples
      ~/task.yaml: ./task.yaml
    
    setup: |
      ls /my-examples
    
    run: |
      ls ~
    ```
- [x] All smoke tests: `pytest tests/test_smoke.py --kubernetes` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
    Make sure `sky logs --sync-down` works for other clouds.
  - [x] `pytest tests/test_smoke.py::test_minimal --aws`
  - [x] `pytest tests/test_smoke.py::test_minimal --gcp`
  - [x] `pytest tests/test_smoke.py::test_minimal --kubernetes`
  - [x] `sky launch --cloud aws --num-nodes 4 -c min echo hi; sky logs --sync-down min` and check the local log dir
- [x] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh` (with stop commands commented out)


Problem statement:
[k8s] Slow `file_mounts` on k8s
Opening a SSH connection on k8s is high latency because of two extra steps:

1. Goes through k8s API server (`kubectl port-forward`)
2. Jumps through SSH jump pod.

This makes many operations slow, including file_mounts. Example, mounting three 1kb files takes 22s on EKS K8s compared to 6s on GCP:

```
file_mounts:
  ~/.gitconfig: ~/.gitconfig
  ~/tmp.txt: ~/tmp.txt
  ~/iam_policy.json: ~/iam_policy.json
```

Kubernetes:
```
I 02-13 10:33:42 backend_utils.py:1268] Syncing (to 1 node): ~/.gitconfig -> ~/.gitconfig
D 02-13 10:33:42 command_runner.py:110] --- Proxy: ssh -tt -i ~/.ssh/sky-key -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -W %h:%p sky@127.0.0.1 -o ProxyCommand='/Users/romilb/.sky/port-forward-proxy-cmd.sh'  ---
D 02-13 10:33:46 command_runner.py:110] --- Proxy: ssh -tt -i ~/.ssh/sky-key -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -W %h:%p sky@127.0.0.1 -o ProxyCommand='/Users/romilb/.sky/port-forward-proxy-cmd.sh'  ---
I 02-13 10:33:49 backend_utils.py:1268] Syncing (to 1 node): ~/tmp.txt -> ~/tmp.txt
D 02-13 10:33:49 command_runner.py:110] --- Proxy: ssh -tt -i ~/.ssh/sky-key -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -W %h:%p sky@127.0.0.1 -o ProxyCommand='/Users/romilb/.sky/port-forward-proxy-cmd.sh'  ---
D 02-13 10:33:54 command_runner.py:110] --- Proxy: ssh -tt -i ~/.ssh/sky-key -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -W %h:%p sky@127.0.0.1 -o ProxyCommand='/Users/romilb/.sky/port-forward-proxy-cmd.sh'  ---
I 02-13 10:33:57 backend_utils.py:1268] Syncing (to 1 node): ~/iam_policy.json -> ~/iam_policy.json
D 02-13 10:33:57 command_runner.py:110] --- Proxy: ssh -tt -i ~/.ssh/sky-key -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -W %h:%p sky@127.0.0.1 -o ProxyCommand='/Users/romilb/.sky/port-forward-proxy-cmd.sh'  ---
D 02-13 10:34:01 command_runner.py:110] --- Proxy: ssh -tt -i ~/.ssh/sky-key -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o IdentitiesOnly=yes -W %h:%p sky@127.0.0.1 -o ProxyCommand='/Users/romilb/.sky/port-forward-proxy-cmd.sh'  ---
D 02-13 10:34:05 cloud_vm_ray_backend.py:4571] File mount sync took 22.776047945022583 seconds.
```

GCP:
```
I 02-13 10:38:11 backend_utils.py:1268] Syncing (to 1 node): ~/.gitconfig -> ~/.gitconfig
I 02-13 10:38:13 backend_utils.py:1268] Syncing (to 1 node): ~/tmp.txt -> ~/tmp.txt
I 02-13 10:38:15 backend_utils.py:1268] Syncing (to 1 node): ~/iam_policy.json -> ~/iam_policy.json
D 02-13 10:38:18 cloud_vm_ray_backend.py:4571] File mount sync took 6.378494024276733 seconds.
```

Note that it is still fairly high throughput, just that the latency of establishing the connection is high.

Possible fixes:
1. Change our file_mounts codepath for k8s to use `kubectl cp` instead of rsync (and do so in a generic manner which works for other clouds, e.g. SLURM, too)
2. Consolidate all file_mounts into one SSH connection. This is generally good for perf on other clouds too.

Hints from issue discussion:

Commit message: longer wait

Patch to review:
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index b1598c7c039..2016949f96a 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -2650,27 +2650,6 @@ def stop_handler(signum, frame):
         raise KeyboardInterrupt(exceptions.SIGTSTP_CODE)
 
 
-def run_command_and_handle_ssh_failure(runner: command_runner.SSHCommandRunner,
-                                       command: str,
-                                       failure_message: str) -> str:
-    """Runs command remotely and returns output with proper error handling."""
-    rc, stdout, stderr = runner.run(command,
-                                    require_outputs=True,
-                                    stream_logs=False)
-    if rc == 255:
-        # SSH failed
-        raise RuntimeError(
-            f'SSH with user {runner.ssh_user} and key {runner.ssh_private_key} '
-            f'to {runner.ip} failed. This is most likely due to incorrect '
-            'credentials or incorrect permissions for the key file. Check '
-            'your credentials and try again.')
-    subprocess_utils.handle_returncode(rc,
-                                       command,
-                                       failure_message,
-                                       stderr=stderr)
-    return stdout
-
-
 def check_rsync_installed() -> None:
     """Checks if rsync is installed.
 
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 6933d3dfa0d..035f25526fa 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -3648,7 +3648,10 @@ def _rsync_down(args) -> None:
             try:
                 os.makedirs(local_log_dir, exist_ok=True)
                 runner.rsync(
-                    source=f'{remote_log_dir}/*',
+                    # Require a `/` at the end to make sure the parent dir
+                    # are not created locally. We do not add additional '*' as
+                    # kubernets's rsync does not work with an ending '*'.
+                    source=f'{remote_log_dir}/',
                     target=local_log_dir,
                     up=False,
                     stream_logs=False,
@@ -3657,7 +3660,7 @@ def _rsync_down(args) -> None:
                 if e.returncode == exceptions.RSYNC_FILE_NOT_FOUND_CODE:
                     # Raised by rsync_down. Remote log dir may not exist, since
                     # the job can be run on some part of the nodes.
-                    logger.debug(f'{runner.ip} does not have the tasks/*.')
+                    logger.debug(f'{runner.node_id} does not have the tasks/*.')
                 else:
                     raise
 
diff --git a/sky/provision/kubernetes/__init__.py b/sky/provision/kubernetes/__init__.py
index ca3938215c9..c72f0c14054 100644
--- a/sky/provision/kubernetes/__init__.py
+++ b/sky/provision/kubernetes/__init__.py
@@ -2,6 +2,7 @@
 
 from sky.provision.kubernetes.config import bootstrap_instances
 from sky.provision.kubernetes.instance import get_cluster_info
+from sky.provision.kubernetes.instance import get_command_runners
 from sky.provision.kubernetes.instance import query_instances
 from sky.provision.kubernetes.instance import run_instances
 from sky.provision.kubernetes.instance import stop_instances
diff --git a/sky/provision/kubernetes/instance.py b/sky/provision/kubernetes/instance.py
index 4f88293525f..782934cc8b2 100644
--- a/sky/provision/kubernetes/instance.py
+++ b/sky/provision/kubernetes/instance.py
@@ -13,6 +13,7 @@
 from sky.provision import docker_utils
 from sky.provision.kubernetes import config as config_lib
 from sky.provision.kubernetes import utils as kubernetes_utils
+from sky.utils import command_runner
 from sky.utils import common_utils
 from sky.utils import kubernetes_enums
 from sky.utils import ux_utils
@@ -158,6 +159,15 @@ def _raise_pod_scheduling_errors(namespace, new_nodes):
     raise config_lib.KubernetesError(f'{timeout_err_msg}')
 
 
+def _raise_command_running_error(message: str, command: str, pod_name: str,
+                                 rc: int, stdout: str):
+    if rc == 0:
+        return
+    raise config_lib.KubernetesError(
+        f'Failed to {message} for pod {pod_name} with return '
+        f'code {rc}: {command!r}\nOutput: {stdout}.')
+
+
 def _wait_for_pods_to_schedule(namespace, new_nodes, timeout: int):
     """Wait for all pods to be scheduled.
 
@@ -250,39 +260,6 @@ def _wait_for_pods_to_run(namespace, new_nodes):
         time.sleep(1)
 
 
-def _run_command_on_pods(node_name: str,
-                         node_namespace: str,
-                         command: List[str],
-                         stream_logs: bool = False):
-    """Run command on Kubernetes pods.
-
-    If `stream_logs` is True, we poll for output and error messages while the
-    command is executing, and the stdout and stderr is written to logger.info.
-    When called from the provisioner, this logger.info is written to the
-    provision.log file (see setup_provision_logging()).
-    """
-    cmd_output = kubernetes.stream()(
-        kubernetes.core_api().connect_get_namespaced_pod_exec,
-        node_name,
-        node_namespace,
-        command=command,
-        stderr=True,
-        stdin=False,
-        stdout=True,
-        tty=False,
-        _preload_content=(not stream_logs),
-        _request_timeout=kubernetes.API_TIMEOUT)
-    if stream_logs:
-        while cmd_output.is_open():
-            cmd_output.update(timeout=1)
-            if cmd_output.peek_stdout():
-                logger.info(f'{cmd_output.read_stdout().strip()}')
-            if cmd_output.peek_stderr():
-                logger.info(f'{cmd_output.read_stderr().strip()}')
-        cmd_output.close()
-    return cmd_output
-
-
 def _set_env_vars_in_pods(namespace: str, new_pods: List):
     """Setting environment variables in pods.
 
@@ -299,42 +276,44 @@ def _set_env_vars_in_pods(namespace: str, new_pods: List):
     /etc/profile.d/, making them available for all users in future
     shell sessions.
     """
-    set_k8s_env_var_cmd = [
-        '/bin/sh',
-        '-c',
-        docker_utils.SETUP_ENV_VARS_CMD,
-    ]
+    set_k8s_env_var_cmd = docker_utils.SETUP_ENV_VARS_CMD
 
     for new_pod in new_pods:
-        _run_command_on_pods(new_pod.metadata.name, namespace,
-                             set_k8s_env_var_cmd)
+        runner = command_runner.KubernetesCommandRunner(
+            (namespace, new_pod.metadata.name))
+        rc, stdout, _ = runner.run(set_k8s_env_var_cmd,
+                                   require_outputs=True,
+                                   stream_logs=False)
+        _raise_command_running_error('set env vars', set_k8s_env_var_cmd,
+                                     new_pod.metadata.name, rc, stdout)
 
 
 def _check_user_privilege(namespace: str, new_nodes: List) -> None:
     # Checks if the default user has sufficient privilege to set up
     # the kubernetes instance pod.
-    check_k8s_user_sudo_cmd = [
-        '/bin/sh',
-        '-c',
-        (
-            'if [ $(id -u) -eq 0 ]; then'
-            # If user is root, create an alias for sudo used in skypilot setup
-            '  echo \'alias sudo=""\' >> ~/.bashrc; '
-            'else '
-            '  if command -v sudo >/dev/null 2>&1; then '
-            '    timeout 2 sudo -l >/dev/null 2>&1 || '
-            f'    ( echo {exceptions.INSUFFICIENT_PRIVILEGES_CODE!r}; ); '
-            '  else '
-            f'    ( echo {exceptions.INSUFFICIENT_PRIVILEGES_CODE!r}; ); '
-            '  fi; '
-            'fi')
-    ]
+    check_k8s_user_sudo_cmd = (
+        'if [ $(id -u) -eq 0 ]; then'
+        # If user is root, create an alias for sudo used in skypilot setup
+        '  echo \'alias sudo=""\' >> ~/.bashrc; echo succeed;'
+        'else '
+        '  if command -v sudo >/dev/null 2>&1; then '
+        '    timeout 2 sudo -l >/dev/null 2>&1 && echo succeed || '
+        f'    ( echo {exceptions.INSUFFICIENT_PRIVILEGES_CODE!r}; ); '
+        '  else '
+        f'    ( echo {exceptions.INSUFFICIENT_PRIVILEGES_CODE!r}; ); '
+        '  fi; '
+        'fi')
 
     for new_node in new_nodes:
-        privilege_check = _run_command_on_pods(new_node.metadata.name,
-                                               namespace,
-                                               check_k8s_user_sudo_cmd)
-        if privilege_check == str(exceptions.INSUFFICIENT_PRIVILEGES_CODE):
+        runner = command_runner.KubernetesCommandRunner(
+            (namespace, new_node.metadata.name))
+        rc, stdout, _ = runner.run(check_k8s_user_sudo_cmd,
+                                   require_outputs=True,
+                                   stream_logs=False)
+        _raise_command_running_error('check user privilege',
+                                     check_k8s_user_sudo_cmd,
+                                     new_node.metadata.name, rc, stdout)
+        if stdout == str(exceptions.INSUFFICIENT_PRIVILEGES_CODE):
             raise config_lib.KubernetesError(
                 'Insufficient system privileges detected. '
                 'Ensure the default user has root access or '
@@ -345,44 +324,42 @@ def _check_user_privilege(namespace: str, new_nodes: List) -> None:
 def _setup_ssh_in_pods(namespace: str, new_nodes: List) -> None:
     # Setting up ssh for the pod instance. This is already setup for
     # the jump pod so it does not need to be run for it.
-    set_k8s_ssh_cmd = [
-        '/bin/sh',
-        '-c',
-        (
-            'set -x; '
-            'prefix_cmd() '
-            '{ if [ $(id -u) -ne 0 ]; then echo "sudo"; else echo ""; fi; }; '
-            'export DEBIAN_FRONTEND=noninteractive;'
-            '$(prefix_cmd) apt-get update;'
-            '$(prefix_cmd) apt install openssh-server rsync -y; '
-            '$(prefix_cmd) mkdir -p /var/run/sshd; '
-            '$(prefix_cmd) '
-            'sed -i "s/PermitRootLogin prohibit-password/PermitRootLogin yes/" '
-            '/etc/ssh/sshd_config; '
-            '$(prefix_cmd) sed '
-            '"s@session\\s*required\\s*pam_loginuid.so@session optional '
-            'pam_loginuid.so@g" -i /etc/pam.d/sshd; '
-            'cd /etc/ssh/ && $(prefix_cmd) ssh-keygen -A; '
-            '$(prefix_cmd) mkdir -p ~/.ssh; '
-            '$(prefix_cmd) chown -R $(whoami) ~/.ssh;'
-            '$(prefix_cmd) chmod 700 ~/.ssh; '
-            '$(prefix_cmd) chmod 644 ~/.ssh/authorized_keys; '
-            '$(prefix_cmd) cat /etc/secret-volume/ssh-publickey* > '
-            '~/.ssh/authorized_keys; '
-            '$(prefix_cmd) service ssh restart; '
-            # Eliminate the error
-            # `mesg: ttyname failed: inappropriate ioctl for device`.
-            # See https://www.educative.io/answers/error-mesg-ttyname-failed-inappropriate-ioctl-for-device  # pylint: disable=line-too-long
-            '$(prefix_cmd) sed -i "s/mesg n/tty -s \\&\\& mesg n/" ~/.profile;')
-    ]
-    # TODO(romilb): Parallelize the setup of SSH in pods for multi-node clusters
+    set_k8s_ssh_cmd = (
+        'set -ex; '
+        'prefix_cmd() '
+        '{ if [ $(id -u) -ne 0 ]; then echo "sudo"; else echo ""; fi; }; '
+        'export DEBIAN_FRONTEND=noninteractive;'
+        '$(prefix_cmd) apt-get update;'
+        '$(prefix_cmd) apt install openssh-server rsync -y; '
+        '$(prefix_cmd) mkdir -p /var/run/sshd; '
+        '$(prefix_cmd) '
+        'sed -i "s/PermitRootLogin prohibit-password/PermitRootLogin yes/" '
+        '/etc/ssh/sshd_config; '
+        '$(prefix_cmd) sed '
+        '"s@session\\s*required\\s*pam_loginuid.so@session optional '
+        'pam_loginuid.so@g" -i /etc/pam.d/sshd; '
+        'cd /etc/ssh/ && $(prefix_cmd) ssh-keygen -A; '
+        '$(prefix_cmd) mkdir -p ~/.ssh; '
+        '$(prefix_cmd) chown -R $(whoami) ~/.ssh;'
+        '$(prefix_cmd) chmod 700 ~/.ssh; '
+        '$(prefix_cmd) cat /etc/secret-volume/ssh-publickey* > '
+        '~/.ssh/authorized_keys; '
+        '$(prefix_cmd) chmod 644 ~/.ssh/authorized_keys; '
+        '$(prefix_cmd) service ssh restart; '
+        # Eliminate the error
+        # `mesg: ttyname failed: inappropriate ioctl for device`.
+        # See https://www.educative.io/answers/error-mesg-ttyname-failed-inappropriate-ioctl-for-device  # pylint: disable=line-too-long
+        '$(prefix_cmd) sed -i "s/mesg n/tty -s \\&\\& mesg n/" ~/.profile;')
+
     for new_node in new_nodes:
         pod_name = new_node.metadata.name
+        runner = command_runner.KubernetesCommandRunner((namespace, pod_name))
         logger.info(f'{"-"*20}Start: Set up SSH in pod {pod_name!r} {"-"*20}')
-        _run_command_on_pods(new_node.metadata.name,
-                             namespace,
-                             set_k8s_ssh_cmd,
-                             stream_logs=True)
+        rc, stdout, _ = runner.run(set_k8s_ssh_cmd,
+                                   require_outputs=True,
+                                   stream_logs=False)
+        _raise_command_running_error('setup ssh', set_k8s_ssh_cmd, pod_name, rc,
+                                     stdout)
         logger.info(f'{"-"*20}End: Set up SSH in pod {pod_name!r} {"-"*20}')
 
 
@@ -709,11 +686,15 @@ def get_cluster_info(
     assert cpu_request is not None, 'cpu_request should not be None'
 
     ssh_user = 'sky'
-    get_k8s_ssh_user_cmd = ['/bin/sh', '-c', ('echo $(whoami)')]
+    get_k8s_ssh_user_cmd = 'echo $(whoami)'
     assert head_pod_name is not None
-    ssh_user = _run_command_on_pods(head_pod_name, namespace,
-                                    get_k8s_ssh_user_cmd)
-    ssh_user = ssh_user.strip()
+    runner = command_runner.KubernetesCommandRunner((namespace, head_pod_name))
+    rc, stdout, _ = runner.run(get_k8s_ssh_user_cmd,
+                               require_outputs=True,
+                               stream_logs=False)
+    _raise_command_running_error('get ssh user', get_k8s_ssh_user_cmd,
+                                 head_pod_name, rc, stdout)
+    ssh_user = stdout.strip()
     logger.debug(
         f'Using ssh user {ssh_user} for cluster {cluster_name_on_cloud}')
 
@@ -776,3 +757,21 @@ def query_instances(
             continue
         cluster_status[pod.metadata.name] = pod_status
     return cluster_status
+
+
+def get_command_runners(
+    cluster_info: common.ClusterInfo,
+    **crednetials: Dict[str, Any],
+) -> List[command_runner.CommandRunner]:
+    """Get a command runner for the given cluster."""
+    assert cluster_info.provider_config is not None, cluster_info
+    instances = cluster_info.instances
+    namespace = _get_namespace(cluster_info.provider_config)
+    node_list = []
+    if cluster_info.head_instance_id is not None:
+        node_list = [(namespace, cluster_info.head_instance_id)]
+    node_list.extend((namespace, pod_name)
+                     for pod_name in instances.keys()
+                     if pod_name != cluster_info.head_instance_id)
+    return command_runner.KubernetesCommandRunner.make_runner_list(
+        node_list=node_list, **crednetials)
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index f43296c2f1e..0f161c30615 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -75,7 +75,7 @@ def ssh_options_list(
         # that case.
         'UserKnownHostsFile': os.devnull,
         # Suppresses the warning messages, such as:
-        #   Warning: Permanently added '34.69.216.203' (ED25519) to the list of
+        #   Warning: Permanently added 'xx.xx.xx.xx' (EDxxx) to the list of
         #   known hosts.
         'LogLevel': 'ERROR',
         # Try fewer extraneous key pairs.
@@ -170,7 +170,7 @@ def _get_command_to_run(
 
         # We need this to correctly run the cmd, and get the output.
         command = [
-            'bash',
+            '/bin/bash',
             '--login',
             '-c',
         ]
@@ -584,3 +584,249 @@ def rsync(
                                            error_msg,
                                            stderr=stdout + stderr,
                                            stream_logs=stream_logs)
+
+
+class KubernetesCommandRunner(CommandRunner):
+    """Runner for Kubernetes commands."""
+
+    def __init__(
+        self,
+        node: Tuple[str, str],
+        **kwargs,
+    ):
+        """Initialize KubernetesCommandRunner.
+
+        Example Usage:
+            runner = KubernetesCommandRunner((namespace, pod_name))
+            runner.run('ls -l')
+            runner.rsync(source, target, up=True)
+
+        Args:
+            node: The namespace and pod_name of the remote machine.
+        """
+        del kwargs
+        super().__init__(node)
+        self.namespace, self.pod_name = node
+
+    @timeline.event
+    def run(
+            self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = None,
+            require_outputs: bool = False,
+            # Advanced options.
+            log_path: str = os.devnull,
+            # If False, do not redirect stdout/stderr to optimize performance.
+            process_stream: bool = True,
+            stream_logs: bool = True,
+            ssh_mode: SshMode = SshMode.NON_INTERACTIVE,
+            separate_stderr: bool = False,
+            connect_timeout: Optional[int] = None,
+            source_bashrc: bool = False,
+            skip_lines: int = 0,
+            **kwargs) -> Union[int, Tuple[int, str, str]]:
+        """Uses 'kubectl exec' to run 'cmd' on a pod by its name and namespace.
+
+        Args:
+            cmd: The command to run.
+            port_forward: This should be None for k8s.
+
+            Advanced options:
+
+            require_outputs: Whether to return the stdout/stderr of the command.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            check: Check the success of the command.
+            ssh_mode: The mode to use for ssh.
+                See SSHMode for more details.
+            separate_stderr: Whether to separate stderr from stdout.
+            connect_timeout: timeout in seconds for the pod connection.
+            source_bashrc: Whether to source the bashrc before running the
+                command.
+            skip_lines: The number of lines to skip at the beginning of the
+                output. This is used when the output is not processed by
+                SkyPilot but we still want to get rid of some warning messages,
+                such as SSH warnings.
+
+        Returns:
+            returncode
+            or
+            A tuple of (returncode, stdout, stderr).
+        """
+        assert port_forward is None, ('port_forward is not supported for k8s '
+                                      f'for now, but got: {port_forward}')
+        if connect_timeout is None:
+            connect_timeout = 30
+        kubectl_args = [
+            '--pod-running-timeout', f'{connect_timeout}s', '-n',
+            self.namespace, self.pod_name
+        ]
+        if ssh_mode == SshMode.LOGIN:
+            assert isinstance(cmd, list), 'cmd must be a list for login mode.'
+            base_cmd = ['kubectl', 'exec', '-it', *kubectl_args, '--']
+            command = base_cmd + cmd
+            proc = subprocess_utils.run(command, shell=False, check=False)
+            return proc.returncode, '', ''
+
+        # Ignore ssh_mode for k8s.
+        assert port_forward is None, ('port_forward is not supported for k8s, '
+                                      f'but got: {port_forward}')
+        kubectl_base_command = ['kubectl', 'exec']
+
+        if ssh_mode == SshMode.INTERACTIVE:
+            kubectl_base_command.append('-i')
+        kubectl_base_command += [*kubectl_args, '--']
+
+        command_str = self._get_command_to_run(
+            cmd,
+            process_stream,
+            separate_stderr,
+            # +1 to skip first new line.
+            skip_lines=skip_lines + 1,
+            source_bashrc=source_bashrc)
+        command = kubectl_base_command + [
+            'bash', '-c', shlex.quote(command_str)
+        ]
+
+        log_dir = os.path.expanduser(os.path.dirname(log_path))
+        os.makedirs(log_dir, exist_ok=True)
+
+        executable = None
+        if not process_stream:
+            if stream_logs:
+                command += [
+                    f'| tee {log_path}',
+                    # This also requires the executor to be '/bin/bash' instead
+                    # of the default '/bin/sh'.
+                    '; exit ${PIPESTATUS[0]}'
+                ]
+            else:
+                command += [f'> {log_path}']
+            executable = '/bin/bash'
+        return log_lib.run_with_log(' '.join(command),
+                                    log_path,
+                                    require_outputs=require_outputs,
+                                    stream_logs=stream_logs,
+                                    process_stream=process_stream,
+                                    shell=True,
+                                    executable=executable,
+                                    **kwargs)
+
+    @timeline.event
+    def rsync(
+        self,
+        source: str,
+        target: str,
+        *,
+        up: bool,
+        # Advanced options.
+        log_path: str = os.devnull,
+        stream_logs: bool = True,
+        max_retry: int = 1,
+    ) -> None:
+        """Uses 'rsync' to sync 'source' to 'target'.
+
+        Args:
+            source: The source path.
+            target: The target path.
+            up: The direction of the sync, True for local to cluster, False
+              for cluster to local.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            max_retry: The maximum number of retries for the rsync command.
+              This value should be non-negative.
+
+        Raises:
+            exceptions.CommandError: rsync command failed.
+        """
+
+        def get_remote_home_dir() -> str:
+            # Use `echo ~` to get the remote home directory, instead of pwd or
+            # echo $HOME, because pwd can be `/` when the remote user is root
+            # and $HOME is not always set.
+            rc, remote_home_dir, _ = self.run('echo ~',
+                                              require_outputs=True,
+                                              stream_logs=False)
+            if rc != 0:
+                raise ValueError('Failed to get remote home directory.')
+            remote_home_dir = remote_home_dir.strip()
+            return remote_home_dir
+
+        # Build command.
+        # TODO(zhwu): This will print a per-file progress bar (with -P),
+        # shooting a lot of messages to the output. --info=progress2 is used
+        # to get a total progress bar, but it requires rsync>=3.1.0 and Mac
+        # OS has a default rsync==2.6.9 (16 years old).
+        helper_path = os.path.join(os.path.abspath(os.path.dirname(__file__)),
+                                   'kubernetes', 'rsync_helper.sh')
+        rsync_command = [
+            f'chmod +x {helper_path} &&', 'rsync', RSYNC_DISPLAY_OPTION
+        ]
+
+        # --filter
+        rsync_command.append(RSYNC_FILTER_OPTION)
+
+        if up:
+            # The source is a local path, so we need to resolve it.
+            # --exclude-from
+            resolved_source = pathlib.Path(source).expanduser().resolve()
+            if (resolved_source / GIT_EXCLUDE).exists():
+                # Ensure file exists; otherwise, rsync will error out.
+                #
+                # We shlex.quote() because the path may contain spaces:
+                #   'my dir/.git/info/exclude'
+                # Without quoting rsync fails.
+                rsync_command.append(
+                    RSYNC_EXCLUDE_OPTION.format(
+                        shlex.quote(str(resolved_source / GIT_EXCLUDE))))
+
+        rsync_command.append(f'-e "{helper_path}"')
+        # rsync with `kubectl` as the rsh command will cause ~/xx parsed as
+        # /~/xx, so we need to replace ~ with the remote home directory. We only
+        # need to do this when ~ is at the beginning of the path.
+        if up:
+            resolved_target = target
+            if target.startswith('~'):
+                remote_home_dir = get_remote_home_dir()
+                resolved_target = target.replace('~', remote_home_dir)
+            full_source_str = str(resolved_source)
+            if resolved_source.is_dir():
+                full_source_str = os.path.join(full_source_str, '')
+            rsync_command.extend([
+                f'{full_source_str!r}',
+                f'{self.pod_name}@{self.namespace}:{resolved_target!r}',
+            ])
+        else:
+            resolved_source = source
+            if source.startswith('~'):
+                remote_home_dir = get_remote_home_dir()
+                resolved_source = source.replace('~', remote_home_dir)
+            rsync_command.extend([
+                f'{self.pod_name}@{self.namespace}:{resolved_source!r}',
+                f'{os.path.expanduser(target)!r}',
+            ])
+        command = ' '.join(rsync_command)
+        logger.debug(f'Running rsync command: {command}')
+
+        backoff = common_utils.Backoff(initial_backoff=5, max_backoff_factor=5)
+        while max_retry >= 0:
+            returncode, _, stderr = log_lib.run_with_log(
+                command,
+                log_path=log_path,
+                stream_logs=stream_logs,
+                shell=True,
+                require_outputs=True)
+            if returncode == 0:
+                break
+            max_retry -= 1
+            time.sleep(backoff.current_backoff())
+
+        direction = 'up' if up else 'down'
+        error_msg = (f'Failed to rsync {direction}: {source} -> {target}. '
+                     'Ensure that the network is stable, then retry.')
+        subprocess_utils.handle_returncode(returncode,
+                                           command,
+                                           error_msg,
+                                           stderr=stderr,
+                                           stream_logs=stream_logs)
diff --git a/sky/utils/command_runner.pyi b/sky/utils/command_runner.pyi
index 9fbad243775..7a506bad2c6 100644
--- a/sky/utils/command_runner.pyi
+++ b/sky/utils/command_runner.pyi
@@ -193,3 +193,72 @@ class SSHCommandRunner(CommandRunner):
               log_path: str = ...,
               stream_logs: bool = ...) -> None:
         ...
+
+
+class KubernetesCommandRunner(CommandRunner):
+
+    def __init__(
+        self,
+        node: Tuple[str, str],
+    ) -> None:
+        ...
+
+    @typing.overload
+    def run(self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = ...,
+            require_outputs: Literal[False] = ...,
+            log_path: str = ...,
+            process_stream: bool = ...,
+            stream_logs: bool = ...,
+            ssh_mode: SshMode = ...,
+            separate_stderr: bool = ...,
+            connect_timeout: Optional[int] = ...,
+            source_bashrc: bool = ...,
+            skip_lines: int = ...,
+            **kwargs) -> int:
+        ...
+
+    @typing.overload
+    def run(self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = ...,
+            require_outputs: Literal[True],
+            log_path: str = ...,
+            process_stream: bool = ...,
+            stream_logs: bool = ...,
+            ssh_mode: SshMode = ...,
+            separate_stderr: bool = ...,
+            connect_timeout: Optional[int] = ...,
+            source_bashrc: bool = ...,
+            skip_lines: int = ...,
+            **kwargs) -> Tuple[int, str, str]:
+        ...
+
+    @typing.overload
+    def run(self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = ...,
+            require_outputs: bool = ...,
+            log_path: str = ...,
+            process_stream: bool = ...,
+            stream_logs: bool = ...,
+            ssh_mode: SshMode = ...,
+            separate_stderr: bool = ...,
+            connect_timeout: Optional[int] = ...,
+            source_bashrc: bool = ...,
+            skip_lines: int = ...,
+            **kwargs) -> Union[Tuple[int, str, str], int]:
+        ...
+
+    def rsync(self,
+              source: str,
+              target: str,
+              *,
+              up: bool,
+              log_path: str = ...,
+              stream_logs: bool = ...) -> None:
+        ...
diff --git a/sky/utils/kubernetes/rsync_helper.sh b/sky/utils/kubernetes/rsync_helper.sh
new file mode 100755
index 00000000000..d40a4c4185f
--- /dev/null
+++ b/sky/utils/kubernetes/rsync_helper.sh
@@ -0,0 +1,2 @@
+# When using pod@namespace, rsync passes args as: {us} -l pod namespace 
+shift;pod=$1;shift;namespace=$1;shift;kubectl exec -i $pod -n $namespace -- "$@"
diff --git a/sky/utils/subprocess_utils.py b/sky/utils/subprocess_utils.py
index bd48a91a796..d1779352a81 100644
--- a/sky/utils/subprocess_utils.py
+++ b/sky/utils/subprocess_utils.py
@@ -78,7 +78,7 @@ def handle_returncode(returncode: int,
         error_msg: The error message to print.
         stderr: The stderr of the command.
     """
-    echo = logger.error if stream_logs else lambda _: None
+    echo = logger.error if stream_logs else logger.debug
     if returncode != 0:
         if stderr is not None:
             echo(stderr)
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index db8f684c228..533302daaea 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -305,20 +305,15 @@ def test_example_app():
 # ---------- A minimal task ----------
 def test_minimal(generic_cloud: str):
     name = _get_cluster_name()
-    validate_output = _VALIDATE_LAUNCH_OUTPUT
-    # Kubernetes will output a SSH Warning for proxy jump, which will cause
-    # the output validation fail. We skip the check for kubernetes for now.
-    if generic_cloud.lower() == 'kubernetes':
-        validate_output = 'true'
     test = Test(
         'minimal',
         [
-            f's=$(sky launch -y -c {name} --cloud {generic_cloud} tests/test_yamls/minimal.yaml) && {validate_output}',
+            f's=$(sky launch -y -c {name} --cloud {generic_cloud} tests/test_yamls/minimal.yaml) && {_VALIDATE_LAUNCH_OUTPUT}',
             # Output validation done.
             f'sky logs {name} 1 --status',
             f'sky logs {name} --status | grep "Job 1: SUCCEEDED"',  # Equivalent.
             # Test launch output again on existing cluster
-            f's=$(sky launch -y -c {name} --cloud {generic_cloud} tests/test_yamls/minimal.yaml) && {validate_output}',
+            f's=$(sky launch -y -c {name} --cloud {generic_cloud} tests/test_yamls/minimal.yaml) && {_VALIDATE_LAUNCH_OUTPUT}',
             f'sky logs {name} 2 --status',
             f'sky logs {name} --status | grep "Job 2: SUCCEEDED"',  # Equivalent.
             # Check the logs downloading
@@ -3663,7 +3658,7 @@ def test_skyserve_fast_update(generic_cloud: str):
             f'{_SERVE_ENDPOINT_WAIT.format(name=name)}; curl http://$endpoint | grep "Hi, SkyPilot here"',
             f'sky serve update {name} --cloud {generic_cloud} --mode blue_green -y tests/skyserve/update/bump_version_after.yaml',
             # sleep to wait for update to be registered.
-            'sleep 30',
+            'sleep 40',
             # 2 on-deamnd (ready) + 1 on-demand (provisioning).
             (
                 _check_replica_in_status(
@@ -3677,7 +3672,7 @@ def test_skyserve_fast_update(generic_cloud: str):
             # Test rolling update
             f'sky serve update {name} --cloud {generic_cloud} -y tests/skyserve/update/bump_version_before.yaml',
             # sleep to wait for update to be registered.
-            'sleep 15',
+            'sleep 25',
             # 2 on-deamnd (ready) + 1 on-demand (shutting down).
             _check_replica_in_status(name, [(2, False, 'READY'),
                                             (1, False, 'SHUTTING_DOWN')]),


Code review comments:
diff_hunk: @@ -158,6 +159,15 @@ def _raise_pod_scheduling_errors(namespace, new_nodes):
     raise config_lib.KubernetesError(f'{timeout_err_msg}')
 
 
+def _raise_command_running_error(message: str, command: str, pod_name: str,
+                                 rc: int, stdout: str):
path: sky/provision/kubernetes/instance.py
text: @user1:
```suggestion
                                 rc: int, stdout: str) -> None:
```

diff_hunk: @@ -584,3 +584,249 @@ def rsync(
                                            error_msg,
                                            stderr=stdout + stderr,
                                            stream_logs=stream_logs)
+
+
+class KubernetesCommandRunner(CommandRunner):
+    """Runner for Kubernetes commands."""
+
+    def __init__(
+        self,
+        node: Tuple[str, str],
+        **kwargs,
+    ):
+        """Initialize KubernetesCommandRunner.
+
+        Example Usage:
+            runner = KubernetesCommandRunner((namespace, pod_name))
+            runner.run('ls -l')
+            runner.rsync(source, target, up=True)
+
+        Args:
+            node: The namespace and pod_name of the remote machine.
+        """
+        del kwargs
+        super().__init__(node)
+        self.namespace, self.pod_name = node
+
+    @timeline.event
+    def run(
+            self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = None,
+            require_outputs: bool = False,
+            # Advanced options.
+            log_path: str = os.devnull,
+            # If False, do not redirect stdout/stderr to optimize performance.
+            process_stream: bool = True,
+            stream_logs: bool = True,
+            ssh_mode: SshMode = SshMode.NON_INTERACTIVE,
+            separate_stderr: bool = False,
+            connect_timeout: Optional[int] = None,
+            source_bashrc: bool = False,
+            skip_lines: int = 0,
+            **kwargs) -> Union[int, Tuple[int, str, str]]:
+        """Uses 'kubectl exec' to run 'cmd' on a pod by its name and namespace.
+
+        Args:
+            cmd: The command to run.
+            port_forward: This should be None for k8s.
+
+            Advanced options:
+
+            require_outputs: Whether to return the stdout/stderr of the command.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            check: Check the success of the command.
+            ssh_mode: The mode to use for ssh.
+                See SSHMode for more details.
+            separate_stderr: Whether to separate stderr from stdout.
+            connect_timeout: timeout in seconds for the pod connection.
+            source_bashrc: Whether to source the bashrc before running the
+                command.
+            skip_lines: The number of lines to skip at the beginning of the
+                output. This is used when the output is not processed by
+                SkyPilot but we still want to get rid of some warning messages,
+                such as SSH warnings.
+
+        Returns:
+            returncode
+            or
+            A tuple of (returncode, stdout, stderr).
+        """
+        assert port_forward is None, ('port_forward is not supported for k8s '
+                                      f'for now, but got: {port_forward}')
+        if connect_timeout is None:
+            connect_timeout = 30
path: sky/utils/command_runner.py
text: @user1:
add a constant for this?

@author:
Added. Thanks!

diff_hunk: @@ -584,3 +584,249 @@ def rsync(
                                            error_msg,
                                            stderr=stdout + stderr,
                                            stream_logs=stream_logs)
+
+
+class KubernetesCommandRunner(CommandRunner):
+    """Runner for Kubernetes commands."""
+
+    def __init__(
+        self,
+        node: Tuple[str, str],
+        **kwargs,
+    ):
+        """Initialize KubernetesCommandRunner.
+
+        Example Usage:
+            runner = KubernetesCommandRunner((namespace, pod_name))
+            runner.run('ls -l')
+            runner.rsync(source, target, up=True)
+
+        Args:
+            node: The namespace and pod_name of the remote machine.
+        """
+        del kwargs
+        super().__init__(node)
+        self.namespace, self.pod_name = node
+
+    @timeline.event
+    def run(
+            self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = None,
+            require_outputs: bool = False,
+            # Advanced options.
+            log_path: str = os.devnull,
+            # If False, do not redirect stdout/stderr to optimize performance.
+            process_stream: bool = True,
+            stream_logs: bool = True,
+            ssh_mode: SshMode = SshMode.NON_INTERACTIVE,
+            separate_stderr: bool = False,
+            connect_timeout: Optional[int] = None,
+            source_bashrc: bool = False,
+            skip_lines: int = 0,
+            **kwargs) -> Union[int, Tuple[int, str, str]]:
+        """Uses 'kubectl exec' to run 'cmd' on a pod by its name and namespace.
+
+        Args:
+            cmd: The command to run.
+            port_forward: This should be None for k8s.
+
+            Advanced options:
+
+            require_outputs: Whether to return the stdout/stderr of the command.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            check: Check the success of the command.
+            ssh_mode: The mode to use for ssh.
+                See SSHMode for more details.
+            separate_stderr: Whether to separate stderr from stdout.
+            connect_timeout: timeout in seconds for the pod connection.
+            source_bashrc: Whether to source the bashrc before running the
+                command.
+            skip_lines: The number of lines to skip at the beginning of the
+                output. This is used when the output is not processed by
+                SkyPilot but we still want to get rid of some warning messages,
+                such as SSH warnings.
+
+        Returns:
+            returncode
+            or
+            A tuple of (returncode, stdout, stderr).
+        """
+        assert port_forward is None, ('port_forward is not supported for k8s '
+                                      f'for now, but got: {port_forward}')
+        if connect_timeout is None:
+            connect_timeout = 30
+        kubectl_args = [
+            '--pod-running-timeout', f'{connect_timeout}s', '-n',
+            self.namespace, self.pod_name
+        ]
+        if ssh_mode == SshMode.LOGIN:
+            assert isinstance(cmd, list), 'cmd must be a list for login mode.'
+            base_cmd = ['kubectl', 'exec', '-it', *kubectl_args, '--']
+            command = base_cmd + cmd
+            proc = subprocess_utils.run(command, shell=False, check=False)
+            return proc.returncode, '', ''
+
+        # Ignore ssh_mode for k8s.
+        assert port_forward is None, ('port_forward is not supported for k8s, '
+                                      f'but got: {port_forward}')
path: sky/utils/command_runner.py
text: @user1:
this is duplicated w/ L657-L658?

diff_hunk: @@ -584,3 +584,249 @@ def rsync(
                                            error_msg,
                                            stderr=stdout + stderr,
                                            stream_logs=stream_logs)
+
+
+class KubernetesCommandRunner(CommandRunner):
+    """Runner for Kubernetes commands."""
+
+    def __init__(
+        self,
+        node: Tuple[str, str],
+        **kwargs,
+    ):
+        """Initialize KubernetesCommandRunner.
+
+        Example Usage:
+            runner = KubernetesCommandRunner((namespace, pod_name))
+            runner.run('ls -l')
+            runner.rsync(source, target, up=True)
+
+        Args:
+            node: The namespace and pod_name of the remote machine.
+        """
+        del kwargs
+        super().__init__(node)
+        self.namespace, self.pod_name = node
+
+    @timeline.event
+    def run(
+            self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = None,
+            require_outputs: bool = False,
+            # Advanced options.
+            log_path: str = os.devnull,
+            # If False, do not redirect stdout/stderr to optimize performance.
+            process_stream: bool = True,
+            stream_logs: bool = True,
+            ssh_mode: SshMode = SshMode.NON_INTERACTIVE,
+            separate_stderr: bool = False,
+            connect_timeout: Optional[int] = None,
+            source_bashrc: bool = False,
+            skip_lines: int = 0,
+            **kwargs) -> Union[int, Tuple[int, str, str]]:
+        """Uses 'kubectl exec' to run 'cmd' on a pod by its name and namespace.
+
+        Args:
+            cmd: The command to run.
+            port_forward: This should be None for k8s.
+
+            Advanced options:
+
+            require_outputs: Whether to return the stdout/stderr of the command.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            check: Check the success of the command.
+            ssh_mode: The mode to use for ssh.
+                See SSHMode for more details.
+            separate_stderr: Whether to separate stderr from stdout.
+            connect_timeout: timeout in seconds for the pod connection.
+            source_bashrc: Whether to source the bashrc before running the
+                command.
+            skip_lines: The number of lines to skip at the beginning of the
+                output. This is used when the output is not processed by
+                SkyPilot but we still want to get rid of some warning messages,
+                such as SSH warnings.
+
+        Returns:
+            returncode
+            or
+            A tuple of (returncode, stdout, stderr).
+        """
+        assert port_forward is None, ('port_forward is not supported for k8s '
+                                      f'for now, but got: {port_forward}')
+        if connect_timeout is None:
+            connect_timeout = 30
+        kubectl_args = [
+            '--pod-running-timeout', f'{connect_timeout}s', '-n',
+            self.namespace, self.pod_name
+        ]
+        if ssh_mode == SshMode.LOGIN:
+            assert isinstance(cmd, list), 'cmd must be a list for login mode.'
+            base_cmd = ['kubectl', 'exec', '-it', *kubectl_args, '--']
+            command = base_cmd + cmd
+            proc = subprocess_utils.run(command, shell=False, check=False)
+            return proc.returncode, '', ''
+
+        # Ignore ssh_mode for k8s.
+        assert port_forward is None, ('port_forward is not supported for k8s, '
+                                      f'but got: {port_forward}')
+        kubectl_base_command = ['kubectl', 'exec']
+
+        if ssh_mode == SshMode.INTERACTIVE:
+            kubectl_base_command.append('-i')
+        kubectl_base_command += [*kubectl_args, '--']
+
+        command_str = self._get_command_to_run(
+            cmd,
+            process_stream,
+            separate_stderr,
+            # +1 to skip first new line.
+            skip_lines=skip_lines + 1,
+            source_bashrc=source_bashrc)
+        command = kubectl_base_command + [
+            'bash', '-c', shlex.quote(command_str)
path: sky/utils/command_runner.py
text: @user1:
```suggestion
            '/bin/bash', '-c', shlex.quote(command_str)
```

@user1:
actually, it seems like there is already an `bash -c` in `_get_command_to_run`. do we still need it?

https://github.com/skypilot-org/skypilot/blob/1a8228696518f78e0b99d397b146c721ed2bde35/sky/utils/command_runner.py#L172-L176

@author:
Added a comment. Thanks! : )

diff_hunk: @@ -584,3 +584,249 @@ def rsync(
                                            error_msg,
                                            stderr=stdout + stderr,
                                            stream_logs=stream_logs)
+
+
+class KubernetesCommandRunner(CommandRunner):
+    """Runner for Kubernetes commands."""
+
+    def __init__(
+        self,
+        node: Tuple[str, str],
+        **kwargs,
+    ):
+        """Initialize KubernetesCommandRunner.
+
+        Example Usage:
+            runner = KubernetesCommandRunner((namespace, pod_name))
+            runner.run('ls -l')
+            runner.rsync(source, target, up=True)
+
+        Args:
+            node: The namespace and pod_name of the remote machine.
+        """
+        del kwargs
+        super().__init__(node)
+        self.namespace, self.pod_name = node
+
+    @timeline.event
+    def run(
+            self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = None,
+            require_outputs: bool = False,
+            # Advanced options.
+            log_path: str = os.devnull,
+            # If False, do not redirect stdout/stderr to optimize performance.
+            process_stream: bool = True,
+            stream_logs: bool = True,
+            ssh_mode: SshMode = SshMode.NON_INTERACTIVE,
+            separate_stderr: bool = False,
+            connect_timeout: Optional[int] = None,
+            source_bashrc: bool = False,
+            skip_lines: int = 0,
+            **kwargs) -> Union[int, Tuple[int, str, str]]:
+        """Uses 'kubectl exec' to run 'cmd' on a pod by its name and namespace.
+
+        Args:
+            cmd: The command to run.
+            port_forward: This should be None for k8s.
+
+            Advanced options:
+
+            require_outputs: Whether to return the stdout/stderr of the command.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            check: Check the success of the command.
+            ssh_mode: The mode to use for ssh.
+                See SSHMode for more details.
+            separate_stderr: Whether to separate stderr from stdout.
+            connect_timeout: timeout in seconds for the pod connection.
+            source_bashrc: Whether to source the bashrc before running the
+                command.
+            skip_lines: The number of lines to skip at the beginning of the
+                output. This is used when the output is not processed by
+                SkyPilot but we still want to get rid of some warning messages,
+                such as SSH warnings.
+
+        Returns:
+            returncode
+            or
+            A tuple of (returncode, stdout, stderr).
+        """
+        assert port_forward is None, ('port_forward is not supported for k8s '
+                                      f'for now, but got: {port_forward}')
+        if connect_timeout is None:
+            connect_timeout = 30
+        kubectl_args = [
+            '--pod-running-timeout', f'{connect_timeout}s', '-n',
+            self.namespace, self.pod_name
+        ]
+        if ssh_mode == SshMode.LOGIN:
+            assert isinstance(cmd, list), 'cmd must be a list for login mode.'
+            base_cmd = ['kubectl', 'exec', '-it', *kubectl_args, '--']
+            command = base_cmd + cmd
+            proc = subprocess_utils.run(command, shell=False, check=False)
+            return proc.returncode, '', ''
+
+        # Ignore ssh_mode for k8s.
+        assert port_forward is None, ('port_forward is not supported for k8s, '
+                                      f'but got: {port_forward}')
+        kubectl_base_command = ['kubectl', 'exec']
+
+        if ssh_mode == SshMode.INTERACTIVE:
+            kubectl_base_command.append('-i')
+        kubectl_base_command += [*kubectl_args, '--']
+
+        command_str = self._get_command_to_run(
+            cmd,
+            process_stream,
+            separate_stderr,
+            # +1 to skip first new line.
+            skip_lines=skip_lines + 1,
+            source_bashrc=source_bashrc)
+        command = kubectl_base_command + [
+            'bash', '-c', shlex.quote(command_str)
+        ]
+
+        log_dir = os.path.expanduser(os.path.dirname(log_path))
+        os.makedirs(log_dir, exist_ok=True)
+
+        executable = None
+        if not process_stream:
+            if stream_logs:
+                command += [
+                    f'| tee {log_path}',
+                    # This also requires the executor to be '/bin/bash' instead
+                    # of the default '/bin/sh'.
+                    '; exit ${PIPESTATUS[0]}'
+                ]
+            else:
+                command += [f'> {log_path}']
+            executable = '/bin/bash'
+        return log_lib.run_with_log(' '.join(command),
+                                    log_path,
+                                    require_outputs=require_outputs,
+                                    stream_logs=stream_logs,
+                                    process_stream=process_stream,
+                                    shell=True,
+                                    executable=executable,
+                                    **kwargs)
+
+    @timeline.event
+    def rsync(
+        self,
+        source: str,
+        target: str,
+        *,
+        up: bool,
+        # Advanced options.
+        log_path: str = os.devnull,
+        stream_logs: bool = True,
+        max_retry: int = 1,
+    ) -> None:
+        """Uses 'rsync' to sync 'source' to 'target'.
+
+        Args:
+            source: The source path.
+            target: The target path.
+            up: The direction of the sync, True for local to cluster, False
+              for cluster to local.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            max_retry: The maximum number of retries for the rsync command.
+              This value should be non-negative.
+
+        Raises:
+            exceptions.CommandError: rsync command failed.
+        """
+
+        def get_remote_home_dir() -> str:
+            # Use `echo ~` to get the remote home directory, instead of pwd or
+            # echo $HOME, because pwd can be `/` when the remote user is root
+            # and $HOME is not always set.
+            rc, remote_home_dir, _ = self.run('echo ~',
+                                              require_outputs=True,
+                                              stream_logs=False)
+            if rc != 0:
+                raise ValueError('Failed to get remote home directory.')
+            remote_home_dir = remote_home_dir.strip()
+            return remote_home_dir
+
+        # Build command.
+        # TODO(zhwu): This will print a per-file progress bar (with -P),
+        # shooting a lot of messages to the output. --info=progress2 is used
+        # to get a total progress bar, but it requires rsync>=3.1.0 and Mac
+        # OS has a default rsync==2.6.9 (16 years old).
+        helper_path = os.path.join(os.path.abspath(os.path.dirname(__file__)),
+                                   'kubernetes', 'rsync_helper.sh')
+        rsync_command = [
+            f'chmod +x {helper_path} &&', 'rsync', RSYNC_DISPLAY_OPTION
+        ]
+
+        # --filter
+        rsync_command.append(RSYNC_FILTER_OPTION)
+
+        if up:
+            # The source is a local path, so we need to resolve it.
+            # --exclude-from
+            resolved_source = pathlib.Path(source).expanduser().resolve()
+            if (resolved_source / GIT_EXCLUDE).exists():
+                # Ensure file exists; otherwise, rsync will error out.
+                #
+                # We shlex.quote() because the path may contain spaces:
+                #   'my dir/.git/info/exclude'
+                # Without quoting rsync fails.
+                rsync_command.append(
+                    RSYNC_EXCLUDE_OPTION.format(
+                        shlex.quote(str(resolved_source / GIT_EXCLUDE))))
+
+        rsync_command.append(f'-e "{helper_path}"')
+        # rsync with `kubectl` as the rsh command will cause ~/xx parsed as
+        # /~/xx, so we need to replace ~ with the remote home directory. We only
+        # need to do this when ~ is at the beginning of the path.
+        if up:
+            resolved_target = target
+            if target.startswith('~'):
+                remote_home_dir = get_remote_home_dir()
+                resolved_target = target.replace('~', remote_home_dir)
+            full_source_str = str(resolved_source)
+            if resolved_source.is_dir():
+                full_source_str = os.path.join(full_source_str, '')
+            rsync_command.extend([
+                f'{full_source_str!r}',
+                f'{self.pod_name}@{self.namespace}:{resolved_target!r}',
+            ])
+        else:
+            resolved_source = source
+            if source.startswith('~'):
+                remote_home_dir = get_remote_home_dir()
+                resolved_source = source.replace('~', remote_home_dir)
+            rsync_command.extend([
+                f'{self.pod_name}@{self.namespace}:{resolved_source!r}',
+                f'{os.path.expanduser(target)!r}',
+            ])
+        command = ' '.join(rsync_command)
+        logger.debug(f'Running rsync command: {command}')
+
+        backoff = common_utils.Backoff(initial_backoff=5, max_backoff_factor=5)
+        while max_retry >= 0:
+            returncode, _, stderr = log_lib.run_with_log(
path: sky/utils/command_runner.py
text: @user1:
why dont we record the stdout here?

@author:
Good point! Added stdout to the output as well

path: sky/utils/kubernetes/rsync_helper.sh
text: @user1:
why do we use `;` to separate them in one single line instead of putting it in multiple lines? is it required to be a one liner by the `rsync -e`?

@author:
Good point! Added new lines for readability.

diff_hunk: @@ -584,3 +584,249 @@ def rsync(
                                            error_msg,
                                            stderr=stdout + stderr,
                                            stream_logs=stream_logs)
+
+
+class KubernetesCommandRunner(CommandRunner):
+    """Runner for Kubernetes commands."""
+
+    def __init__(
+        self,
+        node: Tuple[str, str],
+        **kwargs,
+    ):
+        """Initialize KubernetesCommandRunner.
+
+        Example Usage:
+            runner = KubernetesCommandRunner((namespace, pod_name))
+            runner.run('ls -l')
+            runner.rsync(source, target, up=True)
+
+        Args:
+            node: The namespace and pod_name of the remote machine.
+        """
+        del kwargs
+        super().__init__(node)
+        self.namespace, self.pod_name = node
+
+    @timeline.event
+    def run(
+            self,
+            cmd: Union[str, List[str]],
+            *,
+            port_forward: Optional[List[int]] = None,
+            require_outputs: bool = False,
+            # Advanced options.
+            log_path: str = os.devnull,
+            # If False, do not redirect stdout/stderr to optimize performance.
+            process_stream: bool = True,
+            stream_logs: bool = True,
+            ssh_mode: SshMode = SshMode.NON_INTERACTIVE,
+            separate_stderr: bool = False,
+            connect_timeout: Optional[int] = None,
+            source_bashrc: bool = False,
+            skip_lines: int = 0,
+            **kwargs) -> Union[int, Tuple[int, str, str]]:
+        """Uses 'kubectl exec' to run 'cmd' on a pod by its name and namespace.
+
+        Args:
+            cmd: The command to run.
+            port_forward: This should be None for k8s.
+
+            Advanced options:
+
+            require_outputs: Whether to return the stdout/stderr of the command.
+            log_path: Redirect stdout/stderr to the log_path.
+            stream_logs: Stream logs to the stdout/stderr.
+            check: Check the success of the command.
+            ssh_mode: The mode to use for ssh.
+                See SSHMode for more details.
+            separate_stderr: Whether to separate stderr from stdout.
+            connect_timeout: timeout in seconds for the pod connection.
+            source_bashrc: Whether to source the bashrc before running the
+                command.
+            skip_lines: The number of lines to skip at the beginning of the
+                output. This is used when the output is not processed by
+                SkyPilot but we still want to get rid of some warning messages,
+                such as SSH warnings.
+
+        Returns:
+            returncode
+            or
+            A tuple of (returncode, stdout, stderr).
+        """
+        assert port_forward is None, ('port_forward is not supported for k8s '
+                                      f'for now, but got: {port_forward}')
+        if connect_timeout is None:
+            connect_timeout = 30
+        kubectl_args = [
+            '--pod-running-timeout', f'{connect_timeout}s', '-n',
+            self.namespace, self.pod_name
+        ]
+        if ssh_mode == SshMode.LOGIN:
+            assert isinstance(cmd, list), 'cmd must be a list for login mode.'
+            base_cmd = ['kubectl', 'exec', '-it', *kubectl_args, '--']
+            command = base_cmd + cmd
+            proc = subprocess_utils.run(command, shell=False, check=False)
+            return proc.returncode, '', ''
+
+        # Ignore ssh_mode for k8s.
+        assert port_forward is None, ('port_forward is not supported for k8s, '
+                                      f'but got: {port_forward}')
+        kubectl_base_command = ['kubectl', 'exec']
+
+        if ssh_mode == SshMode.INTERACTIVE:
+            kubectl_base_command.append('-i')
+        kubectl_base_command += [*kubectl_args, '--']
+
+        command_str = self._get_command_to_run(
+            cmd,
+            process_stream,
+            separate_stderr,
+            # +1 to skip first new line.
+            skip_lines=skip_lines + 1,
+            source_bashrc=source_bashrc)
+        command = kubectl_base_command + [
+            'bash', '-c', shlex.quote(command_str)
+        ]
+
+        log_dir = os.path.expanduser(os.path.dirname(log_path))
+        os.makedirs(log_dir, exist_ok=True)
+
+        executable = None
+        if not process_stream:
+            if stream_logs:
+                command += [
+                    f'| tee {log_path}',
+                    # This also requires the executor to be '/bin/bash' instead
+                    # of the default '/bin/sh'.
+                    '; exit ${PIPESTATUS[0]}'
+                ]
+            else:
+                command += [f'> {log_path}']
+            executable = '/bin/bash'
+        return log_lib.run_with_log(' '.join(command),
+                                    log_path,
+                                    require_outputs=require_outputs,
+                                    stream_logs=stream_logs,
+                                    process_stream=process_stream,
+                                    shell=True,
+                                    executable=executable,
+                                    **kwargs)
+
+    @timeline.event
+    def rsync(
path: sky/utils/command_runner.py
text: @user1:
seems like there a lot of common code in this function and `SSHCommandRunner::rsync`. Is it possible to reuse some of them?

@author:
Good catch! Just refactored the code : )

diff_hunk: @@ -3648,7 +3648,10 @@ def _rsync_down(args) -> None:
             try:
                 os.makedirs(local_log_dir, exist_ok=True)
                 runner.rsync(
-                    source=f'{remote_log_dir}/*',
+                    # Require a `/` at the end to make sure the parent dir
+                    # are not created locally. We do not add additional '*' as
+                    # kubernets's rsync does not work with an ending '*'.
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
```suggestion
                    # kubernetes's rsync does not work with an ending '*'.
```

diff_hunk: @@ -776,3 +757,21 @@ def query_instances(
             continue
         cluster_status[pod.metadata.name] = pod_status
     return cluster_status
+
+
+def get_command_runners(
+    cluster_info: common.ClusterInfo,
+    **crednetials: Dict[str, Any],
path: sky/provision/kubernetes/instance.py
text: @user1:
```suggestion
    **credentials: Dict[str, Any],
```

@author:
Fixed. Thanks!