code: def test_multiple_routers(ray_cluster):
      cluster = ray_cluster
      head_node = cluster.add_node(num_cpus=4)
      cluster.add_node(num_cpus=4)
  
      ray.init(head_node.address)
      node_ids = ray._private.state.node_ids()
      assert len(node_ids) == 2
      serve.start(http_options=dict(port=8005, location="EveryNode"))
  
      @serve.deployment(
          num_replicas=2,
          ray_actor_options={"num_cpus": 3},
      )
      class A:
          def __call__(self, *args):
              return "hi"
  
      serve.run(A.bind())
  
      gcs_client = GcsClient(
          address=ray.get_runtime_context().gcs_address
      )
      cluster_node_info_cache = create_cluster_node_info_cache(
          gcs_client
      )
      cluster_node_info_cache.update()
  
      def get_proxy_names():
          proxy_names = []
          for node_id in cluster_node_info_cache.get_alive_node_ids():
              proxy_names.append(
                  format_actor_name(
                      SERVE_PROXY_NAME,
                      node_id,
                  )
              )
          return proxy_names
  
      wait_for_condition(lambda: len(get_proxy_names()) == 2)
      original_proxy_names = get_proxy_names()
  
      # Two actors should be started.
      def get_first_two_actors():
          try:
              ray.get_actor(
                  original_proxy_names[0], namespace=SERVE_NAMESPACE
              )
              ray.get_actor(
                  original_proxy_names[1], namespace=SERVE_NAMESPACE
              )
              return True
          except ValueError:
              return False
  
      wait_for_condition(get_first_two_actors)
  
      # Wait for the actors to come up.
      ray.get(
          block_until_http_ready.remote(
              "http://127.0.0.1:8005/-/routes"
          )
      )
  
      # Kill one of the servers, the HTTP server should still function.
      ray.kill(
          ray.get_actor(
              get_proxy_names()[0], namespace=SERVE_NAMESPACE
          ),
          no_restart=True,
      )
      ray.get(
          block_until_http_ready.remote(
              "http://127.0.0.1:8005/-/routes"
          )
      )
  
      # Add a new node to the cluster. This should trigger a new router to get
      # started.
      new_node = cluster.add_node(num_cpus=4)
      cluster_node_info_cache.update()
  
      wait_for_condition(lambda: len(get_proxy_names()) == 3)
      (third_proxy,) = set(get_proxy_names()) - set(
          original_proxy_names
      )
  
      serve.run(A.options(num_replicas=3).bind())
  
      def get_third_actor():
          try:
              ray.get_actor(third_proxy, namespace=SERVE_NAMESPACE)
              return True
          # IndexErrors covers when cluster resources aren't updated yet.
          except (IndexError, ValueError):
              return False
  
      wait_for_condition(get_third_actor)
  
      # Remove the newly-added node from the cluster. The corresponding actor
      # should be removed as well.
      cluster.remove_node(new_node)
      cluster_node_info_cache.update()
  
      def third_actor_removed():
          try:
              ray.get_actor(third_proxy, namespace=SERVE_NAMESPACE)
              return False
          except ValueError:
              return True
  
      # Check that the actor is gone and the HTTP server still functions.
      wait_for_condition(third_actor_removed)
      ray.get(
          block_until_http_ready.remote(
              "http://127.0.0.1:8005/-/routes"
          )
      )
  
imports: import logging
  import os
  import socket
  import sys
  import time
  import pytest
  import requests
  import ray
  import ray._private.gcs_utils
  from ray import serve
  from ray._private.services import new_port
  from ray._private.test_utils import convert_actor_state, run_string_as_driver, wait_for_condition
  from ray._raylet import GcsClient
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.serve._private import api
  from ray.serve._private.constants import SERVE_CONTROLLER_NAME, SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE, SERVE_PROXY_NAME, SERVE_ROOT_URL_ENV_KEY
  from ray.serve._private.default_impl import create_cluster_node_info_cache
  from ray.serve._private.http_util import set_socket_reuse_port
  from ray.serve._private.utils import block_until_http_ready, format_actor_name
  from ray.serve.config import DeploymentMode, HTTPOptions, ProxyLocation
  from ray.serve.context import _get_global_client
  from ray.serve.exceptions import RayServeException
  from ray.serve.schema import ServeApplicationSchema, ServeDeploySchema
  from ray.tests.conftest import maybe_external_redis
  from ray.tests.conftest import ray_start_with_dashboard
  from ray.util.state import list_actors
label_desc: Use image node_id 10 for calling a global function with example usage: test_multiple_routers(ray_cluster) and returns: proxy_names, str, True, True, False, False, False, True