class_docstr: A process for syncing files and running init commands on a node.
  
  Arguments:
      node_id: the Node ID
      provider_config: Provider section of autoscaler yaml
      provider: NodeProvider Class
      auth_config: Auth section of autoscaler yaml
      cluster_name: the name of the cluster.
      file_mounts: Map of remote to local paths
      initialization_commands: Commands run before container launch
      setup_commands: Commands run before ray starts
      ray_start_commands: Commands to start ray
      runtime_hash: Used to check for config changes
      file_mounts_contents_hash: Used to check for changes to file mounts
      is_head_node: Whether to use head start/setup commands
      rsync_options: Extra options related to the rsync command.
      process_runner: the module to use to run the commands
          in the CommandRunner. E.g., subprocess.
      use_internal_ip: Wwhether the node_id belongs to an internal ip
          or external ip.
      docker_config: Docker section of autoscaler yaml
      restart_only: Whether to skip setup commands & just restart ray
      for_recovery: True if updater is for a recovering node. Only used for
          metric tracking.
class_docstr_tok: ["A","process","for","syncing","files","and","running","init","commands","on","a","node",".","Arguments",":","node_id",":","the","Node","ID","provider_config",":","Provider","section","of","autoscaler","yaml","provider",":","NodeProvider","Class","auth_config",":","Auth","section","of","autoscaler","yaml","cluster_name",":","the","name","of","the","cluster",".","file_mounts",":","Map","of","remote","to","local","paths","initialization_commands",":","Commands","run","before","container","launch","setup_commands",":","Commands","run","before","ray","starts","ray_start_commands",":","Commands","to","start","ray","runtime_hash",":","Used","to","check","for","config","changes","file_mounts_contents_hash",":","Used","to","check","for","changes","to","file","mounts","is_head_node",":","Whether","to","use","head","start\/setup","commands","rsync_options",":","Extra","options","related","to","the","rsync","command",".","process_runner",":","the","module","to","use","to","run","the","commands","in","the","CommandRunner",".","E.g.",",","subprocess",".","use_internal_ip",":","Wwhether","the","node_id","belongs","to","an","internal","ip","or","external","ip",".","docker_config",":","Docker","section","of","autoscaler","yaml","restart_only",":","Whether","to","skip","setup","commands","&","just","restart","ray","for_recovery",":","True","if","updater","is","for","a","recovering","node",".","Only","used","for","metric","tracking","."]
code: def wait_ready(self, deadline):
      with cli_logger.group(
          "Waiting for SSH to become available",
          _numbered=("[]", 1, NUM_SETUP_STEPS),
      ):
          with LogTimer(self.log_prefix + "Got remote shell"):
              cli_logger.print(
                  "Running `{}` as a test.", cf.bold("uptime")
              )
              first_conn_refused_time = None
              while True:
                  if time.time() > deadline:
                      raise Exception("wait_ready timeout exceeded.")
                  if self.provider.is_terminated(self.node_id):
                      raise Exception(
                          "wait_ready aborting because node "
                          "detected as terminated."
                      )
  
                  try:
                      # Run outside of the container
                      self.cmd_runner.run(
                          "uptime", timeout=10, run_env="host"
                      )
                      cli_logger.success("Success.")
                      return True
                  except ProcessRunnerError as e:
                      first_conn_refused_time = (
                          cmd_output_util.handle_ssh_fails(
                              e,
                              first_conn_refused_time,
                              retry_interval=READY_CHECK_INTERVAL,
                          )
                      )
                      time.sleep(READY_CHECK_INTERVAL)
                  except Exception as e:
                      # TODO(maximsmol): we should not be ignoring
                      # exceptions if they get filtered properly
                      # (new style log + non-interactive shells)
                      #
                      # however threading this configuration state
                      # is a pain and I'm leaving it for later
  
                      retry_str = "(" + str(e) + ")"
                      if hasattr(e, "cmd"):
                          if isinstance(e.cmd, str):
                              cmd_ = e.cmd
                          elif isinstance(e.cmd, list):
                              cmd_ = " ".join(e.cmd)
                          else:
                              logger.debug(
                                  f"e.cmd type ({type(e.cmd)}) not list or str."
                              )
                              cmd_ = str(e.cmd)
                          retry_str = "(Exit Status {}): {}".format(
                              e.returncode, cmd_
                          )
  
                      cli_logger.print(
                          "SSH still not available {}, retrying in {} seconds.",
                          cf.dimmed(retry_str),
                          cf.bold(str(READY_CHECK_INTERVAL)),
                      )
  
                      time.sleep(READY_CHECK_INTERVAL)
  
imports: import logging
  import os
  import subprocess
  import time
  import traceback
  from threading import Thread
  import click
  from ray._private.usage import usage_constants, usage_lib
  from ray.autoscaler._private import subprocess_output_util
  from ray.autoscaler._private.cli_logger import cf, cli_logger
  from ray.autoscaler._private.command_runner import AUTOSCALER_NODE_START_WAIT_S, ProcessRunnerError
  from ray.autoscaler._private.constants import LABELS_ENVIRONMENT_VARIABLE, RESOURCES_ENVIRONMENT_VARIABLE
  from ray.autoscaler._private.event_system import CreateClusterEvent, global_event_system
  from ray.autoscaler._private.log_timer import LogTimer
  from ray.autoscaler.tags import STATUS_SETTING_UP, STATUS_SYNCING_FILES, STATUS_UP_TO_DATE, STATUS_UPDATE_FAILED, STATUS_WAITING_FOR_SSH, TAG_RAY_FILE_MOUNTS_CONTENTS, TAG_RAY_NODE_STATUS, TAG_RAY_RUNTIME_CONFIG
label_desc: Use image node_id 4 for calling the NodeUpdater obj's underlying member method code with example usage: obj.wait_ready(deadline) and returns: True