class_docstr: Autoscaling monitor.
  
  This process periodically collects stats from the GCS and triggers
  autoscaler updates.
docstr: Run the monitor loop.
docstr_tok: ["Run","the","monitor","loop","."]
code: def _run(self):
      """Run the monitor loop."""
  
      while True:
          try:
              gcs_request_start_time = time.time()
              self.update_load_metrics()
              gcs_request_time = time.time() - gcs_request_start_time
              self.update_resource_requests()
              self.update_event_summary()
              load_metrics_summary = self.load_metrics.summary()
              status = {
                  "gcs_request_time": gcs_request_time,
                  "time": time.time(),
                  "monitor_pid": os.getpid(),
              }
  
              if self.autoscaler and not self.load_metrics:
                  # load_metrics is Falsey iff we haven't collected any
                  # resource messages from the GCS, which can happen at startup if
                  # the GCS hasn't yet received data from the Raylets.
                  # In this case, do not do an autoscaler update.
                  # Wait to get load metrics.
                  logger.info(
                      "Autoscaler has not yet received load metrics. Waiting."
                  )
              elif self.autoscaler:
                  # Process autoscaling actions
                  update_start_time = time.time()
                  self.autoscaler.update()
                  status["autoscaler_update_time"] = (
                      time.time() - update_start_time
                  )
                  autoscaler_summary = self.autoscaler.summary()
                  try:
                      self.emit_metrics(
                          load_metrics_summary,
                          autoscaler_summary,
                          self.autoscaler.all_node_types,
                      )
                  except Exception:
                      logger.exception("Error emitting metrics")
  
                  if autoscaler_summary:
                      status["autoscaler_report"] = asdict(
                          autoscaler_summary
                      )
                      status[
                          "non_terminated_nodes_time"
                      ] = (
                          self.autoscaler.non_terminated_nodes.non_terminated_nodes_time  # noqa: E501
                      )
  
                  for msg in self.event_summarizer.summary():
                      # Need to prefix each line of the message for the lines to
                      # get pushed to the driver logs.
                      for line in msg.split("\n"):
                          logger.info(
                              "{}{}".format(
                                  ray_constants.LOG_PREFIX_EVENT_SUMMARY,
                                  line,
                              )
                          )
                          if self.event_logger:
                              self.event_logger.info(line)
  
                  self.event_summarizer.clear()
  
              status["load_metrics_report"] = asdict(
                  load_metrics_summary
              )
              as_json = json.dumps(status)
              if _internal_kv_initialized():
                  _internal_kv_put(
                      ray_constants.DEBUG_AUTOSCALING_STATUS,
                      as_json,
                      overwrite=True,
                  )
          except Exception:
              # By default, do not exit the monitor on failure.
              if self.retry_on_failure:
                  logger.exception(
                      "Monitor: Execution exception. Trying again..."
                  )
              else:
                  raise
  
          # Wait for a autoscaler update interval before processing the next
          # round of messages.
          time.sleep(AUTOSCALER_UPDATE_INTERVAL_S)
  
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 6 for calling the Monitor obj's underlying member method code with example usage: obj._run() without return types