code: def run_wrk_on_all_nodes(
      trial_length: str,
      num_connections: int,
      http_host: str,
      http_port: str,
      all_endpoints: List[str] = None,
      ignore_output: bool = False,
      debug: bool = False,
  ):
      """
      Use ray task to run one wrk trial on each node alive, picked randomly
      from all available deployments.
  
      Returns:
          all_metrics: (Dict[str, List[Union[float, int]]]) Parsed wrk metrics
              from each wrk on each running node
          all_wrk_stdout: (List[str]) decoded stdout of each wrk trial for per
              node checks at the end of experiment
      """
      all_metrics = defaultdict(list)
      all_wrk_stdout = []
      rst_ray_refs = []
      for node in ray.nodes():
          if node["Alive"]:
              node_resource = f"node:{node['NodeManagerAddress']}"
              # Randomly pick one from all available endpoints in ray cluster
              endpoint = random.choice(all_endpoints)
              rst_ray_refs.append(
                  run_one_wrk_trial.options(
                      num_cpus=0, resources={node_resource: 0.01}
                  ).remote(
                      trial_length,
                      num_connections,
                      http_host,
                      http_port,
                      endpoint,
                  )
              )
  
      print("Waiting for wrk trials to finish...")
      ray.wait(rst_ray_refs, num_returns=len(rst_ray_refs))
      print("Trials finished!")
  
      if ignore_output:
          return
  
      for i, decoded_output in enumerate(ray.get(rst_ray_refs)):
          if debug:
              print(f"decoded_output {i}: {decoded_output}")
          all_wrk_stdout.append(decoded_output)
          parsed_metrics = parse_wrk_decoded_stdout(decoded_output)
  
          # Per thread metrics
          all_metrics["per_thread_latency_avg_ms"].append(
              parsed_metrics["per_thread_latency_avg_ms"]
          )
          all_metrics["per_thread_latency_max_ms"].append(
              parsed_metrics["per_thread_latency_max_ms"]
          )
          all_metrics["per_thread_tps"].append(
              parsed_metrics["per_thread_tps"]
          )
          all_metrics["per_thread_max_tps"].append(
              parsed_metrics["per_thread_max_tps"]
          )
  
          # Per node metrics
          all_metrics["P50_latency_ms"].append(
              parsed_metrics["P50_latency_ms"]
          )
          all_metrics["P75_latency_ms"].append(
              parsed_metrics["P75_latency_ms"]
          )
          all_metrics["P90_latency_ms"].append(
              parsed_metrics["P90_latency_ms"]
          )
          all_metrics["P99_latency_ms"].append(
              parsed_metrics["P99_latency_ms"]
          )
  
          all_metrics["per_node_total_thoughput"].append(
              parsed_metrics["per_node_total_thoughput"]
          )
          all_metrics["per_node_total_transfer_KB"].append(
              parsed_metrics["per_node_total_transfer_KB"]
          )
  
          all_metrics["per_nodel_tps"].append(
              parsed_metrics["per_nodel_tps"]
          )
          all_metrics["per_node_transfer_per_sec_KB"].append(
              parsed_metrics["per_node_transfer_per_sec_KB"]
          )
          all_metrics["per_node_total_timeout_requests"].append(
              parsed_metrics.get("per_node_total_timeout_requests", 0)
          )
  
      return all_metrics, all_wrk_stdout
  
file_path: ray/release/serve_tests/workloads/serve_test_utils.py
imports: import json
  import logging
  import os
  import random
  import ray
  import re
  import subprocess
  from collections import defaultdict
  from serve_test_cluster_utils import NUM_CPU_PER_NODE
  from subprocess import PIPE
  from typing import Dict, List, Union
label_desc: Use image node_id 8 for calling a global function with example usage: run_wrk_on_all_nodes(trial_length, num_connections, http_host, http_port, all_endpoints, ignore_output, debug) and returns: all_metrics,  all_wrk_stdout