code: def main(
      test_workload: str,
      test_workload_timeout: float,
      test_no_raise_on_timeout: bool,
      results_cloud_storage_uri: Optional[str],
      metrics_cloud_storage_uri: Optional[str],
      output_cloud_storage_uri: Optional[str],
      upload_cloud_storage_uri: Optional[str],
      artifact_path: Optional[str],
      prepare_commands: List[str],
      prepare_commands_timeouts: List[str],
  ):
      """
      This function provides extra functionality for an Anyscale Job.
  
      1. Runs prepare commands and handles their timeouts
      2. Runs the actual test workload and handles its timeout
      3. Uploads test results.json
      4. Gathers prometheus metrics
      5. Uploads prometheus metrics.json
      6. Uploads output.json
      """
      logger.info("### Starting ###")
      start_time = time.monotonic()
  
      if len(prepare_commands) != len(prepare_commands_timeouts):
          raise ValueError(
              "`prepare_commands` and `prepare_commands_timeouts` must "
              "have the same length."
          )
  
      # Run prepare commands. All prepare commands must pass.
      (
          prepare_passed,
          prepare_return_codes,
          last_prepare_time_taken,
      ) = run_prepare_commands(
          prepare_commands, prepare_commands_timeouts
      )
  
      uploaded_results = False
      collected_metrics = False
      uploaded_metrics = False
      uploaded_artifact = artifact_path is not None
      workload_time_taken = None
  
      # If all prepare commands passed, run actual test workload.
      if prepare_passed:
          logger.info("### Starting entrypoint ###")
          command_start_time = time.monotonic()
          return_code = run_bash_command(
              test_workload, test_workload_timeout
          )
          workload_time_taken = time.monotonic() - command_start_time
  
          timed_out = return_code == TIMEOUT_RETURN_CODE
          if timed_out:
              msg = f"Timed out. Time taken: {workload_time_taken}"
              if test_no_raise_on_timeout:
                  logger.info(msg)
              else:
                  logger.error(msg)
          else:
              logger.info(
                  f"Finished with return code {return_code}. "
                  f"Time taken: {workload_time_taken}"
              )
  
          # Upload results.json
          uploaded_results = run_storage_cp(
              os.environ.get("TEST_OUTPUT_JSON", None),
              results_cloud_storage_uri,
          )
  
          # Collect prometheus metrics
          collected_metrics = collect_metrics(workload_time_taken)
          if collected_metrics:
              # Upload prometheus metrics
              uploaded_metrics = run_storage_cp(
                  os.environ.get("METRICS_OUTPUT_JSON", None),
                  metrics_cloud_storage_uri,
              )
  
          uploaded_artifact = run_storage_cp(
              artifact_path,
              os.path.join(
                  upload_cloud_storage_uri,
                  os.environ["USER_GENERATED_ARTIFACT"],
              )
              if "USER_GENERATED_ARTIFACT" in os.environ
              else None,
          )
  
      else:
          return_code = None
  
      total_time_taken = time.monotonic() - start_time
      output_json = {
          "return_code": return_code,
          "prepare_return_codes": prepare_return_codes,
          "last_prepare_time_taken": last_prepare_time_taken,
          "workload_time_taken": workload_time_taken,
          "total_time_taken": total_time_taken,
          "uploaded_results": uploaded_results,
          "collected_metrics": collected_metrics,
          "uploaded_metrics": uploaded_metrics,
          "uploaded_artifact": uploaded_artifact,
      }
      output_json = json.dumps(
          output_json,
          ensure_ascii=True,
          sort_keys=True,
          separators=(",", ":"),
      )
  
      output_json_file = (Path.cwd() / OUTPUT_JSON_FILENAME).resolve()
      with open(output_json_file, "w") as fp:
          fp.write(output_json)
  
      # Upload output.json
      run_storage_cp(str(output_json_file), output_cloud_storage_uri)
  
      logger.info("### Finished ###")
      # This will be read by the AnyscaleJobRunner on the buildkite runner
      # if output.json cannot be obtained from cloud storage
      logger.info(f"### JSON |{output_json}| ###")
  
      # Flush buffers
      logging.shutdown()
      print("", flush=True)
      print("", file=sys.stderr, flush=True)
  
      if (
          return_code == TIMEOUT_RETURN_CODE
          and test_no_raise_on_timeout
      ):
          return_code = 0
      elif return_code is None:
          return_code = 1
  
      time.sleep(1)
      return return_code
  
file_path: ray/release/ray_release/command_runner/_anyscale_job_wrapper.py
imports: import argparse
  import time
  import os
  from pathlib import Path
  import subprocess
  import multiprocessing
  import json
  import sys
  import logging
  from urllib.parse import urlparse
  from typing import Optional, List, Tuple
label_desc: Use image node_id 8 for calling a global function with example usage: main(test_workload, test_workload_timeout, test_no_raise_on_timeout, results_cloud_storage_uri, metrics_cloud_storage_uri, output_cloud_storage_uri, upload_cloud_storage_uri, artifact_path, prepare_commands, prepare_commands_timeouts) and returns: return_code