code: def test_per_func_name_stats(shutdown_only):
      # Test operation stats are available when flag is on.
      comp_metrics = [
          "ray_component_cpu_percentage",
          "ray_component_rss_mb",
          "ray_component_num_fds",
      ]
      if sys.platform == "linux" or sys.platform == "linux2":
          # Uss only available from Linux
          comp_metrics.append("ray_component_uss_mb")
          comp_metrics.append("ray_component_mem_shared_bytes")
      addr = ray.init(num_cpus=2)
  
      @ray.remote
      class Actor:
          def __init__(self):
              self.arr = np.random.rand(5 * 1024 * 1024)  # 40 MB
              self.shared_arr = ray.put(np.random.rand(5 * 1024 * 1024))
  
          def pid(self):
              return os.getpid()
  
      @ray.remote
      class ActorB:
          def __init__(self):
              self.arr = np.random.rand(5 * 1024 * 1024)  # 40 MB
              self.shared_arr = ray.put(np.random.rand(5 * 1024 * 1024))
  
      a = Actor.remote()  # noqa
      b = ActorB.remote()
  
      ray.get(a.__ray_ready__.remote())
      ray.get(b.__ray_ready__.remote())
  
      def verify_components():
          metrics = raw_metrics(addr)
          metric_names = set(metrics.keys())
          components = set()
          for metric in comp_metrics:
              assert metric in metric_names
              samples = metrics[metric]
              for sample in samples:
                  components.add(sample.labels["Component"])
          print(components)
          assert {
              "raylet",
              "agent",
              "ray::Actor",
              "ray::ActorB",
              "ray::IDLE",
          } <= components
          return True
  
      wait_for_condition(verify_components, timeout=30)
  
      def verify_mem_usage():
          metrics = raw_metrics(addr)
          for metric in comp_metrics:
              samples = metrics[metric]
              for sample in samples:
                  if sample.labels["Component"] == "ray::ActorB":
                      assert sample.value > 0.0
                      print(sample)
                      print(sample.value)
                  if sample.labels["Component"] == "ray::Actor":
                      assert sample.value > 0.0
                      print(sample)
                      print(sample.value)
          return True
  
      wait_for_condition(verify_mem_usage, timeout=30)
  
      # Verify ActorB is reported as value 0 because it is killed.
      ray.kill(b)
      # Kill Actor by sigkill, which happens upon OOM.
      pid = ray.get(a.pid.remote())
      os.kill(pid, signal.SIGKILL)
  
      def verify_mem_cleaned():
          metrics = raw_metrics(addr)
          for metric in comp_metrics:
              samples = metrics[metric]
              for sample in samples:
                  if sample.labels["Component"] == "ray::ActorB":
                      assert sample.value == 0.0
                  if sample.labels["Component"] == "ray::Actor":
                      assert sample.value == 0.0
          return True
  
      wait_for_condition(verify_mem_cleaned, timeout=30)
  
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 5 for calling a global function with example usage: test_per_func_name_stats(shutdown_only) and returns: True, True, True, os