Use kubernetes command runner in cluster

runhouse/resources/envs/env.py

@@ -185,7 +185,7 @@ def install(self, force: bool = False, cluster: Cluster = None):
 
     def _full_command(self, command: str):
         if self._run_cmd:
-            return f"{self._run_cmd} $SHELL -c {shlex.quote(command)}"
+            return f"{self._run_cmd} ${{SHELL:-/bin/bash}} -c {shlex.quote(command)}"
         return command
 
     def _run_command(self, command: str, **kwargs):

runhouse/resources/hardware/cluster.py

@@ -451,14 +451,12 @@ def _command_runner(self, node: Optional[str] = None) -> "CommandRunner":
             ssh_control_name = ssh_credentials.pop(
                 "ssh_control_name", f"{node}:{self.ssh_port}"
             )
-            ssh_private_key = ssh_credentials.pop(
-                "ssh_private_key", ssh_credentials.pop("private_key", None)
-            )
 
             runner = SkySSHRunner(
                 (node, self.ssh_port),
-                ssh_private_key=ssh_private_key,
-                **ssh_credentials,
+                ssh_user=ssh_credentials.get("ssh_user"),
+                ssh_private_key=ssh_credentials.get("ssh_private_key"),
+                ssh_proxy_command=ssh_credentials.get("ssh_proxy_command"),
                 ssh_control_name=ssh_control_name,
                 docker_user=self.docker_user,
             )

runhouse/resources/hardware/kubernetes/rsync_helper.sh

@@ -0,0 +1,7 @@
+# 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 -- "$@"

runhouse/resources/hardware/on_demand_cluster.py

@@ -156,7 +156,9 @@ def docker_user(self) -> str:
             return None
 
         if self.launched_properties["cloud"] == "kubernetes":
-            return "root"
+            return self.launched_properties.get(
+                "docker_user", self.launched_properties.get("ssh_user", "root")
+            )
 
         from runhouse.resources.hardware.sky_ssh_runner import get_docker_user
 

runhouse/resources/hardware/sky_ssh_runner.py

@@ -183,7 +183,7 @@ def run(
             base_ssh_command.append("-q")
 
         if self.docker_user:  # RH MODIFIED
-            cmd = " ".join(cmd)
+            cmd = " ".join(cmd) if isinstance(cmd, list) else cmd
             cmd = f"conda deactivate && {cmd}"
 
         command_str = self._get_command_to_run(
@@ -428,7 +428,7 @@ def run(
         kubectl_base_command += [*kubectl_args, "--"]
 
         if self.docker_user:  # RH MODIFIED
-            cmd = " ".join(cmd)
+            cmd = " ".join(cmd) if isinstance(cmd, list) else cmd
             cmd = f"conda deactivate && {cmd}"
 
         command_str = self._get_command_to_run(

runhouse/resources/hardware/ssh_tunnel.py

@@ -55,7 +55,6 @@ def __init__(
             if ssh_control_name is None
             else _generate_ssh_control_hash(ssh_control_name)
         )
-        self.ssh_control_name = ssh_control_name
         self.ssh_proxy_command = ssh_proxy_command
         self.disable_control_master = disable_control_master
 
@@ -75,7 +74,6 @@ def __init__(
         self.tunnel_proc = None
 
     def tunnel(self, local_port, remote_port):
-        # TODO - modifications needed for k8s docker command to work
         base_cmd = _ssh_base_command(
             address=self.ip,
             ssh_user=self.ssh_user,

tests/test_resources/test_clusters/test_on_demand_cluster.py

@@ -175,8 +175,12 @@ def test_cluster_ping_and_is_up(self, cluster):
         cluster.address = None
         assert not cluster._ping(retry=False)
 
-        cluster.address = "00.00.000.11"
-        assert not cluster._ping(retry=False)
+        if not (
+            hasattr(cluster, "launched_properties")
+            and cluster.launched_properties["cloud"] == "kubernetes"
+        ):  # kubernetes does not use ips in command runner
+            cluster.address = "00.00.000.11"
+            assert not cluster._ping(retry=False)
 
         assert cluster._ping(retry=True)
         assert cluster.is_up()