code: def submit(
      address: Optional[str],
      job_id: Optional[str],
      submission_id: Optional[str],
      runtime_env: Optional[str],
      runtime_env_json: Optional[str],
      metadata_json: Optional[str],
      working_dir: Optional[str],
      entrypoint: Tuple[str],
      entrypoint_num_cpus: Optional[Union[int, float]],
      entrypoint_num_gpus: Optional[Union[int, float]],
      entrypoint_memory: Optional[int],
      entrypoint_resources: Optional[str],
      no_wait: bool,
      verify: Union[bool, str],
      headers: Optional[str],
  ):
      """Submits a job to be run on the cluster.
  
      By default (if --no-wait is not set), streams logs to stdout until the job finishes.
      If the job succeeded, exits with 0. If it failed, exits with 1.
  
      Example:
          `ray job submit -- python my_script.py --arg=val`
      """
      if job_id:
          cli_logger.warning(
              "--job-id option is deprecated. Please use --submission-id instead."
          )
      if entrypoint_resources is not None:
          entrypoint_resources = parse_resources_json(
              entrypoint_resources,
              cli_logger,
              cf,
              command_arg="entrypoint-resources",
          )
      if metadata_json is not None:
          metadata_json = parse_metadata_json(
              metadata_json, cli_logger, cf, command_arg="metadata-json"
          )
  
      submission_id = submission_id or job_id
  
      if ray_constants.RAY_JOB_SUBMIT_HOOK in os.environ:
          # Submit all args as **kwargs per the JOB_SUBMIT_HOOK contract.
          _load_class(os.environ[ray_constants.RAY_JOB_SUBMIT_HOOK])(
              address=address,
              job_id=submission_id,
              submission_id=submission_id,
              runtime_env=runtime_env,
              runtime_env_json=runtime_env_json,
              metadata_json=metadata_json,
              working_dir=working_dir,
              entrypoint=entrypoint,
              entrypoint_num_cpus=entrypoint_num_cpus,
              entrypoint_num_gpus=entrypoint_num_gpus,
              entrypoint_memory=entrypoint_memory,
              entrypoint_resources=entrypoint_resources,
              no_wait=no_wait,
          )
  
      client = _get_sdk_client(
          address,
          create_cluster_if_needed=True,
          headers=headers,
          verify=verify,
      )
  
      final_runtime_env = parse_runtime_env_args(
          runtime_env=runtime_env,
          runtime_env_json=runtime_env_json,
          working_dir=working_dir,
      )
      job_id = client.submit_job(
          entrypoint=list2cmdline(entrypoint),
          submission_id=submission_id,
          runtime_env=final_runtime_env,
          metadata=metadata_json,
          entrypoint_num_cpus=entrypoint_num_cpus,
          entrypoint_num_gpus=entrypoint_num_gpus,
          entrypoint_memory=entrypoint_memory,
          entrypoint_resources=entrypoint_resources,
      )
  
      _log_big_success_msg(f"Job '{job_id}' submitted successfully")
  
      with cli_logger.group("Next steps"):
          cli_logger.print("Query the logs of the job:")
          with cli_logger.indented():
              cli_logger.print(cf.bold(f"ray job logs {job_id}"))
  
          cli_logger.print("Query the status of the job:")
          with cli_logger.indented():
              cli_logger.print(cf.bold(f"ray job status {job_id}"))
  
          cli_logger.print("Request the job to be stopped:")
          with cli_logger.indented():
              cli_logger.print(cf.bold(f"ray job stop {job_id}"))
  
      cli_logger.newline()
      sdk_version = client.get_version()
      # sdk version 0 does not have log streaming
      if not no_wait:
          if int(sdk_version) > 0:
              cli_logger.print(
                  "Tailing logs until the job exits "
                  "(disable with --no-wait):"
              )
              job_status = (
                  get_or_create_event_loop().run_until_complete(
                      _tail_logs(client, job_id)
                  )
              )
              if job_status == JobStatus.FAILED:
                  sys.exit(1)
          else:
              cli_logger.warning(
                  "Tailing logs is not enabled for job sdk client version "
                  f"{sdk_version}. Please upgrade Ray to the latest version "
                  "for this feature."
              )
  
imports: import json
  import os
  import sys
  import pprint
  import time
  from subprocess import list2cmdline
  from typing import Optional, Tuple, Union, Dict, Any
  import click
  import ray._private.ray_constants
  from ray._private.storage import _load_class
  from ray._private.utils import get_or_create_event_loop
  from ray.autoscaler._private.cli_logger import add_click_logging_options, cf, cli_logger
  from ray.dashboard.modules.dashboard_sdk import parse_runtime_env_args
  from ray.job_submission import JobStatus, JobSubmissionClient
  from ray.dashboard.modules.job.cli_utils import add_common_job_options
  from ray.dashboard.modules.job.utils import redact_url_password
  from ray.util.annotations import PublicAPI
  from ray._private.utils import parse_resources_json, parse_metadata_json
label_desc: Use image node_id 7 for calling a global function with example usage: submit(address, job_id, submission_id, runtime_env, runtime_env_json, metadata_json, working_dir, entrypoint, entrypoint_num_cpus, entrypoint_num_gpus, entrypoint_memory, entrypoint_resources, no_wait, verify, headers) without return types