code: def is_autoscaler_v2() -> bool:
      """
      Check if the autoscaler is v2 from reading GCS internal KV.
  
      If the method is called multiple times, the result will be cached in the module.
  
      Returns:
          is_v2: True if the autoscaler is v2, False otherwise.
  
      Raises:
          Exception: if GCS address could not be resolved (e.g. ray.init() not called)
      """
      # If env var is set to enable autoscaler v2, we should always return True.
      if ray._config.enable_autoscaler_v2():
          # TODO(rickyx): Once we migrate completely to v2, we should remove this.
          # While this short-circuit may allow client-server inconsistency
          # (e.g. client running v1, while server running v2), it's currently
          # not possible with existing use-cases.
          return True
  
      global cached_is_autoscaler_v2
      if cached_is_autoscaler_v2 is not None:
          return cached_is_autoscaler_v2
  
      if not _internal_kv_initialized():
          raise Exception(
              "GCS address could not be resolved (e.g. ray.init() not called)"
          )
  
      # See src/ray/common/constants.h for the definition of this key.
      cached_is_autoscaler_v2 = (
          _internal_kv_get(
              AUTOSCALER_V2_ENABLED_KEY.encode(),
              namespace=AUTOSCALER_NAMESPACE.encode(),
          )
          == b"1"
      )
  
      return cached_is_autoscaler_v2
  
imports: from collections import Counter, defaultdict
  from copy import deepcopy
  from datetime import datetime
  from itertools import chain
  from typing import Any, Dict, List, Tuple
  import ray
  from ray._private.ray_constants import AUTOSCALER_NAMESPACE, AUTOSCALER_V2_ENABLED_KEY
  from ray._private.utils import binary_to_hex
  from ray.autoscaler._private.autoscaler import AutoscalerSummary
  from ray.autoscaler._private.node_provider_availability_tracker import NodeAvailabilityRecord, NodeAvailabilitySummary, UnavailableNodeInformation
  from ray.autoscaler._private.util import LoadMetricsSummary, format_info_string
  from ray.autoscaler.v2.schema import NODE_DEATH_CAUSE_RAYLET_DIED, ClusterConstraintDemand, ClusterStatus, LaunchRequest, NodeInfo, NodeUsage, PlacementGroupResourceDemand, RayTaskActorDemand, ResourceDemand, ResourceDemandSummary, ResourceRequestByCount, ResourceUsage, Stats
  from ray.core.generated.autoscaler_pb2 import AutoscalingState, ClusterResourceState, GetClusterStatusReply, NodeState, NodeStatus, ResourceRequest
  from ray.core.generated.autoscaler_pb2 import ResourceRequestByCount
  from ray.core.generated.instance_manager_pb2 import Instance
  from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_initialized
label_desc: Use image node_id 3 for calling a global function with example usage: is_autoscaler_v2() and returns: cached_is_autoscaler_v2, True, cached_is_autoscaler_v2