code: def test_metrics(shutdown_only):
      cluster = AutoscalingCluster(
          head_resources={"CPU": 0},
          worker_node_types={
              "type-i": {
                  "resources": {"CPU": 1},
                  "node_config": {},
                  "min_workers": 0,
                  "max_workers": 1,
              },
              "type-ii": {
                  "resources": {"CPU": 1},
                  "node_config": {},
                  "min_workers": 0,
                  "max_workers": 1,
              },
          },
      )
  
      try:
          cluster.start()
          info = ray.init(address="auto")
          autoscaler_export_addr = "{}:{}".format(
              info.address_info["node_ip_address"],
              AUTOSCALER_METRIC_PORT,
          )
  
          @ray.remote(num_cpus=1)
          class Foo:
              def ping(self):
                  return True
  
          zero_reported_condition = get_metric_check_condition(
              [
                  MetricSamplePattern(
                      name="autoscaler_cluster_resources",
                      value=0,
                      partial_label_match={"resource": "CPU"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_pending_resources", value=0
                  ),
                  MetricSamplePattern(
                      name="autoscaler_pending_nodes", value=0
                  ),
                  MetricSamplePattern(
                      name="autoscaler_active_nodes",
                      value=0,
                      partial_label_match={"NodeType": "type-i"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_active_nodes",
                      value=0,
                      partial_label_match={"NodeType": "type-ii"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_active_nodes",
                      value=1,
                      partial_label_match={
                          "NodeType": "ray.head.default"
                      },
                  ),
              ],
              export_addr=autoscaler_export_addr,
          )
          wait_for_condition(zero_reported_condition)
  
          actors = [Foo.remote() for _ in range(2)]
          ray.get([actor.ping.remote() for actor in actors])
  
          two_cpu_no_pending_condition = get_metric_check_condition(
              [
                  MetricSamplePattern(
                      name="autoscaler_cluster_resources",
                      value=2,
                      partial_label_match={"resource": "CPU"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_pending_nodes",
                      value=0,
                      partial_label_match={"NodeType": "type-i"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_pending_nodes",
                      value=0,
                      partial_label_match={"NodeType": "type-ii"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_active_nodes",
                      value=1,
                      partial_label_match={"NodeType": "type-i"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_active_nodes",
                      value=1,
                      partial_label_match={"NodeType": "type-ii"},
                  ),
                  MetricSamplePattern(
                      name="autoscaler_active_nodes",
                      value=1,
                      partial_label_match={
                          "NodeType": "ray.head.default"
                      },
                  ),
              ],
              export_addr=autoscaler_export_addr,
          )
          wait_for_condition(two_cpu_no_pending_condition)
          # TODO (Alex): Ideally we'd also assert that pending increases
          # eventually became 1 or 2, but it's difficult to do that in a
          # non-racey way. (Perhaps we would need to artificially delay the fake
          # autoscaler node launch?).
  
      finally:
          cluster.shutdown()
  
file_path: ray/python/ray/tests/test_autoscaler_e2e.py
imports: import subprocess
  from ray.autoscaler._private.constants import AUTOSCALER_METRIC_PORT
  import pytest
  import ray
  import sys
  from ray._private.test_utils import reset_autoscaler_v2_enabled_cache, wait_for_condition, get_metric_check_condition, MetricSamplePattern
  from ray.cluster_utils import AutoscalingCluster
  from ray.autoscaler.node_launch_exception import NodeLaunchException
label_desc: Use image node_id 3 for calling a global function with example usage: test_metrics(shutdown_only) and returns: True