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 __init__(
      self,
      node_id,
      provider_config,
      provider,
      auth_config,
      cluster_name,
      file_mounts,
      initialization_commands,
      setup_commands,
      ray_start_commands,
      runtime_hash,
      file_mounts_contents_hash,
      is_head_node,
      node_resources=None,
      node_labels=None,
      cluster_synced_files=None,
      rsync_options=None,
      process_runner=subprocess,
      use_internal_ip=False,
      docker_config=None,
      restart_only=False,
      for_recovery=False,
  ):
      self.log_prefix = "NodeUpdater: {}: ".format(node_id)
      use_internal_ip = use_internal_ip or provider_config.get(
          "use_internal_ips", False
      )
      self.cmd_runner = provider.get_command_runner(
          self.log_prefix,
          node_id,
          auth_config,
          cluster_name,
          process_runner,
          use_internal_ip,
          docker_config,
      )
  
      self.daemon = True
      self.node_id = node_id
      self.provider_type = provider_config.get("type")
      self.provider = provider
      # Some node providers don't specify empty structures as
      # defaults. Better to be defensive.
      file_mounts = file_mounts or {}
      self.file_mounts = {
          remote: os.path.expanduser(local)
          for remote, local in file_mounts.items()
      }
  
      self.initialization_commands = initialization_commands
      self.setup_commands = setup_commands
      self.ray_start_commands = ray_start_commands
      self.node_resources = node_resources
      self.node_labels = node_labels
      self.runtime_hash = runtime_hash
      self.file_mounts_contents_hash = file_mounts_contents_hash
      # TODO (Alex): This makes the assumption that $HOME on the head and
      # worker nodes is the same. Also note that `cluster_synced_files` is
      # set on the head -> worker updaters only (so `expanduser` is only run
      # on the head node).
      cluster_synced_files = cluster_synced_files or []
      self.cluster_synced_files = [
          os.path.expanduser(path) for path in cluster_synced_files
      ]
      self.rsync_options = rsync_options or {}
      self.auth_config = auth_config
      self.is_head_node = is_head_node
      self.docker_config = docker_config
      self.restart_only = restart_only
      self.update_time = None
      self.for_recovery = for_recovery
  
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 1 to create a new NodeUpdater object  with example: obj = NodeUpdater(node_id, provider_config, provider, auth_config, cluster_name, file_mounts, initialization_commands, setup_commands, ray_start_commands, runtime_hash, file_mounts_contents_hash, is_head_node, node_resources, node_labels, cluster_synced_files, rsync_options, process_runner, use_internal_ip, docker_config, restart_only, for_recovery)