code: def test_metrics_agent_proxy_record_and_export_from_workers_complicated(
      get_agent,
  ):  # noqa
      """
      Test the complicated worker death case.
      """
      namespace = "test"
      agent, agent_port = get_agent
  
      # Each worker will report 2 metrics.
      # i.e.,
      # worker 1 => test_1, test_2.
      # worker 2 => test_3, test_4.
      # ...
      worker_ids = [WorkerID.from_random() for _ in range(4)]
  
      metrics = []
      for i in range(8):
          m = generate_protobuf_metric(
              f"test_{i}",
              "desc",
              "",
              label_keys=["a", "b"],
              timeseries=[],
          )
          m.timeseries.append(generate_timeseries(["a", str(i)], [3]))
          metrics.append(m)
  
      i = 0
      for worker_id in worker_ids:
          agent.proxy_export_metrics(
              [metrics[i], metrics[i + 1]],
              worker_id_hex=worker_id.hex(),
          )
          i += 2
  
      # All metrics must be available.
      for i in range(len(metrics)):
          assert (
              get_metric(f"{namespace}_test_{i}", agent_port)
              is not None
          )
  
      # Mark the worker as dead and make sure metrics are properly cleaned.
      i = 0
  
      while len(worker_ids):
          for worker_id in worker_ids:
              agent.clean_all_dead_worker_metrics()
              assert (
                  get_metric(f"{namespace}_test_{i}", agent_port)
                  is None
              )
              assert (
                  get_metric(f"{namespace}_test_{i+1}", agent_port)
                  is None
              )
  
          worker_ids.pop(0)
          metrics.pop(0)
          metrics.pop(0)
  
          i = 0
          for worker_id in worker_ids:
              agent.proxy_export_metrics(
                  [metrics[i], metrics[i + 1]],
                  worker_id_hex=worker_id.hex(),
              )
              i += 2
  
          # Make sure the rest of metrics are still there because new metrics
          # are reported.
          for i in range(i + 2, len(metrics)):
              assert (
                  get_metric(f"{namespace}_test_{i}", agent_port)
                  is not None
              ), i
          i += 2
  
imports: import os
  import time
  import sys
  import pytest
  import ray._private.prometheus_exporter
  from typing import List
  from opencensus.stats.view_manager import ViewManager
  from opencensus.stats.stats_recorder import StatsRecorder
  from opencensus.stats import execution_context
  from prometheus_client.core import REGISTRY
  from ray._private.metrics_agent import Gauge, MetricsAgent, Record, RAY_WORKER_TIMEOUT_S
  from ray._private.services import new_port
  from ray.core.generated.metrics_pb2 import Metric, MetricDescriptor, Point, LabelKey, TimeSeries, LabelValue
  from ray._raylet import WorkerID
  from ray._private.test_utils import fetch_prometheus_metrics, fetch_raw_prometheus, wait_for_condition
label_desc: Use image node_id 10 for calling a global function with example usage: test_metrics_agent_proxy_record_and_export_from_workers_complicated(get_agent) without return types