code: def chrome_tracing_dump(
      tasks: List[dict],
  ) -> str:
      """Generate a chrome/perfetto tracing dump using task events.
  
      Args:
          tasks: List of tasks generated by a state API list_tasks(detail=True).
  
      Returns:
          Json serialized dump to create a chrome/perfetto tracing.
      """
      # All events from given tasks.
      all_events = []
  
      # Chrome tracing doesn't have a concept of "node". Instead, we use
      # chrome tracing's pid == ray's node.
      # chrome tracing's tid == ray's process.
      # Note that pid or tid is usually integer, but ray's node/process has
      # ids in string.
      # Unfortunately, perfetto doesn't allow to have string as a value of pid/tid.
      # To workaround it, we use Metadata event from chrome tracing schema
      # (https://docs.google.com/document/d/1CvAClvFfyA5R-PhYUmn5OOQtYMH4h6I0nSsKchNAySU/preview#heading=h.xqopa5m0e28f) # noqa
      # which allows pid/tid -> name mapping. In order to use this schema
      # we build node_ip/(node_ip, worker_id) -> arbitrary index mapping.
  
      # node ip address -> node idx.
      node_to_index = {}
      # Arbitrary index mapped to the ip address.
      node_idx = 0
      # (node index, worker id) -> worker idx
      worker_to_index = {}
      # Arbitrary index mapped to the (node index, worker id).
      worker_idx = 0
  
      for task in tasks:
          profiling_data = task.get("profiling_data", [])
          if profiling_data:
              node_ip_address = profiling_data["node_ip_address"]
              component_events = profiling_data["events"]
              component_type = profiling_data["component_type"]
              component_id = (
                  component_type + ":" + profiling_data["component_id"]
              )
  
              if component_type not in ["worker", "driver"]:
                  continue
  
              for event in component_events:
                  extra_data = event["extra_data"]
                  # Propagate extra data.
                  extra_data["task_id"] = task["task_id"]
                  extra_data["job_id"] = task["job_id"]
                  extra_data["attempt_number"] = task["attempt_number"]
                  extra_data["func_or_class_name"] = task[
                      "func_or_class_name"
                  ]
                  extra_data["actor_id"] = task["actor_id"]
                  event_name = event["event_name"]
  
                  # build a id -> arbitrary index mapping
                  if node_ip_address not in node_to_index:
                      node_to_index[node_ip_address] = node_idx
                      # Whenever new node ip is introduced, we increment the index.
                      node_idx += 1
  
                  if (
                      node_to_index[node_ip_address],
                      component_id,
                  ) not in worker_to_index:  # noqa
                      worker_to_index[
                          (node_to_index[node_ip_address], component_id)
                      ] = worker_idx  # noqa
                      worker_idx += 1
  
                  # Modify the name with the additional user-defined extra data.
                  cname = _default_color_mapping[event["event_name"]]
                  name = event_name
  
                  if "cname" in extra_data:
                      cname = _default_color_mapping[
                          event["extra_data"]["cname"]
                      ]
                  if "name" in extra_data:
                      name = extra_data["name"]
  
                  new_event = ChromeTracingCompleteEvent(
                      cat=event_name,
                      name=name,
                      pid=node_to_index[node_ip_address],
                      tid=worker_to_index[
                          (node_to_index[node_ip_address], component_id)
                      ],
                      ts=event["start_time"] * 1e3,
                      dur=(event["end_time"] * 1e3)
                      - (event["start_time"] * 1e3),
                      cname=cname,
                      args=extra_data,
                  )
                  all_events.append(asdict(new_event))
  
      for node, i in node_to_index.items():
          all_events.append(
              asdict(
                  ChromeTracingMetadataEvent(
                      name="process_name",
                      pid=i,
                      args={"name": f"Node {node}"},
                  )
              )
          )
  
      for worker, i in worker_to_index.items():
          all_events.append(
              asdict(
                  ChromeTracingMetadataEvent(
                      name="thread_name",
                      ph="M",
                      tid=i,
                      pid=worker[0],
                      args={"name": worker[1]},
                  )
              )
          )
  
      # Handle task event disabled.
      return json.dumps(all_events)
  
file_path: ray/python/ray/_private/profiling.py
imports: import os
  import json
  from collections import defaultdict
  from dataclasses import dataclass, asdict
  from typing import List, Dict, Union
  import ray
label_desc: Use image node_id 2 for calling a global function with example usage: chrome_tracing_dump(tasks) and returns: json