code: def test_e2e(serve_instance):
      signal = SignalActor.remote()
  
      @serve.deployment(
          autoscaling_config={
              "metrics_interval_s": 0.1,
              "min_replicas": 1,
              "max_replicas": 2,
              "target_num_ongoing_requests_per_replica": 1,
              "upscale_delay_s": 0,
              "downscale_delay_s": 0,
              "look_back_period_s": 1,
          },
          # We will send over a lot of queries. This will make sure replicas are
          # killed quickly during cleanup.
          graceful_shutdown_timeout_s=1,
          max_concurrent_queries=25,
          version="v1",
      )
      class A:
          def __call__(self):
              ray.get(signal.wait.remote())
  
      handle = serve.run(A.bind())
      dep_id = DeploymentID("A", "default")
      [handle.remote()._to_object_ref_sync() for _ in range(50)]
  
      # Wait for metrics to propagate
      def get_data():
          data = ray.get(
              serve_instance._controller._dump_autoscaling_metrics_for_testing.remote()
          )[dep_id]
          print(data)
          return data
  
      wait_for_condition(lambda: len(get_data()) > 0)
      print("Autoscaling metrics started recording on controller.")
  
      # Many queries should be inflight.
      def last_timestamp_value_high():
          data = get_data()
          metrics = list(data.values())
          assert len(metrics) == 2
          assert metrics[0] > 0 and metrics[1] > 0
          assert sum(metrics) > 40
          return True
  
      wait_for_condition(last_timestamp_value_high)
      print(
          "Confirmed there are metrics from 2 replicas, and many queries are inflight."
      )
      print("Releasing signal.")
      signal.send.remote()
  
      def check_running_replicas(expected):
          replicas = ray.get(
              serve_instance._controller._dump_replica_states_for_testing.remote(
                  dep_id
              )
          )
          running_replicas = replicas.get([ReplicaState.RUNNING])
          assert len(running_replicas) == expected
          return True
  
      # After traffic stops, num replica should drop to 1
      wait_for_condition(check_running_replicas, expected=1, timeout=15)
      print("Num replicas dropped to 1.")
  
      # The metrics stored on controller should only have info on the remaining replica
      wait_for_condition(lambda: len(get_data()) == 1)
      print("Metrics stored on the controller reduced to 1 replica.")
  
file_path: ray/python/ray/serve/tests/test_autoscaling_metrics.py
imports: import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.serve._private.autoscaling_metrics import InMemoryMetricsStore
  from ray.serve._private.common import DeploymentID, ReplicaState
label_desc: Use image node_id 1 for calling a global function with example usage: test_e2e(serve_instance) and returns: data, True, True