code: def test_grpc_proxy_on_draining_nodes(ray_cluster):
      """Test gRPC request on the draining node.
  
      When there are no replicas on head node and some replicas on the worker node, the
      ListApplications and Healthz methods should respond successfully. When there are
      no replicas on any nodes, ListApplications and Healthz methods should continue to
      succeeding on the head node. But should return draining response on the worker node.
  
      Also note, this is to ensure the previous fix to serve downscaling also applies to
      gRPC proxy. Head node will not need to be downscaled and never be in the draining
      state. Worker nodes will be in draining when there is no replicas. We will fail the
      health check in this case, so ALB knows not to route to this node anymore.
      """
      head_node_grpc_port = 9000
      worker_node_grpc_port = 9001
  
      # Setup worker gRPC proxy to be pointing to port 9001. Head node gRPC proxy will
      # continue to be pointing to the default port 9000.
      os.environ["TEST_WORKER_NODE_GRPC_PORT"] = str(
          worker_node_grpc_port
      )
  
      # Set up a cluster with 2 nodes.
      cluster = Cluster()
      cluster.add_node(num_cpus=0)
      cluster.add_node(num_cpus=2)
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      # Start serve with gRPC proxy
      grpc_servicer_functions = [
          "ray.serve.generated.serve_pb2_grpc.add_UserDefinedServiceServicer_to_server",
          "ray.serve.generated.serve_pb2_grpc.add_FruitServiceServicer_to_server",
      ]
      serve.start(
          http_options={"location": "EveryNode"},
          grpc_options=gRPCOptions(
              port=head_node_grpc_port,
              grpc_servicer_functions=grpc_servicer_functions,
          ),
      )
  
      # Deploy 2 replicas, both should be on the worker node.
      @serve.deployment(num_replicas=2)
      class HelloModel:
          def __call__(self):
              return serve_pb2.UserDefinedResponse(greeting="hello")
  
      model = HelloModel.bind()
      app_name = "app1"
      serve.run(target=model, name=app_name)
  
      # Ensure worker node has both replicas.
      def check_replicas_on_worker_nodes():
          _actors = ray._private.state.actors().values()
          replica_nodes = [
              a["Address"]["NodeID"]
              for a in _actors
              if a["ActorClassName"].startswith("ServeReplica")
          ]
          return len(set(replica_nodes)) == 1
  
      wait_for_condition(check_replicas_on_worker_nodes)
  
      # Ensure total actors of 2 proxies, 1 controller, and 2 replicas, and 2 nodes exist.
      wait_for_condition(lambda: len(ray._private.state.actors()) == 5)
      assert len(ray.nodes()) == 2
  
      # Set up gRPC channels.
      head_node_channel = grpc.insecure_channel(
          f"localhost:{head_node_grpc_port}"
      )
      worker_node_channel = grpc.insecure_channel(
          f"localhost:{worker_node_grpc_port}"
      )
  
      # Ensures ListApplications method on the head node is succeeding.
      wait_for_condition(
          ping_grpc_list_applications,
          channel=head_node_channel,
          app_names=[app_name],
      )
  
      # Ensures Healthz method on the head node is succeeding.
      ping_grpc_healthz(head_node_channel)
  
      # Ensures ListApplications method on the worker node is succeeding.
      wait_for_condition(
          ping_grpc_list_applications,
          channel=worker_node_channel,
          app_names=[app_name],
          timeout=30,
      )
  
      # Ensures Healthz method on the worker node is succeeding.
      ping_grpc_healthz(worker_node_channel)
  
      # Delete the deployment should bring the active actors down to 3 and drop
      # replicas on all nodes.
      serve.delete(name=app_name)
  
      def _check():
          _actors = ray._private.state.actors().values()
          return (
              len(
                  list(
                      filter(
                          lambda a: a["State"] == "ALIVE",
                          _actors,
                      )
                  )
              )
              == 3
          )
  
      wait_for_condition(_check)
  
      # Ensures ListApplications method on the head node is succeeding.
      wait_for_condition(
          ping_grpc_list_applications,
          channel=head_node_channel,
          app_names=[],
      )
  
      # Ensures Healthz method on the head node is succeeding.
      ping_grpc_healthz(head_node_channel)
  
      # Ensures ListApplications method on the worker node is draining.
      wait_for_condition(
          ping_grpc_list_applications,
          channel=worker_node_channel,
          app_names=[],
          test_draining=True,
      )
  
      # Ensures Healthz method on the worker node is draining.
      ping_grpc_healthz(worker_node_channel, test_draining=True)
  
imports: import os
  import sys
  from typing import Any
  import grpc
  import pytest
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import Cluster
  from ray.serve._private.common import DeploymentID
  from ray.serve._private.constants import SERVE_NAMESPACE
  from ray.serve._private.test_utils import ping_fruit_stand, ping_grpc_another_method, ping_grpc_call_method, ping_grpc_healthz, ping_grpc_list_applications, ping_grpc_model_multiplexing, ping_grpc_streaming, send_signal_on_cancellation
  from ray.serve.config import gRPCOptions
  from ray.serve.generated import serve_pb2, serve_pb2_grpc
  from ray.serve.grpc_util import RayServegRPCContext
  from ray.serve.tests.test_config_files.grpc_deployment import g, g2
label_desc: Use image node_id 5 for calling a global function with example usage: test_grpc_proxy_on_draining_nodes(ray_cluster) and returns: unknown, unknown, serve_pb2