code: def test_report_per_component_stats():
      dashboard_agent = MagicMock()
      agent = ReporterAgent(dashboard_agent)
      # Assume it is a head node.
      agent._is_head_node = True
  
      # Generate stats.
      test_stats = copy.deepcopy(STATS_TEMPLATE)
      idle_stats = {
          "memory_info": Bunch(
              rss=55934976,
              vms=7026937856,
              uss=1234567,
              pfaults=15354,
              pageins=0,
          ),
          "memory_full_info": Bunch(uss=51428381),
          "cpu_percent": 5.0,
          "num_fds": 11,
          "cmdline": [
              "ray::IDLE",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
          ],
          "create_time": 1614826391.338613,
          "pid": 7174,
          "cpu_times": Bunch(
              user=0.607899328,
              system=0.274044032,
              children_user=0.0,
              children_system=0.0,
          ),
      }
      func_stats = {
          "memory_info": Bunch(
              rss=55934976, vms=7026937856, pfaults=15354, pageins=0
          ),
          "memory_full_info": Bunch(uss=51428381),
          "cpu_percent": 6.0,
          "num_fds": 12,
          "cmdline": [
              "ray::func",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
          ],
          "create_time": 1614826391.338613,
          "pid": 7175,
          "cpu_times": Bunch(
              user=0.607899328,
              system=0.274044032,
              children_user=0.0,
              children_system=0.0,
          ),
      }
      raylet_stast = {
          "memory_info": Bunch(
              rss=18354176, vms=6921486336, pfaults=6206, pageins=3
          ),
          "memory_full_info": Bunch(uss=51428381),
          "cpu_percent": 4.0,
          "num_fds": 13,
          "cmdline": ["fake raylet cmdline"],
          "create_time": 1614826390.274854,
          "pid": 7153,
          "cpu_times": Bunch(
              user=0.03683138,
              system=0.035913716,
              children_user=0.0,
              children_system=0.0,
          ),
      }
      agent_stats = {
          "memory_info": Bunch(
              rss=18354176, vms=6921486336, pfaults=6206, pageins=3
          ),
          "memory_full_info": Bunch(uss=51428381),
          "cpu_percent": 6.0,
          "num_fds": 14,
          "cmdline": ["fake raylet cmdline"],
          "create_time": 1614826390.274854,
          "pid": 7156,
          "cpu_times": Bunch(
              user=0.03683138,
              system=0.035913716,
              children_user=0.0,
              children_system=0.0,
          ),
      }
  
      test_stats["workers"] = [idle_stats, func_stats]
      test_stats["raylet"] = raylet_stast
      test_stats["agent"] = agent_stats
  
      cluster_stats = {
          "autoscaler_report": {
              "active_nodes": {"head_node": 1, "worker-node-0": 2},
              "failed_nodes": [],
              "pending_launches": {},
              "pending_nodes": [],
          }
      }
  
      def get_uss_and_cpu_and_num_fds_records(records):
          component_uss_mb_records = defaultdict(list)
          component_cpu_percentage_records = defaultdict(list)
          component_num_fds_records = defaultdict(list)
          for record in records:
              name = record.gauge.name
              if name == "component_uss_mb":
                  comp = record.tags["Component"]
                  component_uss_mb_records[comp].append(record)
              if name == "component_cpu_percentage":
                  comp = record.tags["Component"]
                  component_cpu_percentage_records[comp].append(record)
              if name == "component_num_fds":
                  comp = record.tags["Component"]
                  component_num_fds_records[comp].append(record)
          return (
              component_uss_mb_records,
              component_cpu_percentage_records,
              component_num_fds_records,
          )
  
      """
      Test basic case.
      """
      records = agent._record_stats(test_stats, cluster_stats)
      (
          uss_records,
          cpu_records,
          num_fds_records,
      ) = get_uss_and_cpu_and_num_fds_records(records)
  
      def verify_metrics_values(
          uss_records,
          cpu_records,
          num_fds_records,
          comp,
          uss,
          cpu_percent,
          num_fds,
      ):
          """Verify the component exists and match the resource usage."""
          assert comp in uss_records
          assert comp in cpu_records
          assert comp in num_fds_records
          uss_metrics = uss_records[comp][0].value
          cpu_percnet_metrics = cpu_records[comp][0].value
          num_fds_metrics = num_fds_records[comp][0].value
          assert uss_metrics == uss
          assert cpu_percnet_metrics == cpu_percent
          assert num_fds_metrics == num_fds
  
      stats_map = {
          "raylet": raylet_stast,
          "agent": agent_stats,
          "ray::IDLE": idle_stats,
          "ray::func": func_stats,
      }
      # Verify metrics are correctly reported with a component name.
      for comp, stats in stats_map.items():
          verify_metrics_values(
              uss_records,
              cpu_records,
              num_fds_records,
              comp,
              float(stats["memory_full_info"].uss) / 1.0e6,
              stats["cpu_percent"],
              stats["num_fds"],
          )
  
      """
      Test metrics are resetted (report metrics with values 0) when
      the proc doesn't exist anymore.
      """
      # Verify the metrics are reset after ray::func is killed.
      test_stats["workers"] = [idle_stats]
      records = agent._record_stats(test_stats, cluster_stats)
      (
          uss_records,
          cpu_records,
          num_fds_records,
      ) = get_uss_and_cpu_and_num_fds_records(records)
      verify_metrics_values(
          uss_records,
          cpu_records,
          num_fds_records,
          "ray::IDLE",
          float(idle_stats["memory_full_info"].uss) / 1.0e6,
          idle_stats["cpu_percent"],
          idle_stats["num_fds"],
      )
  
      comp = "ray::func"
      stats = func_stats
      # Value should be reset since func doesn't exist anymore.
      verify_metrics_values(
          uss_records,
          cpu_records,
          num_fds_records,
          "ray::func",
          0,
          0,
          0,
      )
  
      """
      Verify worker names are only reported when they start with ray::.
      """
      # Verify if the command doesn't start with ray::, metrics are not reported.
      unknown_stats = {
          "memory_info": Bunch(
              rss=55934976, vms=7026937856, pfaults=15354, pageins=0
          ),
          "memory_full_info": Bunch(uss=51428381),
          "cpu_percent": 6.0,
          "num_fds": 8,
          "cmdline": [
              "python mock",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
              "",
          ],
          "create_time": 1614826391.338613,
          "pid": 7175,
          "cpu_times": Bunch(
              user=0.607899328,
              system=0.274044032,
              children_user=0.0,
              children_system=0.0,
          ),
      }
      test_stats["workers"] = [idle_stats, unknown_stats]
  
      records = agent._record_stats(test_stats, cluster_stats)
      (
          uss_records,
          cpu_records,
          num_fds_records,
      ) = get_uss_and_cpu_and_num_fds_records(records)
      assert "python mock" not in uss_records
      assert "python mock" not in cpu_records
      assert "python mock" not in num_fds_records
  
imports: import logging
  import os
  import sys
  import requests
  import numpy
  import time
  import copy
  import pytest
  from collections import defaultdict
  from multiprocessing import Process
  from unittest.mock import MagicMock
  from google.protobuf import text_format
  import psutil
  import ray
  from mock import patch
  from ray._private import ray_constants
  from ray._private.metrics_agent import fix_grpc_metric
  from ray._private.test_utils import fetch_prometheus, format_web_url, wait_for_condition, wait_until_server_available
  from ray.dashboard.modules.reporter.reporter_agent import ReporterAgent
  from ray.dashboard.tests.conftest import *
  from ray.dashboard.utils import Bunch
  from ray.core.generated.metrics_pb2 import Metric
label_desc: Use image node_id 9 for calling a global function with example usage: test_report_per_component_stats() and returns: component_uss_mb_records,  component_cpu_percentage_records,  component_num_fds_records