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 sync_file_mounts(self, sync_cmd, step_numbers=(0, 2)):
      # step_numbers is (# of previous steps, total steps)
      previous_steps, total_steps = step_numbers
  
      nolog_paths = []
      if cli_logger.verbosity == 0:
          nolog_paths = [
              "~/ray_bootstrap_key.pem",
              "~/ray_bootstrap_config.yaml",
          ]
  
      def do_sync(
          remote_path, local_path, allow_non_existing_paths=False
      ):
          if allow_non_existing_paths and not os.path.exists(
              local_path
          ):
              cli_logger.print(
                  "sync: {} does not exist. Skipping.", local_path
              )
              # Ignore missing source files. In the future we should support
              # the --delete-missing-args command to delete files that have
              # been removed
              return
  
          assert os.path.exists(local_path), local_path
  
          if os.path.isdir(local_path):
              if not local_path.endswith("/"):
                  local_path += "/"
              if not remote_path.endswith("/"):
                  remote_path += "/"
  
          with LogTimer(
              self.log_prefix
              + "Synced {} to {}".format(local_path, remote_path)
          ):
              is_docker = (
                  self.docker_config
                  and self.docker_config["container_name"] != ""
              )
              if not is_docker:
                  # The DockerCommandRunner handles this internally.
                  self.cmd_runner.run(
                      "mkdir -p {}".format(
                          os.path.dirname(remote_path)
                      ),
                      run_env="host",
                  )
              sync_cmd(
                  local_path, remote_path, docker_mount_if_possible=True
              )
  
              if remote_path not in nolog_paths:
                  # todo: timed here?
                  cli_logger.print(
                      "{} from {}",
                      cf.bold(remote_path),
                      cf.bold(local_path),
                  )
  
      # Rsync file mounts
      with cli_logger.group(
          "Processing file mounts",
          _numbered=("[]", previous_steps + 1, total_steps),
      ):
          for remote_path, local_path in self.file_mounts.items():
              do_sync(remote_path, local_path)
          previous_steps += 1
  
      if self.cluster_synced_files:
          with cli_logger.group(
              "Processing worker file mounts",
              _numbered=("[]", previous_steps + 1, total_steps),
          ):
              cli_logger.print(
                  "synced files: {}", str(self.cluster_synced_files)
              )
              for path in self.cluster_synced_files:
                  do_sync(path, path, allow_non_existing_paths=True)
              previous_steps += 1
      else:
          cli_logger.print(
              "No worker file mounts to sync",
              _numbered=("[]", previous_steps + 1, total_steps),
          )
  
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 3 for calling the NodeUpdater obj's underlying member method code with example usage: obj.sync_file_mounts(sync_cmd, step_numbers) without return types