python/ray/autoscaler/_private/kuberay/node_provider.py
def terminate_nodes(self, node_ids: List[str]) -> Dict[str, Dict[str, str]]:
        """Batch terminates the specified nodes (= Kubernetes pods)."""
        with self._lock:
            # Split node_ids into groups according to node type and terminate
            # them individually. Note that in most cases, node_ids contains
            # a single element and therefore it is most likely not worth
            # optimizing this code to batch the requests to the API server.
            groups = {}
            current_replica_counts = {}
            label_filters = to_label_selector({"ray.io/cluster": self.cluster_name})
            pods = self._get(
                "pods?labelSelector=" + requests.utils.quote(label_filters)
            )
            for pod in pods["items"]:
                group_name = pod["metadata"]["labels"]["ray.io/group"]
                current_replica_counts[group_name] = (
                    current_replica_counts.get(group_name, 0) + 1
                )
                if pod["metadata"]["name"] in node_ids:
                    groups.setdefault(group_name, []).append(pod["metadata"]["name"])

            url = "rayclusters/{}".format(self.cluster_name)
            raycluster = self._get(url)

            for group_name, nodes in groups.items():
                group_index = _worker_group_index(raycluster, group_name)
                prefix = f"/spec/workerGroupSpecs/{group_index}"
                payload = [
                    {
                        "op": "replace",
                        "path": prefix + "/replicas",
                        "value": current_replica_counts[group_name] - len(nodes),
                    },
                    {
                        "op": "replace",
                        "path": prefix + "/scaleStrategy",
                        "value": {"workersToDelete": nodes},
                    },
                ]
                self._patch(url, payload)
        return {}
Batch terminates the specified nodes (= Kubernetes pods).
Batch terminates the specified nodes (= Kubernetes pods).
Batch terminates the specified nodes ( = Kubernetes pods ) .
def terminate_nodes(self, node_ids: List[str]) -> Dict[str, Dict[str, str]]:
        with self._lock:
            groups = {}
            current_replica_counts = {}
            label_filters = to_label_selector({"ray.io/cluster": self.cluster_name})
            pods = self._get(
                "pods?labelSelector=" + requests.utils.quote(label_filters)
            )
            for pod in pods["items"]:
                group_name = pod["metadata"]["labels"]["ray.io/group"]
                current_replica_counts[group_name] = (
                    current_replica_counts.get(group_name, 0) + 1
                )
                if pod["metadata"]["name"] in node_ids:
                    groups.setdefault(group_name, []).append(pod["metadata"]["name"])
            url = "rayclusters/{}".format(self.cluster_name)
            raycluster = self._get(url)
            for group_name, nodes in groups.items():
                group_index = _worker_group_index(raycluster, group_name)
                prefix = f"/spec/workerGroupSpecs/{group_index}"
                payload = [
                    {
                        "op": "replace",
                        "path": prefix + "/replicas",
                        "value": current_replica_counts[group_name] - len(nodes),
                    },
                    {
                        "op": "replace",
                        "path": prefix + "/scaleStrategy",
                        "value": {"workersToDelete": nodes},
                    },
                ]
                self._patch(url, payload)
        return {}
def terminate_nodes ( self , node_ids : List [ str ] ) -> Dict [ str , Dict [ str , str ] ] : with self . _lock : groups = { } current_replica_counts = { } label_filters = to_label_selector ( { "ray.io/cluster" : self . cluster_name } ) pods = self . _get ( "pods?labelSelector=" + requests . utils . quote ( label_filters ) ) for pod in pods [ "items" ] : group_name = pod [ "metadata" ] [ "labels" ] [ "ray.io/group" ] current_replica_counts [ group_name ] = ( current_replica_counts . get ( group_name , 0 ) + 1 ) if pod [ "metadata" ] [ "name" ] in node_ids : groups . setdefault ( group_name , [ ] ) . append ( pod [ "metadata" ] [ "name" ] ) url = "rayclusters/{}" . format ( self . cluster_name ) raycluster = self . _get ( url ) for group_name , nodes in groups . items ( ) : group_index = _worker_group_index ( raycluster , group_name ) prefix = f"/spec/workerGroupSpecs/{group_index}" payload = [ { "op" : "replace" , "path" : prefix + "/replicas" , "value" : current_replica_counts [ group_name ] - len ( nodes ) , } , { "op" : "replace" , "path" : prefix + "/scaleStrategy" , "value" : { "workersToDelete" : nodes } , } , ] self . _patch ( url , payload ) return { }
Batch terminates the specified nodes (= Kubernetes pods).
Batch terminates the specified nodes ( = Kubernetes pods ) .
"""Batch terminates the specified nodes (= Kubernetes pods).""" # Split node_ids into groups according to node type and terminate # them individually. Note that in most cases, node_ids contains # a single element and therefore it is most likely not worth # optimizing this code to batch the requests to the API server.
self
node_ids
List[str]