class_docstr: Autoscaling monitor.
  
  This process periodically collects stats from the GCS and triggers
  autoscaler updates.
docstr: Fetches resource usage data from GCS and updates load metrics.
docstr_tok: ["Fetches","resource","usage","data","from","GCS","and","updates","load","metrics","."]
code: def update_load_metrics(self):
      """Fetches resource usage data from GCS and updates load metrics."""
  
      response = self.gcs_client.get_all_resource_usage(timeout=60)
      resources_batch_data = response.resource_usage_data
      log_resource_batch_data_if_desired(resources_batch_data)
  
      # Tell the readonly node provider what nodes to report.
      if self.readonly_config:
          new_nodes = []
          for msg in list(resources_batch_data.batch):
              node_id = msg.node_id.hex()
              new_nodes.append((node_id, msg.node_manager_address))
          self.autoscaler.provider._set_nodes(new_nodes)
  
      mirror_node_types = {}
      cluster_full = False
      if (
          hasattr(response, "cluster_full_of_actors_detected_by_gcs")
          and response.cluster_full_of_actors_detected_by_gcs
      ):
          # GCS has detected the cluster full of actors.
          cluster_full = True
      for resource_message in resources_batch_data.batch:
          node_id = resource_message.node_id
          # Generate node type config based on GCS reported node list.
          if self.readonly_config:
              # Keep prefix in sync with ReadonlyNodeProvider.
              node_type = format_readonly_node_type(node_id.hex())
              resources = {}
              for k, v in resource_message.resources_total.items():
                  resources[k] = v
              mirror_node_types[node_type] = {
                  "resources": resources,
                  "node_config": {},
                  "max_workers": 1,
              }
          if (
              hasattr(
                  resource_message, "cluster_full_of_actors_detected"
              )
              and resource_message.cluster_full_of_actors_detected
          ):
              # A worker node has detected the cluster full of actors.
              cluster_full = True
          total_resources = dict(resource_message.resources_total)
          available_resources = dict(
              resource_message.resources_available
          )
  
          waiting_bundles, infeasible_bundles = parse_resource_demands(
              resources_batch_data.resource_load_by_shape
          )
  
          pending_placement_groups = list(
              resources_batch_data.placement_group_load.placement_group_data
          )
  
          use_node_id_as_ip = (
              self.autoscaler is not None
              and self.autoscaler.config["provider"].get(
                  "use_node_id_as_ip", False
              )
          )
  
          # "use_node_id_as_ip" is a hack meant to address situations in
          # which there's more than one Ray node residing at a given ip.
          # TODO (Dmitri): Stop using ips as node identifiers.
          # https://github.com/ray-project/ray/issues/19086
          if use_node_id_as_ip:
              peloton_id = total_resources.get("NODE_ID_AS_RESOURCE")
              # Legacy support https://github.com/ray-project/ray/pull/17312
              if peloton_id is not None:
                  ip = str(int(peloton_id))
              else:
                  ip = node_id.hex()
          else:
              ip = resource_message.node_manager_address
          self.load_metrics.update(
              ip,
              node_id,
              total_resources,
              available_resources,
              waiting_bundles,
              infeasible_bundles,
              pending_placement_groups,
              cluster_full,
          )
      if self.readonly_config:
          self.readonly_config["available_node_types"].update(
              mirror_node_types
          )
  
imports: import argparse
  import json
  import logging
  import os
  import signal
  import sys
  import time
  import traceback
  from collections import Counter
  from dataclasses import asdict
  from typing import Any, Callable, Dict, Optional, Union
  import ray
  import ray._private.ray_constants
  import ray._private.utils
  from ray._private.event.event_logger import get_event_logger
  from ray._private.ray_logging import setup_component_logger
  from ray._raylet import GcsClient
  from ray.autoscaler._private.autoscaler import StandardAutoscaler
  from ray.autoscaler._private.commands import teardown_cluster
  from ray.autoscaler._private.constants import AUTOSCALER_MAX_RESOURCE_DEMAND_VECTOR_SIZE, AUTOSCALER_METRIC_PORT, AUTOSCALER_UPDATE_INTERVAL_S
  from ray.autoscaler._private.event_summarizer import EventSummarizer
  from ray.autoscaler._private.load_metrics import LoadMetrics
  from ray.autoscaler._private.prom_metrics import AutoscalerPrometheusMetrics
  from ray.autoscaler._private.util import format_readonly_node_type
  from ray.core.generated import gcs_pb2
  from ray.core.generated.event_pb2 import Event
  from ray.experimental.internal_kv import _initialize_internal_kv, _internal_kv_del, _internal_kv_get, _internal_kv_initialized, _internal_kv_put
label_desc: Use image node_id 3 for calling the Monitor obj's underlying member method code with example usage: obj.update_load_metrics() without return types