diff --git a/python/ray/autoscaler/_private/commands.py b/python/ray/autoscaler/_private/commands.py
index 8074cd048..f58e5063d 100644
--- a/python/ray/autoscaler/_private/commands.py
+++ b/python/ray/autoscaler/_private/commands.py
@@ -1015,9 +1015,9 @@ def rsync(config_file: str,
             provider.get_node_id(ip_address, use_internal_ip=use_internal_ip)
         ]
     else:
+        nodes = [head_node]
         if all_nodes:
-            nodes = _get_worker_nodes(config, override_cluster_name)
-        nodes += [head_node]
+            nodes.extend(_get_worker_nodes(config, override_cluster_name))
 
     for node_id in nodes:
         rsync_to_node(node_id, is_head_node=(node_id == head_node))
diff --git a/python/ray/autoscaler/sdk.py b/python/ray/autoscaler/sdk.py
index 334f45b3c..7c7f07546 100644
--- a/python/ray/autoscaler/sdk.py
+++ b/python/ray/autoscaler/sdk.py
@@ -41,20 +41,26 @@ def create_or_update_cluster(cluster_config: Union[dict, str],
             use_login_shells=True)
 
 
-def teardown_cluster(cluster_config: Union[dict, str]) -> None:
+def teardown_cluster(cluster_config: Union[dict, str],
+                     workers_only: bool = False,
+                     keep_min_workers: bool = False) -> None:
     """Destroys all nodes of a Ray cluster described by a config json.
 
     Args:
         cluster_config (Union[str, dict]): Either the config dict of the
             cluster, or a path pointing to a file containing the config.
+        workers_only (bool): Whether to keep the head node running and only
+            teardown worker nodes.
+        keep_min_workers (bool): Whether to keep min_workers (as specified
+            in the YAML) still running.
     """
     with _as_config_file(cluster_config) as config_file:
         return commands.teardown_cluster(
             config_file=config_file,
             yes=True,
-            workers_only=False,
+            workers_only=workers_only,
             override_cluster_name=None,
-            keep_min_workers=False)
+            keep_min_workers=keep_min_workers)
 
 
 def run_on_cluster(cluster_config: Union[dict, str],