code: def test_operation_stats(monkeypatch, shutdown_only):
      # Test operation stats are available when flag is on.
      operation_metrics = [
          "ray_operation_count",
          "ray_operation_run_time_ms",
          "ray_operation_queue_time_ms",
          "ray_operation_active_count",
      ]
      with monkeypatch.context() as m:
          m.setenv("RAY_event_stats_metrics", "1")
          addr = ray.init()
  
          signal = SignalActor.remote()
  
          @ray.remote
          class Actor:
              def __init__(self, signal):
                  self.signal = signal
  
              def get_worker_id(self):
                  return ray.get_runtime_context().get_worker_id()
  
              def wait(self):
                  ray.get(self.signal.wait.remote())
  
          actor = Actor.remote(signal)
          worker_id = ray.get(actor.get_worker_id.remote())
          obj_ref = actor.wait.remote()
  
          def verify():
              metrics = raw_metrics(addr)
              samples = metrics["ray_operation_count"]
              found = False
              for sample in samples:
                  if (
                      sample.labels["Method"]
                      == "CoreWorkerService.grpc_client.PushTask"
                      and sample.labels["Component"] == "core_worker"
                      and sample.labels["WorkerId"] == worker_id
                  ):
                      found = True
                      assert sample.value == 1
              if not found:
                  return False
  
              samples = metrics["ray_operation_active_count"]
              found = False
              for sample in samples:
                  if (
                      sample.labels["Method"]
                      == "CoreWorkerService.grpc_client.PushTask"
                      and sample.labels["Component"] == "core_worker"
                      and sample.labels["WorkerId"] == worker_id
                  ):
                      found = True
                      assert sample.value == 1
              if not found:
                  return False
  
              return True
  
          wait_for_condition(verify, timeout=60)
  
          ray.get(signal.send.remote())
          ray.get(obj_ref)
  
          def verify():
              metrics = raw_metrics(addr)
  
              samples = metrics["ray_operation_count"]
              found = False
              for sample in samples:
                  if (
                      sample.labels["Method"]
                      == "CoreWorkerService.grpc_client.PushTask"
                      and sample.labels["Component"] == "core_worker"
                      and sample.labels["WorkerId"] == worker_id
                  ):
                      found = True
                      assert sample.value == 1
              if not found:
                  return False
  
              found = False
              for sample in samples:
                  if (
                      sample.labels["Method"]
                      == "CoreWorkerService.grpc_client.PushTask.OnReplyReceived"
                      and sample.labels["Component"] == "core_worker"
                      and sample.labels["WorkerId"] == worker_id
                  ):
                      found = True
                      assert sample.value == 1
              if not found:
                  return False
  
              samples = metrics["ray_operation_active_count"]
              found = False
              for sample in samples:
                  if (
                      sample.labels["Method"]
                      == "CoreWorkerService.grpc_client.PushTask"
                      and sample.labels["Component"] == "core_worker"
                      and sample.labels["WorkerId"] == worker_id
                  ):
                      found = True
                      assert sample.value == 0
              if not found:
                  return False
  
              found = False
              for sample in samples:
                  if (
                      sample.labels["Method"]
                      == "CoreWorkerService.grpc_client.PushTask.OnReplyReceived"
                      and sample.labels["Component"] == "core_worker"
                      and sample.labels["WorkerId"] == worker_id
                  ):
                      found = True
                      assert sample.value == 0
              if not found:
                  return False
  
              metric_names = set(metrics.keys())
              for op_metric in operation_metrics:
                  assert op_metric in metric_names
                  samples = metrics[op_metric]
                  components = set()
                  for sample in samples:
                      components.add(sample.labels["Component"])
              assert {
                  "raylet",
                  "gcs_server",
                  "core_worker",
              } == components
              return True
  
          wait_for_condition(verify, timeout=60)
  
imports: import signal
  import json
  import os
  import pathlib
  import sys
  import requests
  from pprint import pformat
  from unittest.mock import MagicMock
  import numpy
  import pytest
  import ray
  from ray.util.state import list_nodes
  from ray._private.metrics_agent import PrometheusServiceDiscoveryWriter
  from ray._private.ray_constants import PROMETHEUS_SERVICE_DISCOVERY_FILE
  from ray._private.test_utils import SignalActor, fetch_prometheus, fetch_prometheus_metrics, get_log_batch, wait_for_condition, raw_metrics
  from ray.autoscaler._private.constants import AUTOSCALER_METRIC_PORT
  from ray.dashboard.consts import DASHBOARD_METRIC_PORT
  from ray.util.metrics import Counter, Gauge, Histogram
label_desc: Use image node_id 4 for calling a global function with example usage: test_operation_stats(monkeypatch, shutdown_only) and returns: True, True, ray, False, False, False, False, False, False