code: def test_node_failure(ray_cluster):
      cluster = ray_cluster
  
      cluster.add_node(num_cpus=3)
      cluster.connect(namespace=SERVE_NAMESPACE)
  
      # 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.
      serve.start()
  
      worker_node = cluster.add_node(num_cpus=2)
  
      @serve.deployment(
          version="1",
          num_replicas=5,
          health_check_period_s=1,
          max_concurrent_queries=1,
      )
      def D(*args):
          time.sleep(0.1)
          return os.getpid()
  
      print("Initial deploy.")
      serve.run(D.bind())
      pids1 = get_pids(5)
  
      # Remove the node. There should still be three replicas running.
      print("Kill node.")
      cluster.remove_node(worker_node)
      pids2 = get_pids(3)
      assert pids2.issubset(pids1)
  
      # Add a worker node back. One replica should get placed.
      print("Add back first node.")
      cluster.add_node(num_cpus=1)
      pids3 = get_pids(4)
      assert pids2.issubset(pids3)
  
      # Add another worker node. One more replica should get placed.
      print("Add back second node.")
      cluster.add_node(num_cpus=1)
      pids4 = get_pids(5)
      assert pids3.issubset(pids4)
  
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 4 for calling a global function with example usage: test_node_failure(ray_cluster) and returns: os