code: def test_profiling_api(shutdown_only):
      ray.init(
          num_cpus=2,
          _system_config={
              "task_events_report_interval_ms": 200,
              "enable_timeline": True,
          },
      )
  
      @ray.remote
      def f(delay):
          with profiling.profile(
              "custom_event", extra_data={"name": "custom name"}
          ):
              time.sleep(delay)
              pass
  
      @ray.remote
      def g(input_list):
          # The argument input_list should be a list containing one object ref.
          ray.wait([input_list[0]])
  
      ray.put(1)
      x = f.remote(1)
      ray.get([g.remote([x]), g.remote([x])])
  
      def verify():
          profile_data = ray.timeline()
          actual_types = {event["cat"] for event in profile_data}
          expected_types = {
              "task::f",  # for f
              "task::g",  # for g
              "task:deserialize_arguments",
              "task:execute",
              "task:store_outputs",
              "wait_for_function",
              "ray.get",
              "ray.put",
              "ray.wait",
              "submit_task",
              "fetch_and_run_function",
              "custom_event",  # This is the custom one from ray.profile.
          }
          assert expected_types == actual_types
          return True
  
      wait_for_condition(verify, timeout=20, retry_interval_ms=1000)
  
      # Test for content of the profiling events.
      @ray.remote
      def k():
          exec_time_us = time.time() * (10**6)
          worker_id = (
              ray._private.worker.global_worker.core_worker.get_worker_id().hex()
          )
          return worker_id, exec_time_us
  
      k_worker_id, k_exec_time_us = ray.get(k.remote())
  
      def verify():
          profile_data = ray.timeline()
          k_events = [
              event
              for event in profile_data
              if event["tid"] == f"worker:{k_worker_id}"
          ]
          assert len(k_events) > 0
          for event in k_events:
              if event["name"] == "task:execute":
                  reported_exec_time = event["ts"]
                  # diff smaller than 3 secs, a fine-tuned threshold from running locally.
                  assert abs(
                      reported_exec_time - k_exec_time_us
                  ) < 3 * (10**6)
  
          return True
  
      wait_for_condition(verify, timeout=20, retry_interval_ms=1000)
  
imports: import json
  import logging
  import os
  import sys
  import time
  import numpy
  import pytest
  import ray._private.profiling
  import ray.cluster_utils
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import client_test_enabled, wait_for_condition
  from ray.exceptions import ObjectFreedError
label_desc: Use image node_id 5 for calling a global function with example usage: test_profiling_api(shutdown_only) and returns: True, worker_id,  exec_time_us, True