code: def _setup_cluster_for_test(request, ray_start_cluster):
      enable_metrics_collection = request.param
      NUM_NODES = 2
      cluster = ray_start_cluster
      # Add a head node.
      cluster.add_node(
          _system_config={
              "metrics_report_interval_ms": 1000,
              "event_stats_print_interval_ms": 500,
              "event_stats": True,
              "enable_metrics_collection": enable_metrics_collection,
          }
      )
      # Add worker nodes.
      [cluster.add_node() for _ in range(NUM_NODES - 1)]
      cluster.wait_for_nodes()
      ray_context = ray.init(address=cluster.address)
  
      worker_should_exit = SignalActor.remote()
  
      # Generate a metric in the driver.
      counter = Counter("test_driver_counter", description="desc")
      counter.inc()
  
      # Generate some metrics from actor & tasks.
      @ray.remote
      def f():
          counter = Counter("test_counter", description="desc")
          counter.inc()
          counter = ray.get(ray.put(counter))  # Test serialization.
          counter.inc()
          counter.inc(2)
          ray.get(worker_should_exit.wait.remote())
  
      # Generate some metrics for the placement group.
      pg = ray.util.placement_group(bundles=[{"CPU": 1}])
      ray.get(pg.ready())
      ray.util.remove_placement_group(pg)
  
      @ray.remote
      class A:
          async def ping(self):
              histogram = Histogram(
                  "test_histogram",
                  description="desc",
                  boundaries=[0.1, 1.6],
              )
              histogram = ray.get(
                  ray.put(histogram)
              )  # Test serialization.
              histogram.observe(1.5)
              ray.get(worker_should_exit.wait.remote())
  
      a = A.remote()
      obj_refs = [f.remote(), a.ping.remote()]
      # Infeasible task
      b = f.options(resources={"a": 1})  # noqa
  
      # Make a request to the dashboard to produce some dashboard metrics
      requests.get(f"http://{ray_context.dashboard_url}/nodes")
  
      node_info_list = ray.nodes()
      prom_addresses = []
      for node_info in node_info_list:
          metrics_export_port = node_info["MetricsExportPort"]
          addr = node_info["NodeManagerAddress"]
          prom_addresses.append(f"{addr}:{metrics_export_port}")
      autoscaler_export_addr = "{}:{}".format(
          cluster.head_node.node_ip_address, AUTOSCALER_METRIC_PORT
      )
      dashboard_export_addr = "{}:{}".format(
          cluster.head_node.node_ip_address, DASHBOARD_METRIC_PORT
      )
      yield prom_addresses, autoscaler_export_addr, dashboard_export_addr
  
      ray.get(worker_should_exit.send.remote())
      ray.get(obj_refs)
      ray.shutdown()
      cluster.shutdown()
  
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 1 for calling a global function with example usage: _setup_cluster_for_test(request, ray_start_cluster) without return types