ray-project--ray/python/ray/_private/node.py

import atexit
import collections
import datetime
import errno
import json
import logging
import os
import random
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
import traceback
from typing import IO, TYPE_CHECKING, AnyStr, Optional, Tuple

import ray
import ray._private.ray_constants as ray_constants
import ray._private.services
from ray._common.network_utils import (
    build_address,
    get_localhost_ip,
    is_ipv6,
    parse_address,
)
from ray._common.ray_constants import LOGGING_ROTATE_BACKUP_COUNT, LOGGING_ROTATE_BYTES
from ray._common.utils import try_to_create_directory
from ray._private.resource_and_label_spec import ResourceAndLabelSpec
from ray._private.resource_isolation_config import ResourceIsolationConfig
from ray._private.services import get_address, serialize_config
from ray._private.utils import (
    get_all_node_info_until_retrieved,
    is_in_test,
    open_log,
    try_to_symlink,
    validate_socket_filepath,
)
from ray._raylet import (
    GCS_SERVER_PORT_NAME,
    GcsClient,
    get_port_filename,
    get_session_key_from_storage,
    wait_for_persisted_port,
)
from ray.core.generated.gcs_pb2 import GcsNodeInfo
from ray.core.generated.gcs_service_pb2 import GetAllNodeInfoRequest

import psutil

if TYPE_CHECKING:
    from ray._private.parameter import RayParams

# Logger for this module. It should be configured at the entry point
# into the program using Ray. Ray configures it by default automatically
# using logging.basicConfig in its entry/init points.
logger = logging.getLogger(__name__)


