docstr: Queries K8s for pods in the RayCluster. Converts that pod data into a
  map of pod name to Ray NodeData, as required by BatchingNodeProvider.
docstr_tok: ["Queries","K8s","for","pods","in","the","RayCluster",".","Converts","that","pod","data","into","a","map","of","pod","name","to","Ray","NodeData",",","as","required","by","BatchingNodeProvider","."]
code: def get_node_data(self) -> Dict[NodeID, NodeData]:
      """Queries K8s for pods in the RayCluster. Converts that pod data into a
      map of pod name to Ray NodeData, as required by BatchingNodeProvider.
      """
      # Store the raycluster CR
      self._raycluster = self._get(f"rayclusters/{self.cluster_name}")
  
      # Get the pods resource version.
      # Specifying a resource version in list requests is important for scalability:
      # https://kubernetes.io/docs/reference/using-api/api-concepts/#semantics-for-get-and-list
      resource_version = self._get_pods_resource_version()
      if resource_version:
          logger.info(
              f"Listing pods for RayCluster {self.cluster_name}"
              f" in namespace {self.namespace}"
              f" at pods resource version >= {resource_version}."
          )
  
      # Filter pods by cluster_name.
      label_selector = requests.utils.quote(
          f"ray.io/cluster={self.cluster_name}"
      )
  
      resource_path = f"pods?labelSelector={label_selector}"
      if resource_version:
          resource_path += (
              f"&resourceVersion={resource_version}"
              + "&resourceVersionMatch=NotOlderThan"
          )
  
      pod_list = self._get(resource_path)
      fetched_resource_version = pod_list["metadata"]["resourceVersion"]
      logger.info(
          f"Fetched pod data at resource version"
          f" {fetched_resource_version}."
      )
  
      # Extract node data from the pod list.
      node_data_dict = {}
      for pod in pod_list["items"]:
          # Kubernetes sets metadata.deletionTimestamp immediately after admitting a
          # request to delete an object. Full removal of the object may take some time
          # after the deletion timestamp is set. See link for details:
          # https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-deletion
          if "deletionTimestamp" in pod["metadata"]:
              # Ignore pods marked for termination.
              continue
          pod_name = pod["metadata"]["name"]
          node_data_dict[pod_name] = node_data_from_pod(pod)
      return node_data_dict
  
imports: import json
  import logging
  import os
  from collections import defaultdict
  from typing import Any, Dict, List, Optional, Tuple
  import requests
  from ray.autoscaler._private.constants import WORKER_LIVENESS_CHECK_KEY, WORKER_RPC_DRAIN_KEY
  from ray.autoscaler._private.util import NodeID, NodeIP, NodeKind, NodeStatus, NodeType
  from ray.autoscaler.batching_node_provider import BatchingNodeProvider, NodeData, ScaleRequest
  from ray.autoscaler.tags import NODE_KIND_HEAD, NODE_KIND_WORKER, STATUS_UP_TO_DATE, STATUS_UPDATE_FAILED, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 2 for calling the KubeRayNodeProvider obj's underlying member method code with example usage: obj.get_node_data() and returns: node_data_dict