code: def test_rest_api(manage_ray_with_telemetry, tmp_dir):
      """
      Check that telemetry works with REST API.
      """
  
      subprocess.check_output(["ray", "start", "--head"])
      wait_for_condition(check_ray_started, timeout=5)
  
      storage = TelemetryStorage.remote()
  
      serve.run(
          receiver_app,
          host="0.0.0.0",
          name="telemetry",
          route_prefix=TELEMETRY_ROUTE_PREFIX,
      )
  
      wait_for_condition(
          lambda: ray.get(storage.get_reports_received.remote()) > 1,
          timeout=15,
      )
  
      # Check that REST API telemetry is not set
      report = ray.get(storage.get_report.remote())
      assert (
          ServeUsageTag.REST_API_VERSION.get_value_from_report(report)
          is None
      )
  
      serve.delete(name="telemetry", _blocking=True)
  
      config = {
          "applications": [
              {
                  "name": "receiver_app",
                  "import_path": "ray.serve._private.test_utils.receiver_app",
                  "route_prefix": TELEMETRY_ROUTE_PREFIX,
              },
              {
                  "name": "stub_app",
                  "import_path": "ray.serve.tests.test_telemetry.stub_app",
                  "route_prefix": "/stub",
              },
          ]
      }
      config_file_path = f"{tmp_dir}/config.yaml"
      with open(config_file_path, "w+") as f:
          yaml.safe_dump(config, f)
  
      subprocess.check_output(["serve", "deploy", config_file_path])
  
      client = _get_global_client()
      # Make sure the applications are RUNNING.
      wait_for_condition(
          lambda: serve.status().applications["receiver_app"].status
          == ApplicationStatus.RUNNING,
          timeout=15,
      )
      wait_for_condition(
          lambda: serve.status().applications["stub_app"].status
          == ApplicationStatus.RUNNING,
          timeout=15,
      )
  
      current_num_reports = ray.get(
          storage.get_reports_received.remote()
      )
  
      wait_for_condition(
          lambda: ray.get(storage.get_reports_received.remote())
          > current_num_reports,
          timeout=5,
      )
      report = ray.get(storage.get_report.remote())
  
      # Check all telemetry relevant to the Serve apps on this cluster
      assert (
          ServeUsageTag.REST_API_VERSION.get_value_from_report(report)
          == "v2"
      )
      assert (
          ServeUsageTag.API_VERSION.get_value_from_report(report)
          == "v2"
      )
      assert (
          int(
              ServeUsageTag.NUM_GPU_DEPLOYMENTS.get_value_from_report(
                  report
              )
          )
          == 0
      )
  
      # Assert num of deployments from controller.
      assert len(client.get_all_deployment_statuses()) == 2
      assert (
          int(ServeUsageTag.NUM_APPS.get_value_from_report(report)) == 2
      )
      assert (
          int(
              ServeUsageTag.NUM_DEPLOYMENTS.get_value_from_report(
                  report
              )
          )
          == 2
      )
  
      # Check that Serve telemetry not relevant to the running apps is omitted
      assert (
          ServeUsageTag.FASTAPI_USED.get_value_from_report(report)
          is None
      )
      assert (
          ServeUsageTag.GRPC_INGRESS_USED.get_value_from_report(report)
          is None
      )
      assert (
          ServeUsageTag.HTTP_ADAPTER_USED.get_value_from_report(report)
          is None
      )
      assert (
          ServeUsageTag.DAG_DRIVER_USED.get_value_from_report(report)
          is None
      )
  
      # Check that app deletions are tracked.
      new_config = {
          "applications": [
              {
                  "name": "receiver_app",
                  "import_path": "ray.serve._private.test_utils.receiver_app",
                  "route_prefix": TELEMETRY_ROUTE_PREFIX,
              },
          ]
      }
  
      with open(config_file_path, "w+") as f:
          yaml.safe_dump(new_config, f)
  
      subprocess.check_output(["serve", "deploy", config_file_path])
  
      wait_for_condition(
          lambda: int(
              ServeUsageTag.NUM_APPS.get_value_from_report(
                  ray.get(storage.get_report.remote())
              )
          )
          == 1,
          timeout=15,
      )
      report = ray.get(storage.get_report.remote())
      assert (
          int(ServeUsageTag.NUM_APPS.get_value_from_report(report)) == 1
      )
      assert (
          int(
              ServeUsageTag.NUM_DEPLOYMENTS.get_value_from_report(
                  report
              )
          )
          == 1
      )
  
imports: import subprocess
  import sys
  import time
  import pytest
  import requests
  import yaml
  from fastapi import FastAPI
  import ray
  from ray import serve
  from ray._private.test_utils import wait_for_condition
  from ray._private.usage.usage_lib import get_extra_usage_tags_to_report
  from ray.serve._private.common import ApplicationStatus
  from ray.serve._private.constants import SERVE_MULTIPLEXED_MODEL_ID
  from ray.serve._private.test_utils import TELEMETRY_ROUTE_PREFIX, TelemetryStorage, check_ray_started, receiver_app, start_telemetry_app
  from ray.serve._private.usage import ServeUsageTag
  from ray.serve.context import _get_global_client
  from ray.serve.schema import ServeDeploySchema
label_desc: Use image node_id 2 for calling a global function with example usage: test_rest_api(manage_ray_with_telemetry, tmp_dir) without return types