class Node:
    """An encapsulation of the Ray processes on a single node.

    This class is responsible for starting Ray processes and killing them,
    and it also controls the temp file policy.

    Attributes:
        all_processes: A mapping from process type (str) to a list of
            ProcessInfo objects. All lists have length one except for the Redis
            server list, which has multiple.
    """

    def __init__(
        self,
        ray_params: "RayParams",
        head: bool = False,
        shutdown_at_exit: bool = True,
        spawn_reaper: bool = True,
        connect_only: bool = False,
        default_worker: bool = False,
        ray_init_cluster: bool = False,
    ):
        """Start a node.

        Args:
            ray_params: The RayParams to use to configure the node.
            head: True if this is the head node, which means it will
                start additional processes like the Redis servers, monitor
                processes, and web UI.
            shutdown_at_exit: If true, spawned processes will be cleaned
                up if this process exits normally.
            spawn_reaper: If true, spawns a process that will clean up
                other spawned processes if this process dies unexpectedly.
            connect_only: If true, connect to the node without starting
                new processes.
            default_worker: Whether it's running from a ray worker or not
            ray_init_cluster: Whether it's a cluster created by ray.init()
        """
        if shutdown_at_exit:
            if connect_only:
                raise ValueError(
                    "'shutdown_at_exit' and 'connect_only' cannot both be true."
                )
            self._register_shutdown_hooks()
        self._default_worker = default_worker
        self.head = head
        self.kernel_fate_share = bool(
            spawn_reaper and ray._private.utils.detect_fate_sharing_support()
        )
        self.resource_isolation_config: ResourceIsolationConfig = (
            ray_params.resource_isolation_config
        )
        self.all_processes: dict = {}
        self.removal_lock = threading.Lock()

        self.ray_init_cluster = ray_init_cluster
        if ray_init_cluster:
            assert head, "ray.init() created cluster only has the head node"

        # Set up external Redis when `RAY_REDIS_ADDRESS` is specified.
        redis_address_env = os.environ.get("RAY_REDIS_ADDRESS")
        if ray_params.external_addresses is None and redis_address_env is not None:
            external_redis = redis_address_env.split(",")

            # Reuse primary Redis as Redis shard when there's only one
            # instance provided.
            if len(external_redis) == 1:
                external_redis.append(external_redis[0])
            ray_params.external_addresses = external_redis
            ray_params.num_redis_shards = len(external_redis) - 1

        if (
            ray_params._system_config
            and len(ray_params._system_config) > 0
            and (not head and not connect_only)
        ):
            raise ValueError(
                "System config parameters can only be set on the head node."
            )

        ray_params.update_if_absent(
            include_log_monitor=True,
            resources={},
            worker_path=os.path.join(
                os.path.dirname(os.path.abspath(__file__)),
                "workers",
                "default_worker.py",
            ),
            setup_worker_path=os.path.join(
                os.path.dirname(os.path.abspath(__file__)),
                "workers",
                ray_constants.SETUP_WORKER_FILENAME,
            ),
        )

        self._resource_and_label_spec = None
        self._localhost = get_localhost_ip()
        self._ray_params = ray_params
        self._config = ray_params._system_config or {}

        # Configure log rotation parameters.
        self.max_bytes = int(os.getenv("RAY_ROTATION_MAX_BYTES", LOGGING_ROTATE_BYTES))
        self.backup_count = int(
            os.getenv("RAY_ROTATION_BACKUP_COUNT", LOGGING_ROTATE_BACKUP_COUNT)
        )

        assert self.max_bytes >= 0
        assert self.backup_count >= 0

        self._redis_address = ray_params.redis_address
        if head:
            ray_params.update_if_absent(num_redis_shards=1)
        self._gcs_address = ray_params.gcs_address
        self._gcs_client = None

        if not self.head:
            self.validate_ip_port(self.address)
            self._init_gcs_client()

        # Register the temp dir.
        self._session_name = ray_params.session_name
        if self._session_name is None:
            if head:
                # We expect this the first time we initialize a cluster, but not during
                # subsequent restarts of the head node.
                maybe_key = self.check_persisted_session_name()
                if maybe_key is None:
                    # date including microsecond
                    date_str = datetime.datetime.today().strftime(
                        "%Y-%m-%d_%H-%M-%S_%f"
                    )
                    self._session_name = f"session_{date_str}_{os.getpid()}"
                else:
                    self._session_name = ray._common.utils.decode(maybe_key)
            else:
                assert not self._default_worker
                session_name = ray._private.utils.internal_kv_get_with_retry(
                    self.get_gcs_client(),
                    "session_name",
                    ray_constants.KV_NAMESPACE_SESSION,
                    num_retries=ray_constants.NUM_REDIS_GET_RETRIES,
                )
                self._session_name = ray._common.utils.decode(session_name)

        # Initialize webui url
        if head:
            self._webui_url = None
        else:
            if ray_params.webui is None:
                assert not self._default_worker
                self._webui_url = ray._private.services.get_webui_url_from_internal_kv()
            else:
                self._webui_url = build_address(
                    ray_params.dashboard_host, ray_params.dashboard_port
                )

        # Resolve node to connect to
        node_to_connect_info = None
        if connect_only and not self._default_worker:
            node_to_connect_info = ray._private.services.get_node_to_connect_for_driver(
                self.get_gcs_client(),
                node_ip_address=ray_params.node_ip_address,
                node_name=ray_params.node_name,
                temp_dir=ray_params.temp_dir,
            )

        # Resolve node ID
        if connect_only:
            self._node_id = ray_params.node_id
            if self._node_id is None:
                self._node_id = node_to_connect_info.node_id.hex()
        else:
            if (
                self._ray_params.env_vars is not None
                and "RAY_OVERRIDE_NODE_ID_FOR_TESTING" in self._ray_params.env_vars
            ):
                node_id = self._ray_params.env_vars["RAY_OVERRIDE_NODE_ID_FOR_TESTING"]
                logger.debug(
                    f"Setting node ID to {node_id} "
                    "based on ray_params.env_vars override"
                )
                self._node_id = node_id
            elif os.environ.get("RAY_OVERRIDE_NODE_ID_FOR_TESTING"):
                node_id = os.environ["RAY_OVERRIDE_NODE_ID_FOR_TESTING"]
                logger.debug(f"Setting node ID to {node_id} based on env override")
                self._node_id = node_id
            else:
                node_id = ray.NodeID.from_random().hex()
                logger.debug(f"Setting node ID to {node_id}")
                self._node_id = node_id

        # Resolve node ip address
        node_ip_address = ray_params.node_ip_address
        if node_ip_address is None:
            if connect_only:
                assert node_to_connect_info is not None
                node_ip_address = node_to_connect_info.node_manager_address
            else:
                node_ip_address = ray.util.get_node_ip_address()
        assert node_ip_address is not None
        ray_params.update_if_absent(node_ip_address=node_ip_address)
        self._node_ip_address = node_ip_address

        # Resolve head node directories for defaults
        if not self.head and not connect_only:
            try:
                head_node_selector = GetAllNodeInfoRequest.NodeSelector()
                head_node_selector.is_head_node = True

                node_infos = get_all_node_info_until_retrieved(
                    self.get_gcs_client(),
                    timeout_per_retry=ray_constants.GCS_SERVER_REQUEST_TIMEOUT_SECONDS,
                    node_selectors=[head_node_selector],
                )
            except Exception as e:
                logger.exception(f"Failed to get head node info: {repr(e)}")
                raise e

            node_info = None
            for info in node_infos:
                node_info = info
                if info.state == GcsNodeInfo.GcsNodeState.ALIVE:
                    break
            self._head_temp_dir = getattr(node_info, "temp_dir", None)
            if self._head_temp_dir is None:
                raise Exception(
                    "Head node temp_dir not found in NodeInfo, "
                    "either GCS or head node's raylet may not have started successfully."
                )
            self._head_session_dir = getattr(node_info, "session_dir", None)
            if self._head_session_dir is None:
                raise Exception(
                    "Head node session dir not found in NodeInfo, "
                    "either GCS or head node's raylet may not have started successfully."
                )

        # It creates a session_dir.
        self._init_temp(node_to_connect_info)

        # Resolve socket and port names
        if connect_only:
            # Get socket names from the configuration.
            self._plasma_store_socket_name = ray_params.plasma_store_socket_name
            self._raylet_socket_name = ray_params.raylet_socket_name

            # If user does not provide the socket name, get it from GCS.
            if (
                self._plasma_store_socket_name is None
                or self._raylet_socket_name is None
                or self._ray_params.node_manager_port is None
            ):
                # Get the address info of the processes to connect to
                # from Redis or GCS.
                assert node_to_connect_info is not None
                self._plasma_store_socket_name = (
                    node_to_connect_info.object_store_socket_name
                )
                self._raylet_socket_name = node_to_connect_info.raylet_socket_name
                self._ray_params.node_manager_port = (
                    node_to_connect_info.node_manager_port
                )
        else:
            # If the user specified a socket name, use it.
            self._plasma_store_socket_name = self._prepare_socket_file(
                self._ray_params.plasma_store_socket_name, default_prefix="plasma_store"
            )
            self._raylet_socket_name = self._prepare_socket_file(
                self._ray_params.raylet_socket_name, default_prefix="raylet"
            )

        self._object_spilling_config = self._get_object_spilling_config()
        logger.debug(
            f"Starting node with object spilling config: {self._object_spilling_config}"
        )

        # Obtain the fallback directoy from the object spilling config
        # Currently, we set the fallback directory to be the same as the object spilling
        # path when the object spills to file system
        self._fallback_directory = None
        if self._object_spilling_config:
            config = json.loads(self._object_spilling_config)
            if config.get("type") == "filesystem":
                directory_path = config.get("params", {}).get("directory_path")
                if isinstance(directory_path, list):
                    self._fallback_directory = directory_path[0]
                elif isinstance(directory_path, str):
                    self._fallback_directory = directory_path
                if self._fallback_directory is None:
                    raise ValueError(
                        f"Failed to obtain fallback directory from "
                        f"object spilling config: {self._object_spilling_config}"
                    )

        # If it is a head node, try validating if external storage is configurable.
        if head:
            self.validate_external_storage()

        ray_params.update_if_absent(
            metrics_agent_port=ray_params.metrics_agent_port or 0,
            metrics_export_port=ray_params.metrics_export_port or 0,
            dashboard_agent_listen_port=ray_params.dashboard_agent_listen_port or 0,
            runtime_env_agent_port=ray_params.runtime_env_agent_port or 0,
        )

        # Pick a GCS server port.
        if head:
            # For GCS fault tolerance: if the port file already exists in the
            # current session directory, this indicates a GCS restart scenario.
            # We reuse the existing port so that other components can reconnect
            # to GCS after it restarts.
            gcs_port_filename = get_port_filename(self._node_id, GCS_SERVER_PORT_NAME)
            gcs_port_file = os.path.join(self._session_dir, gcs_port_filename)
            if os.path.exists(gcs_port_file):
                gcs_port = wait_for_persisted_port(
                    self._session_dir,
                    self._node_id,
                    GCS_SERVER_PORT_NAME,
                    timeout_ms=0,
                )
                ray_params.update_if_absent(gcs_server_port=gcs_port)

            else:
                gcs_server_port = os.getenv(ray_constants.GCS_PORT_ENVIRONMENT_VARIABLE)
                ray_params.update_if_absent(
                    gcs_server_port=int(gcs_server_port) if gcs_server_port else 0
                )

        # For worker nodes, check version compatibility before spawning
        # any processes (including the reaper). If the check fails (e.g.
        # Ray / Python version mismatch) we raise immediately rather than
        # leave orphaned child processes behind.
        if not head and not connect_only:
            self.check_version_info()

        if not connect_only and spawn_reaper and not self.kernel_fate_share:
            self.start_reaper_process()
        if not connect_only:
            self._ray_params.update_pre_selected_port()

        # Start processes.
        if head:
            self.start_head_processes()

        node_info = None
        if not connect_only:
            self.start_ray_processes()
            # Wait for the node info to be available in the GCS so that
            # we know it's started up.

            # Grace period to let the Raylet register with the GCS.
            # We retry in a loop in case it takes longer than expected.
            time.sleep(0.1)
            start_time = time.monotonic()
            raylet_start_wait_time_s = 30
            while True:
                try:
                    # Will raise a RuntimeError if the node info is not available.
                    node_info = ray._private.services.get_node(
                        self.gcs_address,
                        self._node_id,
                    )
                    break
                except RuntimeError as e:
                    logger.info(f"Failed to get node info {e}")
                if time.monotonic() - start_time > raylet_start_wait_time_s:
                    raise Exception(
                        "The current node timed out during startup. This "
                        "could happen because some of the raylet failed to "
                        "startup or the GCS has become overloaded."
                    )

        if connect_only:
            # Fetch node info to get labels.
            node_info = ray._private.services.get_node(
                self.gcs_address,
                self._node_id,
            )
        # Set node labels from GCS
        self._node_labels = node_info.get("labels", {}) if node_info else {}

        # port can be 0 or None for two cases:
        # 1. user is starting a new ray cluster and does not specify the port, components self-bind.
        # 2. user is connecting to an existing ray cluster, no port info is provided.
        # We always update port info from GCS to ensure consistency.
        self._ray_params.node_manager_port = node_info["node_manager_port"]
        self._ray_params.runtime_env_agent_port = node_info["runtime_env_agent_port"]
        self._ray_params.metrics_agent_port = node_info["metrics_agent_port"]
        self._ray_params.metrics_export_port = node_info["metrics_export_port"]
        self._ray_params.dashboard_agent_listen_port = node_info[
            "dashboard_agent_listen_port"
        ]

        # Makes sure the Node object has valid addresses after setup.
        self.validate_ip_port(self.address)
        self.validate_ip_port(self.gcs_address)

        if not connect_only:
            self._record_stats()

    def _resolve_ray_config(self, name, default):
        """Resolve a RAY_CONFIG value the same way the C++ GCS does.

        ``RayConfig::initialize`` (src/ray/common/ray_config.cc) first
        reads the ``RAY_<name>`` env var, then overrides it from the
        ``--config_list`` JSON, which Python passes as ``_system_config``.
        So ``_system_config`` wins over the env var, which wins over the
        default. Checking only ``os.environ`` here would miss a backend
        selected purely via ``_system_config`` and desync the Python-side
        rocksdb startup logic (marker file, GCS port wait) from the
        backend the GCS process actually picks.
        """
        return self._config.get(name, os.environ.get("RAY_" + name, default))

    def _is_rocksdb_gcs(self):
        return self._resolve_ray_config("gcs_storage", "memory") == "rocksdb"

    def _check_persisted_rocksdb_session_name(self):
        # Read the session_name marker file written by the previous head
        # process. GCS isn't up yet at this point in startup, so we can't
        # query the rocksdb-backed internal_kv directly; the marker file
        # bridges that gap. (This is a plain file on the GCS storage
        # volume, not a Kubernetes sidecar container.) See
        # src/ray/gcs/store_client/rocksdb_session_name_recovery.md for
        # the full rationale and write-path invariants.
        rocksdb_storage_path = self._resolve_ray_config("gcs_storage_path", "")
        if not rocksdb_storage_path:
            # Mirrors the C++ RAY_CHECK in
            # gcs_server.cc::GetStorageType(); fail loudly rather
            # than silently skipping recovery, which would
            # generate a fresh session_name and trip the assert
            # at the persisted-value check below.
            raise ValueError(
                "RAY_gcs_storage=rocksdb requires RAY_gcs_storage_path "
                "to be set to a writable directory."
            )
        session_name_file = os.path.join(rocksdb_storage_path, "session_name")
        try:
            with open(session_name_file, "rb") as f:
                persisted = f.read().strip()
                return persisted if persisted else None
        except FileNotFoundError:
            return None

    def check_persisted_session_name(self):
        # For the rocksdb GCS backend the session_name lives in a marker
        # file on the storage volume; delegate to the helper. Non-rocksdb
        # deployments fall through to the Redis path below.
        if self._is_rocksdb_gcs():
            return self._check_persisted_rocksdb_session_name()

        if self._ray_params.external_addresses is None:
            return None
        self._redis_address = self._ray_params.external_addresses[0]
        redis_ip_address, redis_port, enable_redis_ssl = get_address(
            self._redis_address,
        )
        # Address is ip:port or redis://ip:port
        if int(redis_port) < 0:
            raise ValueError(
                f"Invalid Redis port provided: {redis_port}."
                "The port must be a non-negative integer."
            )

        return get_session_key_from_storage(
            redis_ip_address,
            int(redis_port),
            self._ray_params.redis_username,
            self._ray_params.redis_password,
            enable_redis_ssl,
            serialize_config(self._config),
            b"session_name",
        )

    def _persist_rocksdb_session_name_file(self):
        """Durably write session_name to the RocksDB GCS storage directory.

        Companion to check_persisted_session_name(): on a head restart,
        that method reads this marker file before GCS (and thus
        internal_kv) is back up. The write is atomic and durable
        (tmp + fsync + rename + dir fsync) so the file survives a
        power-loss crash, matching the durability of the internal_kv_put
        that follows it.
        """
        rocksdb_storage_path = self._resolve_ray_config("gcs_storage_path", "")
        if not rocksdb_storage_path:
            # Symmetric with check_persisted_session_name(); see
            # there for the rationale. Without the marker file, the
            # internal_kv_put would persist a session_name in rocksdb
            # with no companion file, breaking restart recovery.
            raise ValueError(
                "RAY_gcs_storage=rocksdb requires RAY_gcs_storage_path "
                "to be set to a writable directory."
            )
        session_name_file = os.path.join(rocksdb_storage_path, "session_name")
        os.makedirs(rocksdb_storage_path, exist_ok=True)
        tmp_fd, tmp_path = tempfile.mkstemp(
            dir=rocksdb_storage_path,
            prefix="session_name.",
            suffix=".tmp",
        )
        try:
            with os.fdopen(tmp_fd, "wb") as f:
                f.write(self._session_name.encode("utf-8"))
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp_path, session_name_file)
            dir_fd = os.open(rocksdb_storage_path, os.O_DIRECTORY)
            try:
                os.fsync(dir_fd)
            finally:
                os.close(dir_fd)
        except BaseException:
            if os.path.exists(tmp_path):
                os.unlink(tmp_path)
            raise

    @staticmethod
    def validate_ip_port(ip_port):
        """Validates the address is in the ip:port format"""
        parts = parse_address(ip_port)
        if parts is None:
            raise ValueError(f"Port is not specified for address {ip_port}")
        try:
            _ = int(parts[1])
        except ValueError:
            raise ValueError(
                f"Unable to parse port number from {parts[1]} (full address = {ip_port})"
            )

    def check_version_info(self):
        """Check if the Python and Ray version of this process matches that in GCS.

        This will be used to detect if workers or drivers are started using
        different versions of Python, or Ray.

        Returns:
            None.

        Raises:
            Exception: An exception is raised if there is a version mismatch.
        """
        import ray._common.usage.usage_lib as ray_usage_lib

        cluster_metadata = ray_usage_lib.get_cluster_metadata(self.get_gcs_client())
        if cluster_metadata is None:
            cluster_metadata = ray_usage_lib.get_cluster_metadata(self.get_gcs_client())

        if not cluster_metadata:
            return
        node_ip_address = ray._private.services.get_node_ip_address()
        ray._private.utils.check_version_info(
            cluster_metadata, f"node {node_ip_address}"
        )

    def _register_shutdown_hooks(self):
        # Register the atexit handler. In this case, we shouldn't call sys.exit
        # as we're already in the exit procedure.
        def atexit_handler(*args):
            self.kill_all_processes(check_alive=False, allow_graceful=True)

        atexit.register(atexit_handler)

        # Register the handler to be called if we get a SIGTERM.
        # In this case, we want to exit with an error code (1) after
        # cleaning up child processes.
        def sigterm_handler(signum, frame):
            self.kill_all_processes(check_alive=False, allow_graceful=True)
            sys.exit(1)

        ray._private.utils.set_sigterm_handler(sigterm_handler)

    def _init_temp(self, node_to_connect_info: Optional[GcsNodeInfo]):
        # Create a dictionary to store temp file index.
        self._incremental_dict = collections.defaultdict(lambda: 0)

        self.temp_dir = self._ray_params.temp_dir
        if self.temp_dir is None:
            if node_to_connect_info is not None:
                self.temp_dir = node_to_connect_info.temp_dir
            else:
                if self.head:
                    self.temp_dir = ray._common.utils.get_default_ray_temp_dir()
                else:
                    assert not self._default_worker
                    self.temp_dir = self._head_temp_dir

        assert self._session_name is not None
        self._session_dir = os.path.join(self.temp_dir, self._session_name)
        session_symlink = os.path.join(self.temp_dir, ray_constants.SESSION_LATEST)
        self._sockets_dir = os.path.join(self._session_dir, "sockets")
        self._logs_dir = os.path.join(self._session_dir, "logs")
        old_logs_dir = os.path.join(self._logs_dir, "old")
        # Create a directory to be used for runtime environment.
        self._runtime_env_dir = os.path.join(
            self._session_dir, self._ray_params.runtime_env_dir_name
        )

        if node_to_connect_info is None:
            # Only create the temp dir on node creation
            try_to_create_directory(self.temp_dir)
            # Send a warning message if the session exists.
            try_to_create_directory(self._session_dir)
            try_to_symlink(session_symlink, self._session_dir)
            # Create a directory to be used for socket files.
            try_to_create_directory(self._sockets_dir)
            # Create a directory to be used for process log files.
            try_to_create_directory(self._logs_dir)
            try_to_create_directory(old_logs_dir)
            try_to_create_directory(self._runtime_env_dir)

        # Create a symlink to the libtpu tpu_logs directory if it exists.
        if "TPU_LOG_DIR" in os.environ and os.path.isdir(os.environ["TPU_LOG_DIR"]):
            tpu_log_dir = os.environ["TPU_LOG_DIR"]
        else:
            tpu_log_dir = "/tmp/tpu_logs"

        if os.path.isdir(tpu_log_dir):
            tpu_logs_symlink = os.path.join(self._logs_dir, "tpu_logs")
            try_to_symlink(tpu_logs_symlink, tpu_log_dir)

    def get_resource_and_label_spec(self):
        """Resolve and return the current ResourceAndLabelSpec for the node."""
        if not self._resource_and_label_spec:
            self._resource_and_label_spec = ResourceAndLabelSpec(
                self._ray_params.num_cpus,
                self._ray_params.num_gpus,
                self._ray_params.memory,
                self._ray_params.object_store_memory,
                self._ray_params.resources,
                self._ray_params.labels,
            ).resolve(
                is_head=self.head,
                node_ip_address=self.node_ip_address,
                resource_isolation_config=self.resource_isolation_config,
            )
        return self._resource_and_label_spec

    @property
    def node_id(self):
        """Get the node ID."""
        return self._node_id

    @property
    def session_name(self):
        """Get the current Ray session name."""
        return self._session_name

    @property
    def node_ip_address(self):
        """Get the IP address of this node."""
        return self._node_ip_address

    @property
    def address(self):
        """Get the address for bootstrapping, e.g. the address to pass to
        `ray start` or `ray.init()` to start worker nodes, that has been
        converted to ip:port format.
        """
        return self._gcs_address

    @property
    def gcs_address(self):
        """Get the gcs address."""
        assert self._gcs_address is not None, "Gcs address is not set"
        return self._gcs_address

    @property
    def redis_address(self):
        """Get the cluster Redis address."""
        return self._redis_address

    @property
    def redis_username(self):
        """Get the cluster Redis username."""
        return self._ray_params.redis_username

    @property
    def redis_password(self):
        """Get the cluster Redis password."""
        return self._ray_params.redis_password

    @property
    def plasma_store_socket_name(self):
        """Get the node's plasma store socket name."""
        return self._plasma_store_socket_name

    @property
    def unique_id(self):
        """Get a unique identifier for this node."""
        return f"{self.node_ip_address}:{self._plasma_store_socket_name}"

    @property
    def webui_url(self):
        """Get the cluster's web UI url."""
        return self._webui_url

    @property
    def raylet_socket_name(self):
        """Get the node's raylet socket name."""
        return self._raylet_socket_name

    @property
    def node_manager_port(self):
        """Get the node manager's port."""
        return self._ray_params.node_manager_port

    @property
    def metrics_export_port(self):
        """Get the port that exposes metrics"""
        return self._ray_params.metrics_export_port

    @property
    def metrics_agent_port(self):
        """Get the metrics agent gRPC port"""
        return self._ray_params.metrics_agent_port

    @property
    def runtime_env_agent_port(self):
        """Get the port that exposes runtime env agent as http"""
        return self._ray_params.runtime_env_agent_port

    @property
    def runtime_env_agent_address(self):
        """Get the address that exposes runtime env agent as http"""
        return f"http://{build_address(self._node_ip_address, self._ray_params.runtime_env_agent_port)}"

    @property
    def dashboard_agent_listen_port(self):
        """Get the dashboard agent's listen port"""
        return self._ray_params.dashboard_agent_listen_port

    @property
    def logging_config(self):
        """Get the logging config of the current node."""
        return {
            "log_rotation_max_bytes": self.max_bytes,
            "log_rotation_backup_count": self.backup_count,
        }

    @property
    def address_info(self):
        """Get a dictionary of addresses."""
        return {
            "node_ip_address": self._node_ip_address,
            "redis_address": self.redis_address,
            "object_store_address": self._plasma_store_socket_name,
            "raylet_socket_name": self._raylet_socket_name,
            "webui_url": self._webui_url,
            "session_dir": self._session_dir,
            "metrics_export_port": self._ray_params.metrics_export_port,
            "gcs_address": self.gcs_address,
            "address": self.address,
            "dashboard_agent_listen_port": self._ray_params.dashboard_agent_listen_port,
        }

    @property
    def node_labels(self):
        """Get the node labels."""
        return self._node_labels

    def is_head(self):
        return self.head

    def get_gcs_client(self):
        if self._gcs_client is None:
            self._init_gcs_client()
        return self._gcs_client

    def _init_gcs_client(self):
        if self.head:
            gcs_process = self.all_processes[ray_constants.PROCESS_TYPE_GCS_SERVER][
                0
            ].process
        else:
            gcs_process = None

        # TODO(ryw) instead of create a new GcsClient, wrap the one from
        # CoreWorkerProcess to save a grpc channel.
        #
        # RocksDB GCS recovery (open two on-disk DBs, WAL replay, GetAll
        # table scans that grow with cluster state) is slow -- worst on a
        # head restart over existing state -- and can exceed the default
        # ~20s connect window (NUM_REDIS_GET_RETRIES x ~1s sleep below).
        # This affects any node connecting while the head is reopening
        # RocksDB: the head itself (whose fault-tolerance restart reuses a
        # fixed GCS port, skipping start_gcs_server()'s port-file wait that
        # is gated on gcs_server_port == 0), and also workers/drivers
        # joining during that window.
        num_retries = ray_constants.NUM_REDIS_GET_RETRIES
        gcs_wait_override = os.environ.get("RAY_gcs_server_port_wait_time_s")
        if gcs_wait_override is not None:
            # Explicit operator budget: honor it exactly -- may lengthen OR
            # shorten the window -- so this env var behaves identically here
            # and in start_gcs_server()'s port-file wait. Floor at one
            # attempt (~1s per iteration below) so a 0 value still tries
            # once rather than skipping the connect entirely. This is also
            # the opt-in for workers/drivers whose pods lack RAY_gcs_storage
            # / RAY_gcs_storage_path and thus can't auto-detect the backend.
            num_retries = max(1, int(gcs_wait_override))
        elif self._is_rocksdb_gcs():
            # No explicit override, but we can see the backend is RocksDB:
            # extend to the same 120s budget as the port-zero path, without
            # shrinking a larger operator-configured retry count.
            num_retries = max(num_retries, 120)
        for _ in range(num_retries):
            gcs_address = None
            last_ex = None
            try:
                gcs_address = self.gcs_address
                client = GcsClient(
                    address=gcs_address,
                    cluster_id=self._ray_params.cluster_id,  # Hex string
                )
                self.cluster_id = client.cluster_id
                if self.head:
                    # Send a simple request to make sure GCS is alive
                    # if it's a head node.
                    client.internal_kv_get(b"dummy", None)
                self._gcs_client = client
                break
            except Exception:
                if gcs_process is not None and gcs_process.poll() is not None:
                    # GCS has exited.
                    break
                last_ex = traceback.format_exc()
                logger.debug(f"Connecting to GCS: {last_ex}")
                time.sleep(1)

        if self._gcs_client is None:
            if hasattr(self, "_logs_dir"):
                with open(os.path.join(self._logs_dir, "gcs_server.err")) as err:
                    # Use " C " or " E " to exclude the stacktrace.
                    # This should work for most cases, especitally
                    # it's when GCS is starting. Only display last 10 lines of logs.
                    errors = [e for e in err.readlines() if " C " in e or " E " in e][
                        -10:
                    ]
                error_msg = "\n" + "".join(errors) + "\n"
                raise RuntimeError(
                    f"Failed to {'start' if self.head else 'connect to'} GCS. "
                    f" Last {len(errors)} lines of error files:"
                    f"{error_msg}."
                    f"Please check {os.path.join(self._logs_dir, 'gcs_server.out')}"
                    f" for details. Last connection error: {last_ex}"
                )
            else:
                raise RuntimeError(
                    f"Failed to {'start' if self.head else 'connect to'} GCS. Last "
                    f"connection error: {last_ex}"
                )

        ray.experimental.internal_kv._initialize_internal_kv(self._gcs_client)

    def get_temp_dir_path(self):
        """Get the path of the temporary directory."""
        return self.temp_dir

    def get_runtime_env_dir_path(self):
        """Get the path of the runtime env."""
        return self._runtime_env_dir

    def get_session_dir_path(self):
        """Get the path of the session directory."""
        return self._session_dir

    def get_logs_dir_path(self):
        """Get the path of the log files directory."""
        return self._logs_dir

    def get_sockets_dir_path(self):
        """Get the path of the sockets directory."""
        return self._sockets_dir

    def _make_inc_temp(
        self, suffix: str = "", prefix: str = "", directory_name: Optional[str] = None
    ):
        """Return an incremental temporary file name. The file is not created.

        Args:
            suffix: The suffix of the temp file.
            prefix: The prefix of the temp file.
            directory_name: The base directory of the temp file.

        Returns:
            A string of file name. If there existing a file having
                the same name, the returned name will look like
                "{directory_name}/{prefix}.{unique_index}{suffix}"
        """
        if directory_name is None:
            directory_name = self.temp_dir
        directory_name = os.path.expanduser(directory_name)
        index = self._incremental_dict[suffix, prefix, directory_name]
        # `tempfile.TMP_MAX` could be extremely large,
        # so using `range` in Python2.x should be avoided.
        while index < tempfile.TMP_MAX:
            if index == 0:
                filename = os.path.join(directory_name, prefix + suffix)
            else:
                filename = os.path.join(
                    directory_name, prefix + "." + str(index) + suffix
                )
            index += 1
            if not os.path.exists(filename):
                # Save the index.
                self._incremental_dict[suffix, prefix, directory_name] = index
                return filename

        raise FileExistsError(errno.EEXIST, "No usable temporary filename found")

    def should_redirect_logs(self):
        # Preferred: thread the setting explicitly via RayParams.log_to_stderr.
        # This avoids relying on process-global environment variables.
        if getattr(self._ray_params, "log_to_stderr", None) is not None:
            return not self._ray_params.log_to_stderr

        # Deprecated (kept for backward compatibility): RayParams.redirect_output.
        redirect_output = self._ray_params.redirect_output
        if redirect_output is not None:
            return redirect_output

            # Fall back to stderr redirect environment variable.
        return (
            os.environ.get(ray_constants.LOGGING_REDIRECT_STDERR_ENVIRONMENT_VARIABLE)
            != "1"
        )

    # TODO(hjiang): Re-implement the logic in C++, and expose via cython.
    def get_log_file_names(
        self,
        name: str,
        unique: bool = False,
        create_out: bool = True,
        create_err: bool = True,
    ) -> Tuple[Optional[str], Optional[str]]:
        """Get filename to dump logs for stdout and stderr, with no files opened.
        If output redirection has been disabled, no files will
        be opened and `(None, None)` will be returned.

        Args:
            name: descriptive string for this log file.
            unique: if true, a counter will be attached to `name` to
                ensure the returned filename is not already used.
            create_out: if True, create a .out file.
            create_err: if True, create a .err file.

        Returns:
            A tuple of two file handles for redirecting optional (stdout, stderr),
            or `(None, None)` if output redirection is disabled.
        """
        if not self.should_redirect_logs():
            return None, None

        log_stdout = None
        log_stderr = None

        if create_out:
            log_stdout = self._get_log_file_name(name, "out", unique=unique)
        if create_err:
            log_stderr = self._get_log_file_name(name, "err", unique=unique)
        return log_stdout, log_stderr

    def get_log_file_handles(
        self,
        name: str,
        unique: bool = False,
        create_out: bool = True,
        create_err: bool = True,
    ) -> Tuple[Optional[IO[AnyStr]], Optional[IO[AnyStr]]]:
        """Open log files with partially randomized filenames, returning the
        file handles. If output redirection has been disabled, no files will
        be opened and `(None, None)` will be returned.

        Args:
            name: descriptive string for this log file.
            unique: if true, a counter will be attached to `name` to
                ensure the returned filename is not already used.
            create_out: if True, create a .out file.
            create_err: if True, create a .err file.

        Returns:
            A tuple of two file handles for redirecting optional (stdout, stderr),
            or `(None, None)` if output redirection is disabled.
        """
        log_stdout_fname, log_stderr_fname = self.get_log_file_names(
            name, unique=unique, create_out=create_out, create_err=create_err
        )
        log_stdout = None if log_stdout_fname is None else open_log(log_stdout_fname)
        log_stderr = None if log_stderr_fname is None else open_log(log_stderr_fname)
        return log_stdout, log_stderr

    def _get_log_file_name(
        self,
        name: str,
        suffix: str,
        unique: bool = False,
    ) -> str:
        """Generate partially randomized filenames for log files.

        Args:
            name: descriptive string for this log file.
            suffix: suffix of the file. Usually it is .out of .err.
            unique: if true, a counter will be attached to `name` to
                ensure the returned filename is not already used.

        Returns:
            A tuple of two file names for redirecting (stdout, stderr).
        """
        # strip if the suffix is something like .out.
        suffix = suffix.strip(".")

        if unique:
            filename = self._make_inc_temp(
                suffix=f".{suffix}", prefix=name, directory_name=self._logs_dir
            )
        else:
            filename = os.path.join(self._logs_dir, f"{name}.{suffix}")
        return filename

    def _get_unused_port(self, allocated_ports=None):
        if allocated_ports is None:
            allocated_ports = set()

        s = socket.socket(
            socket.AF_INET6 if is_ipv6(self._node_ip_address) else socket.AF_INET,
            socket.SOCK_STREAM,
        )
        s.bind(("", 0))
        port = s.getsockname()[1]

        # Try to generate a port that is far above the 'next available' one.
        # This solves issue #8254 where GRPC fails because the port assigned
        # from this method has been used by a different process.
        for _ in range(ray_constants.NUM_PORT_RETRIES):
            new_port = random.randint(port, 65535)
            if new_port in allocated_ports:
                # This port is allocated for other usage already,
                # so we shouldn't use it even if it's not in use right now.
                continue
            new_s = socket.socket(
                socket.AF_INET6 if is_ipv6(self._node_ip_address) else socket.AF_INET,
                socket.SOCK_STREAM,
            )
            try:
                new_s.bind(("", new_port))
            except OSError:
                new_s.close()
                continue
            s.close()
            new_s.close()
            return new_port
        logger.error("Unable to succeed in selecting a random port.")
        s.close()
        return port

    def _prepare_socket_file(self, socket_path: str, default_prefix: str):
        """Prepare the socket file for raylet and plasma.

        This method helps to prepare a socket file.
        1. Make the directory if the directory does not exist.
        2. If the socket file exists, do nothing (this just means we aren't the
           first worker on the node).

        Args:
            socket_path: the socket file to prepare.
            default_prefix: the filename prefix to use when ``socket_path`` is
                ``None`` and a new socket path needs to be generated.

        Returns:
            The resolved socket path string.
        """
        result = socket_path
        if sys.platform == "win32":
            if socket_path is None:
                result = (
                    f"tcp://{build_address(self._localhost, self._get_unused_port())}"
                )
        else:
            if socket_path is None:
                result = self._make_inc_temp(
                    prefix=default_prefix, directory_name=self._sockets_dir
                )
            else:
                try_to_create_directory(os.path.dirname(socket_path))

            validate_socket_filepath(result.split("://", 1)[-1])
        return result

    def start_reaper_process(self):
        """
        Start the reaper process.

        This must be the first process spawned and should only be called when
        ray processes should be cleaned up if this process dies.
        """
        assert (
            not self.kernel_fate_share
        ), "a reaper should not be used with kernel fate-sharing"
        process_info = ray._private.services.start_reaper(fate_share=False)
        assert ray_constants.PROCESS_TYPE_REAPER not in self.all_processes
        if process_info is not None:
            self.all_processes[ray_constants.PROCESS_TYPE_REAPER] = [
                process_info,
            ]

    def start_log_monitor(self):
        """Start the log monitor."""
        stdout_log_fname, stderr_log_fname = self.get_log_file_names(
            "log_monitor", unique=True, create_out=True, create_err=True
        )
        process_info = ray._private.services.start_log_monitor(
            self.get_session_dir_path(),
            self._logs_dir,
            self.gcs_address,
            self._node_ip_address,
            fate_share=self.kernel_fate_share,
            max_bytes=self.max_bytes,
            backup_count=self.backup_count,
            stdout_filepath=stdout_log_fname,
            stderr_filepath=stderr_log_fname,
        )
        assert ray_constants.PROCESS_TYPE_LOG_MONITOR not in self.all_processes
        self.all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] = [
            process_info,
        ]

    def start_api_server(
        self, *, include_dashboard: Optional[bool], raise_on_failure: bool
    ):
        """Start the dashboard.

        Args:
            include_dashboard: If true, this will load all dashboard-related modules
                when starting the API server. Otherwise, it will only
                start the modules that are not relevant to the dashboard.
            raise_on_failure: If true, this will raise an exception
                if we fail to start the API server. Otherwise it will print
                a warning if we fail to start the API server.
        """
        stdout_log_fname, stderr_log_fname = self.get_log_file_names(
            "dashboard", unique=True, create_out=True, create_err=True
        )
        self._webui_url, process_info = ray._private.services.start_api_server(
            include_dashboard,
            raise_on_failure,
            self._ray_params.dashboard_host,
            self.gcs_address,
            self.cluster_id.hex(),
            self._node_ip_address,
            self.temp_dir,
            self._logs_dir,
            self._session_dir,
            port=self._ray_params.dashboard_port,
            fate_share=self.kernel_fate_share,
            max_bytes=self.max_bytes,
            backup_count=self.backup_count,
            stdout_filepath=stdout_log_fname,
            stderr_filepath=stderr_log_fname,
            proxy_server_url=self._ray_params.proxy_server_url,
        )
        assert ray_constants.PROCESS_TYPE_DASHBOARD not in self.all_processes
        if process_info is not None:
            self.all_processes[ray_constants.PROCESS_TYPE_DASHBOARD] = [
                process_info,
            ]
            self.get_gcs_client().internal_kv_put(
                b"webui:url",
                self._webui_url.encode(),
                True,
                ray_constants.KV_NAMESPACE_DASHBOARD,
            )

    def start_gcs_server(self):
        """Start the gcs server."""
        assert self._ray_params.gcs_server_port >= 0
        assert self._gcs_address is None, "GCS server is already running."
        assert self._gcs_client is None, "GCS client is already connected."

        stdout_log_fname, stderr_log_fname = self.get_log_file_names(
            "gcs_server", unique=True, create_out=True, create_err=True
        )
        process_info = ray._private.services.start_gcs_server(
            self.redis_address,
            log_dir=self._logs_dir,
            stdout_filepath=stdout_log_fname,
            stderr_filepath=stderr_log_fname,
            session_name=self.session_name,
            redis_username=self._ray_params.redis_username,
            redis_password=self._ray_params.redis_password,
            config=self._config,
            fate_share=self.kernel_fate_share,
            gcs_server_port=self._ray_params.gcs_server_port,
            metrics_agent_port=self._ray_params.metrics_agent_port,
            node_ip_address=self._node_ip_address,
            session_dir=self._session_dir,
            node_id=self._node_id,
        )
        assert ray_constants.PROCESS_TYPE_GCS_SERVER not in self.all_processes
        self.all_processes[ray_constants.PROCESS_TYPE_GCS_SERVER] = [
            process_info,
        ]

        if self._ray_params.gcs_server_port == 0:
            # The GCS port file is published only at the very end of
            # startup: GcsServer binds its RPC port after GcsInitData
            # loads the full persisted table set (GetAll over the job,
            # node, actor, actor-task-spec, and placement-group tables)
            # and every manager is constructed. With the RocksDB backend
            # this is meaningfully slower than in-memory -- two on-disk
            # DB instances are opened (column-family discovery, WAL
            # replay, manifest read) and those table scans hit disk, the
            # latter growing with cluster state on restart recovery (the
            # port file lives on ephemeral /tmp, so a restarted head
            # re-enters this path over an existing store). The default
            # 30s wait can elapse before the port is bound, so bump it to
            # 120s when the GCS backend is rocksdb; operator override via
            # RAY_gcs_server_port_wait_time_s.
            default_wait = "120" if self._is_rocksdb_gcs() else "30"
            gcs_port_wait_s = int(
                os.environ.get("RAY_gcs_server_port_wait_time_s", default_wait)
            )
            self._ray_params.gcs_server_port = wait_for_persisted_port(
                self._session_dir,
                self._node_id,
                GCS_SERVER_PORT_NAME,
                timeout_ms=gcs_port_wait_s * 1000,
            )

        # Connecting via non-localhost address may be blocked by firewall rule,
        # e.g. https://github.com/ray-project/ray/issues/15780
        # TODO(mwtian): figure out a way to use 127.0.0.1 for local connection
        # when possible.
        self._gcs_address = build_address(
            self._node_ip_address, self._ray_params.gcs_server_port
        )

    def start_raylet(
        self,
        plasma_directory: str,
        fallback_directory: str,
        object_store_memory: int,
        use_valgrind: bool = False,
        use_profiler: bool = False,
    ):
        """Start the raylet.

        Args:
            plasma_directory: Filesystem directory backing the plasma store's
                shared memory.
            fallback_directory: Directory used by plasma when memory mapped
                files cannot be created in ``plasma_directory``.
            object_store_memory: Maximum number of bytes the object store can
                use.
            use_valgrind: True if we should start the process in
                valgrind.
            use_profiler: True if we should start the process in the
                valgrind profiler.
        """
        raylet_stdout_filepath, raylet_stderr_filepath = self.get_log_file_names(
            ray_constants.PROCESS_TYPE_RAYLET,
            unique=True,
            create_out=True,
            create_err=True,
        )
        (
            dashboard_agent_stdout_filepath,
            dashboard_agent_stderr_filepath,
        ) = self.get_log_file_names(
            ray_constants.PROCESS_TYPE_DASHBOARD_AGENT,
            unique=True,
            create_out=True,
            create_err=True,
        )
        (
            runtime_env_agent_stdout_filepath,
            runtime_env_agent_stderr_filepath,
        ) = self.get_log_file_names(
            ray_constants.PROCESS_TYPE_RUNTIME_ENV_AGENT,
            unique=True,
            create_out=True,
            create_err=True,
        )

        dashboard_agent_log_filepath = None
        if dashboard_agent_stdout_filepath is not None:
            dashboard_agent_log_filepath = self._get_log_file_name(
                ray_constants.PROCESS_TYPE_DASHBOARD_AGENT, "log", unique=True
            )

        runtime_env_agent_log_filepath = None
        if runtime_env_agent_stdout_filepath is not None:
            runtime_env_agent_log_filepath = self._get_log_file_name(
                ray_constants.PROCESS_TYPE_RUNTIME_ENV_AGENT, "log", unique=True
            )

        self.resource_isolation_config.add_system_pids(
            self._get_system_processes_for_resource_isolation()
        )

        process_info = ray._private.services.start_raylet(
            self.redis_address,
            self.gcs_address,
            self._node_id,
            self._node_ip_address,
            self._ray_params.node_manager_port,
            self._raylet_socket_name,
            self._plasma_store_socket_name,
            self.cluster_id.hex(),
            self._ray_params.worker_path,
            self._ray_params.setup_worker_path,
            self.temp_dir,
            self._session_dir,
            self._runtime_env_dir,
            self._logs_dir,
            self.get_resource_and_label_spec(),
            plasma_directory,
            fallback_directory,
            object_store_memory,
            self.session_name,
            is_head_node=self.is_head(),
            min_worker_port=self._ray_params.min_worker_port,
            max_worker_port=self._ray_params.max_worker_port,
            worker_port_list=self._ray_params.worker_port_list,
            object_manager_port=self._ray_params.object_manager_port,
            redis_username=self._ray_params.redis_username,
            redis_password=self._ray_params.redis_password,
            metrics_agent_port=self._ray_params.metrics_agent_port,
            runtime_env_agent_port=self._ray_params.runtime_env_agent_port,
            metrics_export_port=self._ray_params.metrics_export_port,
            dashboard_agent_listen_port=self._ray_params.dashboard_agent_listen_port,
            use_valgrind=use_valgrind,
            use_profiler=use_profiler,
            raylet_stdout_filepath=raylet_stdout_filepath,
            raylet_stderr_filepath=raylet_stderr_filepath,
            dashboard_agent_stdout_filepath=dashboard_agent_stdout_filepath,
            dashboard_agent_stderr_filepath=dashboard_agent_stderr_filepath,
            dashboard_agent_log_filepath=dashboard_agent_log_filepath,
            runtime_env_agent_stdout_filepath=runtime_env_agent_stdout_filepath,
            runtime_env_agent_stderr_filepath=runtime_env_agent_stderr_filepath,
            runtime_env_agent_log_filepath=runtime_env_agent_log_filepath,
            huge_pages=self._ray_params.huge_pages,
            fate_share=self.kernel_fate_share,
            socket_to_use=None,
            max_bytes=self.max_bytes,
            backup_count=self.backup_count,
            ray_debugger_external=self._ray_params.ray_debugger_external,
            env_updates=self._ray_params.env_vars,
            node_name=self._ray_params.node_name,
            webui=self._webui_url,
            resource_isolation_config=self.resource_isolation_config,
        )
        assert ray_constants.PROCESS_TYPE_RAYLET not in self.all_processes
        self.all_processes[ray_constants.PROCESS_TYPE_RAYLET] = [process_info]

    def start_monitor(self):
        """Start the monitor.

        Autoscaling output goes to these monitor.err/out files, and
        any modification to these files may break existing
        cluster launching commands.
        """
        from ray.autoscaler.v2.utils import is_autoscaler_v2

        stdout_log_fname, stderr_log_fname = self.get_log_file_names(
            "monitor", unique=True, create_out=True, create_err=True
        )
        process_info = ray._private.services.start_monitor(
            self.gcs_address,
            self._logs_dir,
            stdout_filepath=stdout_log_fname,
            stderr_filepath=stderr_log_fname,
            autoscaling_config=self._ray_params.autoscaling_config,
            fate_share=self.kernel_fate_share,
            max_bytes=self.max_bytes,
            backup_count=self.backup_count,
            monitor_ip=self._node_ip_address,
            autoscaler_v2=is_autoscaler_v2(fetch_from_server=True),
        )
        assert ray_constants.PROCESS_TYPE_MONITOR not in self.all_processes
        self.all_processes[ray_constants.PROCESS_TYPE_MONITOR] = [process_info]

    def start_ray_client_server(self):
        """Start the ray client server process."""
        stdout_file, stderr_file = self.get_log_file_handles(
            "ray_client_server", unique=True
        )
        process_info = ray._private.services.start_ray_client_server(
            self.address,
            self._node_ip_address,
            self._ray_params.ray_client_server_port,
            stdout_file=stdout_file,
            stderr_file=stderr_file,
            redis_username=self._ray_params.redis_username,
            redis_password=self._ray_params.redis_password,
            fate_share=self.kernel_fate_share,
            runtime_env_agent_address=self.runtime_env_agent_address,
            node_id=self._node_id,
        )
        assert ray_constants.PROCESS_TYPE_RAY_CLIENT_SERVER not in self.all_processes
        self.all_processes[ray_constants.PROCESS_TYPE_RAY_CLIENT_SERVER] = [
            process_info
        ]

    def _write_cluster_info_to_kv(self):
        """Write the cluster metadata to GCS.
        Cluster metadata is always recorded, but they are
        not reported unless usage report is enabled.
        Check `usage_stats_head.py` for more details.
        """
        # Make sure the cluster metadata wasn't reported before.
        import ray._common.usage.usage_lib as ray_usage_lib

        ray_usage_lib.put_cluster_metadata(
            self.get_gcs_client(), ray_init_cluster=self.ray_init_cluster
        )
        # On restart with the RocksDB GCS backend, check_persisted_session_name()
        # needs the previous session_name before GCS (and thus internal_kv) is
        # back up. We bridge that gap by also writing session_name to a plain
        # file in the GCS storage directory. Only done for the RocksDB backend;
        # others are unaffected.
        #
        # Write the file BEFORE the internal_kv_put below so the two never
        # disagree: if we crash in between, the next restart reads the file and
        # the internal_kv_put inserts cleanly. A write failure here is fatal --
        # the storage path also holds RocksDB's own files, so an unwritable path
        # means GCS can't run anyway.
        if self._is_rocksdb_gcs():
            self._persist_rocksdb_session_name_file()

        # Make sure GCS is up.
        added = self.get_gcs_client().internal_kv_put(
            b"session_name",
            self._session_name.encode(),
            False,
            ray_constants.KV_NAMESPACE_SESSION,
        )
        if not added:
            curr_val = self.get_gcs_client().internal_kv_get(
                b"session_name", ray_constants.KV_NAMESPACE_SESSION
            )
            assert curr_val == self._session_name.encode("utf-8"), (
                f"Session name {self._session_name} does not match "
                f"persisted value {curr_val}. Perhaps there was an "
                f"error connecting to the GCS storage backend."
            )

        # Add tracing_startup_hook to redis / internal kv manually
        # since internal kv is not yet initialized.
        if self._ray_params.tracing_startup_hook:
            self.get_gcs_client().internal_kv_put(
                b"tracing_startup_hook",
                self._ray_params.tracing_startup_hook.encode(),
                True,
                ray_constants.KV_NAMESPACE_TRACING,
            )

    def start_head_processes(self):
        """Start head processes on the node."""
        logger.debug(
            f"Process STDOUT and STDERR is being " f"redirected to {self._logs_dir}."
        )
        assert self._gcs_address is None
        assert self._gcs_client is None

        self.start_gcs_server()
        assert self.get_gcs_client() is not None
        self._write_cluster_info_to_kv()

        if not self._ray_params.no_monitor:
            self.start_monitor()

        if self._ray_params.ray_client_server_port:
            self.start_ray_client_server()

        if self._ray_params.include_dashboard is None:
            # Default
            raise_on_api_server_failure = False
        else:
            raise_on_api_server_failure = self._ray_params.include_dashboard

        self.start_api_server(
            include_dashboard=self._ray_params.include_dashboard,
            raise_on_failure=raise_on_api_server_failure,
        )

    def start_ray_processes(self):
        """Start all of the processes on the node."""
        logger.debug(
            f"Process STDOUT and STDERR is being " f"redirected to {self._logs_dir}."
        )

        if not self.head:
            # Get the system config from GCS first if this is a non-head node.
            gcs_options = ray._raylet.GcsClientOptions.create(
                self.gcs_address,
                self.cluster_id.hex(),
                allow_cluster_id_nil=False,
                fetch_cluster_id_if_nil=False,
            )
            global_state = ray._private.state.GlobalState()
            global_state._initialize_global_state(gcs_options)
            new_config = global_state.get_system_config()
            assert self._config.items() <= new_config.items(), (
                "The system config from GCS is not a superset of the local"
                " system config. There might be a configuration inconsistency"
                " issue between the head node and non-head nodes."
                f" Local system config: {self._config},"
                f" GCS system config: {new_config}"
            )

            # Note: We decide which object spilling directory to use based on the following policy:
            # 1. If this node specifies an object spilling directory, use it.
            # 2. If the head node specifies an object spilling directory, and the worker node doesn't specify one,
            #    use the head node's object spilling directory.
            # 3. If the head node doesn't specify an object spilling directory, and the worker node doesn't specify one,
            #    use the temp_dir of the worker node as the object spilling directory.
            try:
                if new_config["automatic_object_spilling_enabled"]:
                    config = json.loads(new_config["object_spilling_config"])
                    if config.get("type") == "filesystem":
                        fetched_config_directory_path = config["params"][
                            "directory_path"
                        ]
                        head_spill_directory_path = None
                        if isinstance(fetched_config_directory_path, list):
                            head_spill_directory_path = fetched_config_directory_path[0]
                        elif isinstance(fetched_config_directory_path, str):
                            head_spill_directory_path = fetched_config_directory_path
                        if head_spill_directory_path is None:
                            raise ValueError(
                                f"Failed to obtain spill directory path from "
                                f"object spilling config: {fetched_config_directory_path}"
                            )

                        if (
                            self._fallback_directory != self._session_dir
                            or head_spill_directory_path == self._head_session_dir
                        ):
                            config["params"][
                                "directory_path"
                            ] = self._fallback_directory
                            new_config["object_spilling_config"] = json.dumps(config)
                        else:
                            self._fallback_directory = head_spill_directory_path

                        try_to_create_directory(self._fallback_directory)
                self._config = new_config
            except Exception as e:
                raise Exception(
                    "Expected valid object_spilling_config to be received from head node "
                    f"but got: {repr(e)}"
                )

        # Make sure we don't call `determine_plasma_store_config` multiple
        # times to avoid printing multiple warnings.
        resource_and_label_spec = self.get_resource_and_label_spec()
        if resource_and_label_spec.labels.get(
            ray._raylet.RAY_NODE_ACCELERATOR_TYPE_KEY
        ):
            from ray._common.usage import usage_lib

            usage_lib.record_hardware_usage(
                resource_and_label_spec.labels.get(
                    ray._raylet.RAY_NODE_ACCELERATOR_TYPE_KEY
                )
            )

        (
            plasma_directory,
            fallback_directory,
            object_store_memory,
        ) = ray._private.services.determine_plasma_store_config(
            resource_and_label_spec.object_store_memory,
            self.temp_dir,
            plasma_directory=self._ray_params.plasma_directory,
            fallback_directory=self._fallback_directory,
            huge_pages=self._ray_params.huge_pages,
        )

        if self._ray_params.include_log_monitor:
            self.start_log_monitor()

        self.start_raylet(plasma_directory, fallback_directory, object_store_memory)

    def _get_system_processes_for_resource_isolation(self) -> str:
        """Returns a list of system processes that will be isolated by raylet.

        NOTE: If a new system process is started before the raylet starts up, it needs to be
        added to self.all_processes so it can be moved into the raylet's managed cgroup
        hierarchy.
        """
        system_process_pids = [
            str(p[0].process.pid) for p in self.all_processes.values()
        ]

        # If the dashboard api server was started on the head node, then include all of the api server's
        # child processes.
        if ray_constants.PROCESS_TYPE_DASHBOARD in self.all_processes:
            dashboard_pid = self.all_processes[ray_constants.PROCESS_TYPE_DASHBOARD][
                0
            ].process.pid
            dashboard_process = psutil.Process(dashboard_pid)
            system_process_pids += [str(p.pid) for p in dashboard_process.children()]

        return ",".join(system_process_pids)

    def _kill_process_type(
        self,
        process_type: str,
        allow_graceful: bool = False,
        check_alive: bool = True,
        wait: bool = False,
    ):
        """Kill a process of a given type.

        If the process type is PROCESS_TYPE_REDIS_SERVER, then we will kill all
        of the Redis servers.

        If the process was started in valgrind, then we will raise an exception
        if the process has a non-zero exit code.

        Args:
            process_type: The type of the process to kill.
            allow_graceful: Send a SIGTERM first and give the process
                time to exit gracefully. If that doesn't work, then use
                SIGKILL. We usually want to do this outside of tests.
            check_alive: If true, then we expect the process to be alive
                and will raise an exception if the process is already dead.
            wait: If true, then this method will not return until the
                process in question has exited.

        Raises:
            This process raises an exception in the following cases:
                1. The process had already died and check_alive is true.
                2. The process had been started in valgrind and had a non-zero
                   exit code.
        """

        # Ensure thread safety
        with self.removal_lock:
            self._kill_process_impl(
                process_type,
                allow_graceful=allow_graceful,
                check_alive=check_alive,
                wait=wait,
            )

    def _kill_process_impl(
        self, process_type, allow_graceful=False, check_alive=True, wait=False
    ):
        """See `_kill_process_type`."""
        if process_type not in self.all_processes:
            return
        process_infos = self.all_processes[process_type]
        if process_type != ray_constants.PROCESS_TYPE_REDIS_SERVER:
            assert len(process_infos) == 1
        wait_timeout_seconds = 1
        for process_info in process_infos:
            process = process_info.process
            # Handle the case where the process has already exited.
            if process.poll() is not None:
                if check_alive:
                    raise RuntimeError(
                        "Attempting to kill a process of type "
                        f"'{process_type}', but this process is already dead."
                    )
                else:
                    continue

            if process_info.use_valgrind:
                process.terminate()
                process.wait()
                if process.returncode != 0:
                    message = (
                        "Valgrind detected some errors in process of "
                        f"type {process_type}. Error code {process.returncode}."
                    )
                    if process_info.stdout_file is not None:
                        with open(process_info.stdout_file, "r") as f:
                            message += "\nPROCESS STDOUT:\n" + f.read()
                    if process_info.stderr_file is not None:
                        with open(process_info.stderr_file, "r") as f:
                            message += "\nPROCESS STDERR:\n" + f.read()
                    raise RuntimeError(message)
                continue

            if process_info.use_valgrind_profiler:
                # Give process signal to write profiler data.
                os.kill(process.pid, signal.SIGINT)
                # Wait for profiling data to be written.
                time.sleep(0.1)

            if allow_graceful:
                process.terminate()
                # Allow the process one second to exit gracefully.
                try:
                    process.wait(timeout=wait_timeout_seconds)
                except subprocess.TimeoutExpired:
                    pass

            # If the process did not exit, force kill it.
            if process.poll() is None:
                process.kill()
                # After kill, wait must be called
                # The reason we usually don't set timeout=None here is that
                # there's some chance we'd end up waiting a really long time.
                try:
                    process.wait(timeout=None if wait else wait_timeout_seconds)
                except subprocess.TimeoutExpired:
                    pass

        del self.all_processes[process_type]

    def kill_redis(self, check_alive: bool = True):
        """Kill the Redis servers.

        Args:
            check_alive: Raise an exception if any of the processes
                were already dead.
        """
        self._kill_process_type(
            ray_constants.PROCESS_TYPE_REDIS_SERVER, check_alive=check_alive
        )

    def kill_raylet(self, check_alive: bool = True):
        """Kill the raylet.

        Args:
            check_alive: Raise an exception if the process was already
                dead.
        """
        self._kill_process_type(
            ray_constants.PROCESS_TYPE_RAYLET, check_alive=check_alive
        )

    def kill_log_monitor(self, check_alive: bool = True):
        """Kill the log monitor.

        Args:
            check_alive: Raise an exception if the process was already
                dead.
        """
        self._kill_process_type(
            ray_constants.PROCESS_TYPE_LOG_MONITOR, check_alive=check_alive
        )

    def kill_dashboard(self, check_alive: bool = True):
        """Kill the dashboard.

        Args:
            check_alive: Raise an exception if the process was already
                dead.
        """
        self._kill_process_type(
            ray_constants.PROCESS_TYPE_DASHBOARD, check_alive=check_alive
        )

    def kill_monitor(self, check_alive: bool = True):
        """Kill the monitor.

        Args:
            check_alive: Raise an exception if the process was already
                dead.
        """
        self._kill_process_type(
            ray_constants.PROCESS_TYPE_MONITOR, check_alive=check_alive
        )

    def kill_gcs_server(self, check_alive: bool = True):
        """Kill the gcs server.

        Args:
            check_alive: Raise an exception if the process was already
                dead.
        """
        self._kill_process_type(
            ray_constants.PROCESS_TYPE_GCS_SERVER, check_alive=check_alive, wait=True
        )
        # Clear GCS client and address to indicate no GCS server is running.
        self._gcs_address = None
        self._gcs_client = None

    def kill_reaper(self, check_alive: bool = True):
        """Kill the reaper process.

        Args:
            check_alive: Raise an exception if the process was already
                dead.
        """
        self._kill_process_type(
            ray_constants.PROCESS_TYPE_REAPER, check_alive=check_alive
        )

    def kill_all_processes(
        self,
        check_alive: bool = True,
        allow_graceful: bool = False,
        wait: bool = False,
    ):
        """Kill all of the processes.

        Note that This is slower than necessary because it calls kill, wait,
        kill, wait, ... instead of kill, kill, ..., wait, wait, ...

        Args:
            check_alive: Raise an exception if any of the processes were
                already dead.
            allow_graceful: Send a SIGTERM first and give each process time
                to exit gracefully before falling back to SIGKILL.
            wait: If true, then this method will not return until the
                process in question has exited.
        """
        # Kill the raylet first. This is important for suppressing errors at
        # shutdown because we give the raylet a chance to exit gracefully and
        # clean up its child worker processes. If we were to kill the plasma
        # store (or Redis) first, that could cause the raylet to exit
        # ungracefully, leading to more verbose output from the workers.
        if ray_constants.PROCESS_TYPE_RAYLET in self.all_processes:
            self._kill_process_type(
                ray_constants.PROCESS_TYPE_RAYLET,
                check_alive=check_alive,
                allow_graceful=allow_graceful,
                wait=wait,
            )

        if ray_constants.PROCESS_TYPE_GCS_SERVER in self.all_processes:
            self._kill_process_type(
                ray_constants.PROCESS_TYPE_GCS_SERVER,
                check_alive=check_alive,
                allow_graceful=allow_graceful,
                wait=wait,
            )

        # We call "list" to copy the keys because we are modifying the
        # dictionary while iterating over it.
        for process_type in list(self.all_processes.keys()):
            # Need to kill the reaper process last in case we die unexpectedly
            # while cleaning up.
            if process_type != ray_constants.PROCESS_TYPE_REAPER:
                self._kill_process_type(
                    process_type,
                    check_alive=check_alive,
                    allow_graceful=allow_graceful,
                    wait=wait,
                )

        if ray_constants.PROCESS_TYPE_REAPER in self.all_processes:
            self._kill_process_type(
                ray_constants.PROCESS_TYPE_REAPER,
                check_alive=check_alive,
                allow_graceful=allow_graceful,
                wait=wait,
            )

    def live_processes(self):
        """Return a list of the live processes.

        Returns:
            A list of the live processes.
        """
        result = []
        for process_type, process_infos in self.all_processes.items():
            for process_info in process_infos:
                if process_info.process.poll() is None:
                    result.append((process_type, process_info.process))
        return result

    def dead_processes(self):
        """Return a list of the dead processes.

        Note that this ignores processes that have been explicitly killed,
        e.g., via a command like node.kill_raylet().

        Returns:
            A list of the dead processes ignoring the ones that have been
                explicitly killed.
        """
        result = []
        for process_type, process_infos in self.all_processes.items():
            for process_info in process_infos:
                if process_info.process.poll() is not None:
                    result.append((process_type, process_info.process))
        return result

    def any_processes_alive(self):
        """Return true if any processes are still alive.

        Returns:
            True if any process is still alive.
        """
        return any(self.live_processes())

    def remaining_processes_alive(self):
        """Return true if all remaining processes are still alive.

        Note that this ignores processes that have been explicitly killed,
        e.g., via a command like node.kill_raylet().

        Returns:
            True if any process that wasn't explicitly killed is still alive.
        """
        return not any(self.dead_processes())

    def destroy_external_storage(self):
        object_spilling_config = self._config.get("object_spilling_config", {})
        if object_spilling_config:
            object_spilling_config = json.loads(object_spilling_config)
            from ray._private import external_storage

            storage = external_storage.setup_external_storage(
                object_spilling_config, self._node_id, self._session_name
            )
            storage.destroy_external_storage()

    def validate_external_storage(self):
        """Make sure we can setup the object spilling external storage."""

        automatic_spilling_enabled = self._config.get(
            "automatic_object_spilling_enabled", True
        )
        if not automatic_spilling_enabled:
            return
        self._config["automatic_object_spilling_enabled"] = True

        object_spilling_config = self._object_spilling_config
        # Try setting up the storage.
        # Configure the proper system config.
        # We need to set both ray param's system config and self._config
        # because they could've been diverged at this point.
        deserialized_config = json.loads(object_spilling_config)
        self._ray_params._system_config[
            "object_spilling_config"
        ] = object_spilling_config
        self._config["object_spilling_config"] = object_spilling_config

        is_external_storage_type_fs = deserialized_config["type"] == "filesystem"
        self._ray_params._system_config[
            "is_external_storage_type_fs"
        ] = is_external_storage_type_fs
        self._config["is_external_storage_type_fs"] = is_external_storage_type_fs

        # Validate external storage usage.
        from ray._private import external_storage

        # Node ID is available only after GCS is connected. However,
        # validate_external_storage() needs to be called before it to
        # be able to validate the configs early. Therefore, we use a
        # dummy node ID here and make sure external storage can be set
        # up based on the provided config. This storage is destroyed
        # right after the validation.
        dummy_node_id = ray.NodeID.from_random().hex()
        storage = external_storage.setup_external_storage(
            deserialized_config, dummy_node_id, self._session_name
        )
        storage.destroy_external_storage()
        external_storage.reset_external_storage()

    def _get_object_spilling_config(self):
        """Consolidate the object spilling config from the ray params, environment
        variable, and system config. The object spilling directory specified through
        ray params will override the one specified through environment variable and
        system config."""

        object_spilling_directory = self._ray_params.object_spilling_directory
        if not object_spilling_directory:
            object_spilling_directory = self._config.get(
                "object_spilling_directory", ""
            )

        if not object_spilling_directory:
            object_spilling_directory = os.environ.get(
                "RAY_object_spilling_directory", ""
            )

        if object_spilling_directory:
            return json.dumps(
                {
                    "type": "filesystem",
                    "params": {"directory_path": object_spilling_directory},
                }
            )

        object_spilling_config = self._config.get("object_spilling_config", {})
        if not object_spilling_config:
            object_spilling_config = os.environ.get("RAY_object_spilling_config", "")

        # If the config is not specified in ray params, system config or environment
        # variable, we fill up the default.
        if not object_spilling_config:
            object_spilling_config = json.dumps(
                {"type": "filesystem", "params": {"directory_path": self._session_dir}}
            )
        else:
            if not is_in_test():
                logger.warning(
                    "The object spilling config is specified from an unstable "
                    "API - system config or environment variable. This is "
                    "subject to change in the future. You can use the stable "
                    "API - --object-spilling-directory in ray start or "
                    "object_spilling_directory in ray.init() to specify the "
                    "object spilling directory instead. If you need more "
                    "advanced settings, please open a github issue with the "
                    "Ray team."
                )

        return object_spilling_config

    def _record_stats(self):
        # This is only called when a new node is started.
        # Initialize the internal kv so that the metrics can be put
        from ray._common.usage.usage_lib import (
            TagKey,
            record_extra_usage_tag,
            record_hardware_usage,
        )

        if not ray.experimental.internal_kv._internal_kv_initialized():
            ray.experimental.internal_kv._initialize_internal_kv(self.get_gcs_client())
        assert ray.experimental.internal_kv._internal_kv_initialized()
        if self.head:
            # record head node stats
            if self._is_rocksdb_gcs():
                gcs_storage_type = "rocksdb"
            elif os.environ.get("RAY_REDIS_ADDRESS") is not None:
                gcs_storage_type = "redis"
            else:
                gcs_storage_type = "memory"
            record_extra_usage_tag(TagKey.GCS_STORAGE, gcs_storage_type)
        cpu_model_name = ray._private.utils.get_current_node_cpu_model_name()
        if cpu_model_name:
            # CPU model name can be an arbitrary long string
            # so we truncate it to the first 50 characters
            # to avoid any issues.
            record_hardware_usage(cpu_model_name[:50])