[autoscaler] SDK fixes (#11517)

* [autoscaler] SDK Fxies

* add docs

* remove all_nodes

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))

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],