docstr: Adds a node to the local Ray Cluster.
  
  All nodes are by default started with the following settings:
      cleanup=True,
      num_cpus=1,
      object_store_memory=150 * 1024 * 1024  # 150 MiB
  
  Args:
      wait: Whether to wait until the node is alive.
      node_args: Keyword arguments used in `start_ray_head` and
          `start_ray_node`. Overrides defaults.
  
  Returns:
      Node object of the added Ray node.
docstr_tok: ["Adds","a","node","to","the","local","Ray","Cluster",".","All","nodes","are","by","default","started","with","the","following","settings",":","cleanup=True",",","num_cpus=1",",","object_store_memory=150","*","1024","*","1024","#","150","MiB","Args",":","wait",":","Whether","to","wait","until","the","node","is","alive",".","node_args",":","Keyword","arguments","used","in","`","start_ray_head","`","and","`","start_ray_node","`",".","Overrides","defaults",".","Returns",":","Node","object","of","the","added","Ray","node","."]
code: def add_node(self, wait: bool = True, **node_args):
      """Adds a node to the local Ray Cluster.
  
      All nodes are by default started with the following settings:
          cleanup=True,
          num_cpus=1,
          object_store_memory=150 * 1024 * 1024  # 150 MiB
  
      Args:
          wait: Whether to wait until the node is alive.
          node_args: Keyword arguments used in `start_ray_head` and
              `start_ray_node`. Overrides defaults.
  
      Returns:
          Node object of the added Ray node.
      """
      default_kwargs = {
          "num_cpus": 1,
          "num_gpus": 0,
          "object_store_memory": 150 * 1024 * 1024,  # 150 MiB
          "min_worker_port": 0,
          "max_worker_port": 0,
          "dashboard_port": None,
      }
      ray_params = ray._private.parameter.RayParams(**node_args)
      ray_params.update_if_absent(**default_kwargs)
      with disable_client_hook():
          if self.head_node is None:
              node = ray._private.node.Node(
                  ray_params,
                  head=True,
                  shutdown_at_exit=self._shutdown_at_exit,
                  spawn_reaper=self._shutdown_at_exit,
              )
              self.head_node = node
              self.redis_address = self.head_node.redis_address
              self.redis_password = node_args.get(
                  "redis_password", ray_constants.REDIS_DEFAULT_PASSWORD
              )
              self.webui_url = self.head_node.webui_url
              # Init global state accessor when creating head node.
              gcs_options = GcsClientOptions.from_gcs_address(
                  node.gcs_address
              )
              self.global_state._initialize_global_state(gcs_options)
              # Write the Ray cluster address for convenience in unit
              # testing. ray.init() and ray.init(address="auto") will connect
              # to the local cluster.
              ray._private.utils.write_ray_address(
                  self.head_node.gcs_address
              )
          else:
              ray_params.update_if_absent(
                  redis_address=self.redis_address
              )
              ray_params.update_if_absent(gcs_address=self.gcs_address)
              # We only need one log monitor per physical node.
              ray_params.update_if_absent(include_log_monitor=False)
              # Let grpc pick a port.
              ray_params.update_if_absent(node_manager_port=0)
  
              node = ray._private.node.Node(
                  ray_params,
                  head=False,
                  shutdown_at_exit=self._shutdown_at_exit,
                  spawn_reaper=self._shutdown_at_exit,
              )
              self.worker_nodes.add(node)
  
          if wait:
              # Wait for the node to appear in the client table. We do this
              # so that the nodes appears in the client table in the order
              # that the corresponding calls to add_node were made. We do
              # this because in the tests we assume that the driver is
              # connected to the first node that is added.
              self._wait_for_node(node)
  
      return node
  
imports: import copy
  import json
  import logging
  import os
  import subprocess
  import tempfile
  import time
  from typing import Dict, Optional
  import yaml
  import ray
  import ray._private.services
  from ray._private import ray_constants
  from ray._private.client_mode_hook import disable_client_hook
  from ray._raylet import GcsClientOptions
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 5 for calling the Cluster obj's underlying member method code with example usage: obj.add_node(wait) and returns: node