code: def test_metrics_export_end_to_end(_setup_cluster_for_test):
      TEST_TIMEOUT_S = 30
      (
          prom_addresses,
          autoscaler_export_addr,
          dashboard_export_addr,
      ) = _setup_cluster_for_test
  
      def test_cases():
          (
              components_dict,
              metric_names,
              metric_samples,
          ) = fetch_prometheus(prom_addresses)
          session_name = (
              ray._private.worker.global_worker.node.session_name
          )
  
          # Raylet should be on every node
          assert all(
              "raylet" in components
              for components in components_dict.values()
          )
  
          # GCS server should be on one node
          assert any(
              "gcs_server" in components
              for components in components_dict.values()
          )
  
          # Core worker should be on at least on node
          assert any(
              "core_worker" in components
              for components in components_dict.values()
          )
  
          # Make sure our user defined metrics exist
          for metric_name in [
              "test_counter",
              "test_histogram",
              "test_driver_counter",
          ]:
              assert any(
                  metric_name in full_name for full_name in metric_names
              )
  
          # Make sure metrics are recorded.
          for metric in _METRICS:
              assert (
                  metric in metric_names
              ), f"metric {metric} not in {metric_names}"
  
          for sample in metric_samples:
              if sample.name in _METRICS:
                  assert sample.labels["SessionName"] == session_name
              if sample.name in _DASHBOARD_METRICS:
                  assert sample.labels["SessionName"] == session_name
  
          # Make sure the numeric values are correct
          test_counter_sample = [
              m for m in metric_samples if "test_counter" in m.name
          ][0]
          assert test_counter_sample.value == 4.0
  
          test_driver_counter_sample = [
              m
              for m in metric_samples
              if "test_driver_counter" in m.name
          ][0]
          assert test_driver_counter_sample.value == 1.0
  
          test_histogram_samples = [
              m for m in metric_samples if "test_histogram" in m.name
          ]
          buckets = {
              m.labels["le"]: m.value
              for m in test_histogram_samples
              if "_bucket" in m.name
          }
          # We recorded value 1.5 for the histogram. In Prometheus data model
          # the histogram is cumulative. So we expect the count to appear in
          # <1.1 and <+Inf buckets.
          assert buckets == {"0.1": 0.0, "1.6": 1.0, "+Inf": 1.0}
          hist_count = [
              m for m in test_histogram_samples if "_count" in m.name
          ][0].value
          hist_sum = [
              m for m in test_histogram_samples if "_sum" in m.name
          ][0].value
          assert hist_count == 1
          assert hist_sum == 1.5
  
          # Make sure the gRPC stats are not reported from workers. We disabled
          # it there because it has too high cardinality.
          grpc_metrics = [
              "ray_grpc_server_req_process_time_ms",
              "ray_grpc_server_req_new_total",
              "ray_grpc_server_req_handling_total",
              "ray_grpc_server_req_finished_total",
          ]
          for grpc_metric in grpc_metrics:
              grpc_samples = [
                  m for m in metric_samples if grpc_metric in m.name
              ]
              for grpc_sample in grpc_samples:
                  assert (
                      grpc_sample.labels["Component"] != "core_worker"
                  )
  
          # Autoscaler metrics
          (
              _,
              autoscaler_metric_names,
              autoscaler_samples,
          ) = fetch_prometheus(
              [autoscaler_export_addr]
          )  # noqa
          for metric in _AUTOSCALER_METRICS:
              # Metric name should appear with some suffix (_count, _total,
              # etc...) in the list of all names
              assert any(
                  name.startswith(metric)
                  for name in autoscaler_metric_names
              ), f"{metric} not in {autoscaler_metric_names}"
              for sample in autoscaler_samples:
                  assert sample.labels["SessionName"] == session_name
  
          # Dashboard metrics
          _, dashboard_metric_names, _ = fetch_prometheus(
              [dashboard_export_addr]
          )
          for metric in _DASHBOARD_METRICS:
              # Metric name should appear with some suffix (_count, _total,
              # etc...) in the list of all names
              assert any(
                  name.startswith(metric)
                  for name in dashboard_metric_names
              ), f"{metric} not in {dashboard_metric_names}"
  
      def wrap_test_case_for_retry():
          try:
              test_cases()
              return True
          except AssertionError:
              return False
  
      try:
          wait_for_condition(
              wrap_test_case_for_retry,
              timeout=TEST_TIMEOUT_S,
              retry_interval_ms=1000,  # Yield resource for other processes
          )
      except RuntimeError:
          print(
              f"The components are {pformat(fetch_prometheus(prom_addresses))}"
          )
          test_cases()
  
imports: import signal
  import json
  import os
  import pathlib
  import sys
  import requests
  from pprint import pformat
  from unittest.mock import MagicMock
  import numpy
  import pytest
  import ray
  from ray.util.state import list_nodes
  from ray._private.metrics_agent import PrometheusServiceDiscoveryWriter
  from ray._private.ray_constants import PROMETHEUS_SERVICE_DISCOVERY_FILE
  from ray._private.test_utils import SignalActor, fetch_prometheus, fetch_prometheus_metrics, get_log_batch, wait_for_condition, raw_metrics
  from ray.autoscaler._private.constants import AUTOSCALER_METRIC_PORT
  from ray.dashboard.consts import DASHBOARD_METRIC_PORT
  from ray.util.metrics import Counter, Gauge, Histogram
label_desc: Use image node_id 2 for calling a global function with example usage: test_metrics_export_end_to_end(_setup_cluster_for_test) and returns: True, False