class_docstr: ThreadedRayInstaller is responsible for install ray on new nodes.
class_docstr_tok: ["ThreadedRayInstaller","is","responsible","for","install","ray","on","new","nodes","."]
code: def _install_ray_on_single_node(self, instance: Instance) -> None:
      assert instance.status == Instance.ALLOCATED
      success, version = self._instance_storage.upsert_instance(
          instance, expected_instance_version=instance.version
      )
      if not success:
          logger.warning(
              f"Failed to update instance {instance.instance_id} to RAY_INSTALLING"
          )
          # Do not need to handle failures, it will be covered by
          # garbage collection.
          return
  
      # install with exponential backoff
      installed = False
      backoff_factor = 1
      for _ in range(self._max_install_attempts):
          installed = self._ray_installer.install_ray(
              instance, self._head_node_ip
          )
          if installed:
              break
          logger.warning("Failed to install ray, retrying...")
          time.sleep(self._install_retry_interval * backoff_factor)
          backoff_factor *= 2
  
      if not installed:
          instance.status = Instance.RAY_INSTALL_FAILED
          success, version = self._instance_storage.upsert_instance(
              instance,
              expected_instance_version=version,
          )
      else:
          instance.status = Instance.RAY_RUNNING
          success, version = self._instance_storage.upsert_instance(
              instance,
              expected_instance_version=version,
          )
      if not success:
          logger.warning(
              f"Failed to update instance {instance.instance_id} to {instance.status}"
          )
          # Do not need to handle failures, it will be covered by
          # garbage collection.
          return
  
file_path: ray/python/ray/autoscaler/v2/instance_manager/subscribers/threaded_ray_installer.py
imports: import logging
  import time
  from concurrent.futures import ThreadPoolExecutor
  from typing import List
  from ray.autoscaler.v2.instance_manager.instance_storage import InstanceStorage, InstanceUpdatedSubscriber
  from ray.autoscaler.v2.instance_manager.ray_installer import RayInstaller
  from ray.core.generated.instance_manager_pb2 import Instance, InstanceUpdateEvent
label_desc: Use image node_id 4 for calling the ThreadedRayInstaller obj's underlying member method code with example usage: obj._install_ray_on_single_node(instance) without return types