code: def test_prometheus_file_based_service_discovery(ray_start_cluster):
      # Make sure Prometheus service discovery file is correctly written
      # when number of nodes are dynamically changed.
      NUM_NODES = 5
      cluster = ray_start_cluster
      nodes = [cluster.add_node() for _ in range(NUM_NODES)]
      cluster.wait_for_nodes()
      addr = ray.init(address=cluster.address)
      writer = PrometheusServiceDiscoveryWriter(
          addr["gcs_address"],
          "/tmp/ray",
      )
  
      def get_metrics_export_address_from_node(nodes):
          node_export_addrs = [
              "{}:{}".format(
                  node.node_ip_address, node.metrics_export_port
              )
              for node in nodes
          ]
          # monitor should be run on head node for `ray_start_cluster` fixture
          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
          )
          return node_export_addrs + [
              autoscaler_export_addr,
              dashboard_export_addr,
          ]
  
      loaded_json_data = json.loads(
          writer.get_file_discovery_content()
      )[0]
      assert set(get_metrics_export_address_from_node(nodes)) == set(
          loaded_json_data["targets"]
      )
  
      # Let's update nodes.
      for _ in range(3):
          nodes.append(cluster.add_node())
  
      # Make sure service discovery file content is correctly updated.
      loaded_json_data = json.loads(
          writer.get_file_discovery_content()
      )[0]
      assert set(get_metrics_export_address_from_node(nodes)) == set(
          loaded_json_data["targets"]
      )
  
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 6 for calling a global function with example usage: test_prometheus_file_based_service_discovery(ray_start_cluster) and returns: unknown