code: def test_replica_spread(ray_cluster):
      cluster = ray_cluster
  
      cluster.add_node(num_cpus=2)
  
      # NOTE(edoakes): we need to start serve before adding the worker node to
      # guarantee that the controller is placed on the head node (we should be
      # able to tolerate being placed on workers, but there's currently a bug).
      # We should add an explicit test for that in the future when it's fixed.
      cluster.connect(namespace=SERVE_NAMESPACE)
      serve.start()
  
      worker_node = cluster.add_node(num_cpus=2)
  
      @serve.deployment(
          num_replicas=2,
          health_check_period_s=1,
      )
      def D():
          return "hi"
  
      serve.run(D.bind())
  
      def get_num_nodes():
          client = _get_global_client()
          details = client.get_serve_details()
          dep = details["applications"]["default"]["deployments"]["D"]
          nodes = {r["node_id"] for r in dep["replicas"]}
          print("replica nodes", nodes)
          return len(nodes)
  
      # Check that the two replicas are spread across the two nodes.
      wait_for_condition(lambda: get_num_nodes() == 2)
  
      # Kill the worker node. The second replica should get rescheduled on
      # the head node.
      print("Removing worker node. Replica should be rescheduled.")
      cluster.remove_node(worker_node)
  
      # Check that the replica on the dead node can be rescheduled.
      wait_for_condition(lambda: get_num_nodes() == 1)
  
imports: import os
  import sys
  import time
  from collections import defaultdict
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import Cluster
  from ray.exceptions import RayActorError
  from ray.serve._private.common import DeploymentID, ReplicaState
  from ray.serve._private.constants import SERVE_NAMESPACE
  from ray.serve._private.deployment_state import ReplicaStartupStatus
  from ray.serve._private.utils import get_head_node_id
  from ray.serve.context import _get_global_client
  from ray.serve.handle import RayServeHandle
  from ray.serve.schema import ServeDeploySchema
label_desc: Use image node_id 8 for calling a global function with example usage: test_replica_spread(ray_cluster) and returns: str, len