code: def verify_load_metrics(
      monitor, expected_resource_usage=None, timeout=30
  ):
      request_resources(num_cpus=42)
  
      # add placement groups.
      pg_demands = [{"GPU": 2}, {"extra_resource": 2}]
      strategy = "STRICT_PACK"
      pg = placement_group(pg_demands, strategy=strategy)
      pg.ready()
      time.sleep(2)  # wait for placement groups to propagate.
  
      # Disable event clearing for test.
      monitor.event_summarizer.clear = lambda *a: None
  
      visited_atleast_once = [set(), set()]
      while True:
          monitor.update_load_metrics()
          monitor.update_resource_requests()
          monitor.update_event_summary()
          resource_usage = monitor.load_metrics._get_resource_usage()
  
          # Check resource request propagation.
          req = monitor.load_metrics.resource_requests
          assert req == [{"CPU": 1}] * 42, req
  
          pg_response_data = (
              monitor.load_metrics.pending_placement_groups
          )
          assert_correct_pg(pg_response_data, pg_demands, strategy)
  
          if "memory" in resource_usage[0]:
              del resource_usage[0]["memory"]
              visited_atleast_once[0].add("memory")
          if "object_store_memory" in resource_usage[0]:
              del resource_usage[0]["object_store_memory"]
              visited_atleast_once[0].add("object_store_memory")
          if "memory" in resource_usage[1]:
              del resource_usage[1]["memory"]
              visited_atleast_once[1].add("memory")
          if "object_store_memory" in resource_usage[1]:
              del resource_usage[1]["object_store_memory"]
              visited_atleast_once[1].add("object_store_memory")
          for key in list(resource_usage[0].keys()):
              if key.startswith("node:"):
                  del resource_usage[0][key]
                  visited_atleast_once[0].add("node:")
          for key in list(resource_usage[1].keys()):
              if key.startswith("node:"):
                  del resource_usage[1][key]
                  visited_atleast_once[1].add("node:")
          if expected_resource_usage is None:
              if all(x for x in resource_usage[0:]):
                  break
          elif all(
              x == y
              for x, y in zip(resource_usage, expected_resource_usage)
          ):
              break
          else:
              timeout -= 1
              time.sleep(1)
  
          if timeout <= 0:
              raise ValueError(
                  "Timeout. {} != {}".format(
                      resource_usage, expected_resource_usage
                  )
              )
  
      # Sanity check we emitted a resize event.
      assert any(
          "Resized to" in x for x in monitor.event_summarizer.summary()
      )
  
      assert visited_atleast_once[0] == {
          "memory",
          "object_store_memory",
          "node:",
      }
      assert visited_atleast_once[0] == visited_atleast_once[1]
  
      remove_placement_group(pg)
  
      return resource_usage
  
file_path: ray/python/ray/tests/test_multi_node_2.py
imports: import logging
  import time
  import pytest
  import ray
  from ray._private.test_utils import SignalActor, generate_system_config_map, wait_for_condition
  from ray.autoscaler._private.monitor import Monitor
  from ray.autoscaler.sdk import request_resources
  from ray.cluster_utils import Cluster
  from ray.util.placement_group import placement_group, remove_placement_group
label_desc: Use image node_id 6 for calling a global function with example usage: verify_load_metrics(monitor, expected_resource_usage, timeout) and returns: resource_usage