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 do_update(self):
      self.provider.set_node_tags(
          self.node_id, {TAG_RAY_NODE_STATUS: STATUS_WAITING_FOR_SSH}
      )
      cli_logger.labeled_value("New status", STATUS_WAITING_FOR_SSH)
  
      deadline = time.time() + AUTOSCALER_NODE_START_WAIT_S
      self.wait_ready(deadline)
      global_event_system.execute_callback(
          CreateClusterEvent.ssh_control_acquired
      )
  
      node_tags = self.provider.node_tags(self.node_id)
      logger.debug("Node tags: {}".format(str(node_tags)))
  
      if self.provider_type == "aws" and self.provider.provider_config:
          from ray.autoscaler._private.aws.cloudwatch.cloudwatch_helper import (
              CloudwatchHelper,
          )
  
          CloudwatchHelper(
              self.provider.provider_config,
              self.node_id,
              self.provider.cluster_name,
          ).update_from_config(self.is_head_node)
  
      if node_tags.get(TAG_RAY_RUNTIME_CONFIG) == self.runtime_hash:
          # When resuming from a stopped instance the runtime_hash may be the
          # same, but the container will not be started.
          init_required = self.cmd_runner.run_init(
              as_head=self.is_head_node,
              file_mounts=self.file_mounts,
              sync_run_yet=False,
          )
          if init_required:
              node_tags[TAG_RAY_RUNTIME_CONFIG] += "-invalidate"
              # This ensures that `setup_commands` are not removed
              self.restart_only = False
  
      if self.restart_only:
          self.setup_commands = []
  
      # runtime_hash will only change whenever the user restarts
      # or updates their cluster with `get_or_create_head_node`
      if node_tags.get(
          TAG_RAY_RUNTIME_CONFIG
      ) == self.runtime_hash and (
          not self.file_mounts_contents_hash
          or node_tags.get(TAG_RAY_FILE_MOUNTS_CONTENTS)
          == self.file_mounts_contents_hash
      ):
          # todo: we lie in the confirmation message since
          # full setup might be cancelled here
          cli_logger.print(
              "Configuration already up to date, "
              "skipping file mounts, initalization and setup commands.",
              _numbered=("[]", "2-6", NUM_SETUP_STEPS),
          )
  
      else:
          cli_logger.print(
              "Updating cluster configuration.",
              _tags=dict(hash=self.runtime_hash),
          )
  
          self.provider.set_node_tags(
              self.node_id, {TAG_RAY_NODE_STATUS: STATUS_SYNCING_FILES}
          )
          cli_logger.labeled_value("New status", STATUS_SYNCING_FILES)
          self.sync_file_mounts(
              self.rsync_up, step_numbers=(1, NUM_SETUP_STEPS)
          )
  
          # Only run setup commands if runtime_hash has changed because
          # we don't want to run setup_commands every time the head node
          # file_mounts folders have changed.
          if node_tags.get(TAG_RAY_RUNTIME_CONFIG) != self.runtime_hash:
              # Run init commands
              self.provider.set_node_tags(
                  self.node_id, {TAG_RAY_NODE_STATUS: STATUS_SETTING_UP}
              )
              cli_logger.labeled_value("New status", STATUS_SETTING_UP)
  
              if self.initialization_commands:
                  with cli_logger.group(
                      "Running initialization commands",
                      _numbered=("[]", 4, NUM_SETUP_STEPS),
                  ):
                      global_event_system.execute_callback(
                          CreateClusterEvent.run_initialization_cmd
                      )
                      with LogTimer(
                          self.log_prefix + "Initialization commands",
                          show_status=True,
                      ):
                          for cmd in self.initialization_commands:
                              global_event_system.execute_callback(
                                  CreateClusterEvent.run_initialization_cmd,
                                  {"command": cmd},
                              )
                              try:
                                  # Overriding the existing SSHOptions class
                                  # with a new SSHOptions class that uses
                                  # this ssh_private_key as its only __init__
                                  # argument.
                                  # Run outside docker.
                                  self.cmd_runner.run(
                                      cmd,
                                      ssh_options_override_ssh_key=self.auth_config.get(  # noqa: E501
                                          "ssh_private_key"
                                      ),
                                      run_env="host",
                                  )
                              except ProcessRunnerError as e:
                                  if e.msg_type == "ssh_command_failed":
                                      cli_logger.error("Failed.")
                                      cli_logger.error(
                                          "See above for stderr."
                                      )
  
                                  raise click.ClickException(
                                      "Initialization command failed."
                                  ) from None
              else:
                  cli_logger.print(
                      "No initialization commands to run.",
                      _numbered=("[]", 4, NUM_SETUP_STEPS),
                  )
              with cli_logger.group(
                  "Initializing command runner",
                  # todo: fix command numbering
                  _numbered=("[]", 5, NUM_SETUP_STEPS),
              ):
                  self.cmd_runner.run_init(
                      as_head=self.is_head_node,
                      file_mounts=self.file_mounts,
                      sync_run_yet=True,
                  )
              if self.setup_commands:
                  with cli_logger.group(
                      "Running setup commands",
                      # todo: fix command numbering
                      _numbered=("[]", 6, NUM_SETUP_STEPS),
                  ):
                      global_event_system.execute_callback(
                          CreateClusterEvent.run_setup_cmd
                      )
                      with LogTimer(
                          self.log_prefix + "Setup commands",
                          show_status=True,
                      ):
                          total = len(self.setup_commands)
                          for i, cmd in enumerate(self.setup_commands):
                              global_event_system.execute_callback(
                                  CreateClusterEvent.run_setup_cmd,
                                  {"command": cmd},
                              )
                              if (
                                  cli_logger.verbosity == 0
                                  and len(cmd) > 30
                              ):
                                  cmd_to_print = (
                                      cf.bold(cmd[:30]) + "..."
                                  )
                              else:
                                  cmd_to_print = cf.bold(cmd)
  
                              cli_logger.print(
                                  "{}",
                                  cmd_to_print,
                                  _numbered=("()", i, total),
                              )
  
                              try:
                                  # Runs in the container if docker is in use
                                  self.cmd_runner.run(
                                      cmd, run_env="auto"
                                  )
                              except ProcessRunnerError as e:
                                  if e.msg_type == "ssh_command_failed":
                                      cli_logger.error("Failed.")
                                      cli_logger.error(
                                          "See above for stderr."
                                      )
  
                                  raise click.ClickException(
                                      "Setup command failed."
                                  )
              else:
                  cli_logger.print(
                      "No setup commands to run.",
                      _numbered=("[]", 6, NUM_SETUP_STEPS),
                  )
  
      with cli_logger.group(
          "Starting the Ray runtime",
          _numbered=("[]", 7, NUM_SETUP_STEPS),
      ):
          global_event_system.execute_callback(
              CreateClusterEvent.start_ray_runtime
          )
          with LogTimer(
              self.log_prefix + "Ray start commands", show_status=True
          ):
              for cmd in self.ray_start_commands:
                  env_vars = {}
                  if self.is_head_node:
                      if usage_lib.usage_stats_enabled():
                          env_vars[
                              usage_constants.USAGE_STATS_ENABLED_ENV_VAR
                          ] = 1
                      else:
                          # Disable usage stats collection in the cluster.
                          env_vars[
                              usage_constants.USAGE_STATS_ENABLED_ENV_VAR
                          ] = 0
  
                  # Add a resource override env variable if needed.
                  # Local NodeProvider doesn't need resource and label override.
                  if self.provider_type != "local":
                      if self.node_resources:
                          env_vars[
                              RESOURCES_ENVIRONMENT_VARIABLE
                          ] = self.node_resources
                      if self.node_labels:
                          env_vars[
                              LABELS_ENVIRONMENT_VARIABLE
                          ] = self.node_labels
  
                  try:
                      old_redirected = (
                          cmd_output_util.is_output_redirected()
                      )
                      cmd_output_util.set_output_redirected(False)
                      # Runs in the container if docker is in use
                      self.cmd_runner.run(
                          cmd,
                          environment_variables=env_vars,
                          run_env="auto",
                      )
                      cmd_output_util.set_output_redirected(
                          old_redirected
                      )
                  except ProcessRunnerError as e:
                      if e.msg_type == "ssh_command_failed":
                          cli_logger.error("Failed.")
                          cli_logger.error("See above for stderr.")
  
                      raise click.ClickException(
                          "Start command failed."
                      )
          global_event_system.execute_callback(
              CreateClusterEvent.start_ray_runtime_completed
          )
  
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 5 for calling the NodeUpdater obj's underlying member method code with example usage: obj.do_update() without return types