class_docstr: Manages metrics for the replica.
  
  A variety of metrics are managed:
      - Fine-grained metrics are set for every request.
      - Autoscaling statistics are periodically pushed to the controller.
      - Queue length metrics are periodically recorded as user-facing gauges.
class_docstr_tok: ["Manages","metrics","for","the","replica",".","A","variety","of","metrics","are","managed",":","-","Fine-grained","metrics","are","set","for","every","request",".","-","Autoscaling","statistics","are","periodically","pushed","to","the","controller",".","-","Queue","length","metrics","are","periodically","recorded","as","user-facing","gauges","."]
code: def __init__(
      self,
      replica_tag: ReplicaTag,
      deployment_id: DeploymentID,
      autoscaling_config: Optional[AutoscalingConfig],
  ):
      self._replica_tag = replica_tag
      self._deployment_id = deployment_id
      self._metrics_pusher = MetricsPusher()
      self._metrics_store = InMemoryMetricsStore()
      self._autoscaling_config = autoscaling_config
      self._controller_handle = ray.get_actor(
          SERVE_CONTROLLER_NAME, namespace=SERVE_NAMESPACE
      )
  
      # Request counter (only set on replica startup).
      self._restart_counter = metrics.Counter(
          "serve_deployment_replica_starts",
          description=(
              "The number of times this replica has been restarted due to failure."
          ),
      )
      self._restart_counter.inc()
  
      # Per-request metrics.
      self._request_counter = metrics.Counter(
          "serve_deployment_request_counter",
          description=(
              "The number of queries that have been processed in this replica."
          ),
          tag_keys=("route",),
      )
  
      self._error_counter = metrics.Counter(
          "serve_deployment_error_counter",
          description=(
              "The number of exceptions that have occurred in this replica."
          ),
          tag_keys=("route",),
      )
  
      self._processing_latency_tracker = metrics.Histogram(
          "serve_deployment_processing_latency_ms",
          description="The latency for queries to be processed.",
          boundaries=DEFAULT_LATENCY_BUCKET_MS,
          tag_keys=("route",),
      )
  
      # User-facing Prometheus gauges.
      self._num_pending_items = metrics.Gauge(
          "serve_replica_pending_queries",
          description="The current number of pending queries.",
      )
      self._num_processing_items = metrics.Gauge(
          "serve_replica_processing_queries",
          description="The current number of queries being processed.",
      )
  
      # Set user-facing gauges periodically.
      self._metrics_pusher.register_task(
          self._set_replica_requests_metrics,
          RAY_SERVE_GAUGE_METRIC_SET_PERIOD_S,
      )
  
      if self._autoscaling_config:
          # Push autoscaling metrics to the controller periodically.
          self._metrics_pusher.register_task(
              self._collect_autoscaling_metrics,
              self._autoscaling_config.metrics_interval_s,
              self._controller_handle.record_autoscaling_metrics.remote,
          )
          # Collect autoscaling metrics locally periodically.
          self._metrics_pusher.register_task(
              self.get_num_pending_and_running_requests,
              min(
                  RAY_SERVE_REPLICA_AUTOSCALING_METRIC_RECORD_PERIOD_S,
                  self._autoscaling_config.metrics_interval_s,
              ),
              self._add_autoscaling_metrics_point,
          )
  
imports: import asyncio
  import inspect
  import logging
  import os
  import pickle
  import time
  import traceback
  from contextlib import contextmanager
  from importlib import import_module
  from typing import Any, AsyncGenerator, Callable, Dict, Optional, Tuple, Union
  import starlette.responses
  import ray
  from ray import cloudpickle
  from ray._private.utils import get_or_create_event_loop
  from ray.actor import ActorClass
  from ray.remote_function import RemoteFunction
  from ray.serve import metrics
  from ray.serve._private.autoscaling_metrics import InMemoryMetricsStore
  from ray.serve._private.common import DeploymentID, ReplicaName, ReplicaTag, ServeComponentType, StreamingHTTPRequest, gRPCRequest
  from ray.serve._private.config import DeploymentConfig
  from ray.serve._private.constants import DEFAULT_LATENCY_BUCKET_MS, GRPC_CONTEXT_ARG_NAME, HEALTH_CHECK_METHOD, RAY_SERVE_GAUGE_METRIC_SET_PERIOD_S, RAY_SERVE_REPLICA_AUTOSCALING_METRIC_RECORD_PERIOD_S, RECONFIGURE_METHOD, REPLICA_CONTROL_PLANE_CONCURRENCY_GROUP, SERVE_CONTROLLER_NAME, SERVE_LOGGER_NAME, SERVE_NAMESPACE
  from ray.serve._private.http_util import ASGIAppReplicaWrapper, ASGIArgs, ASGIReceiveProxy, MessageQueue, Response
  from ray.serve._private.logging_utils import access_log_msg, configure_component_cpu_profiler, configure_component_logger, configure_component_memory_profiler, get_component_logger_file_path
  from ray.serve._private.router import RequestMetadata
  from ray.serve._private.utils import MetricsPusher, merge_dict, parse_import_path, wrap_to_ray_error
  from ray.serve._private.version import DeploymentVersion
  from ray.serve.config import AutoscalingConfig
  from ray.serve.deployment import Deployment
  from ray.serve.exceptions import RayServeException
  from ray.serve.schema import LoggingConfig
label_desc: Use image node_id 1 to create a new ReplicaMetricsManager object  with example: obj = ReplicaMetricsManager(replica_tag, deployment_id, autoscaling_config)