code: def step(
      ctx,
      step_name,
      code_package_sha,
      code_package_url,
      executable=None,
      image=None,
      image_pull_policy=None,
      service_account=None,
      secrets=None,
      node_selector=None,
      k8s_namespace=None,
      cpu=None,
      disk=None,
      memory=None,
      gpu=None,
      gpu_vendor=None,
      use_tmpfs=None,
      tmpfs_tempdir=None,
      tmpfs_size=None,
      tmpfs_path=None,
      run_time_limit=None,
      persistent_volume_claims=None,
      tolerations=None,
      **kwargs,
  ):
      def echo(msg, stream="stderr", job_id=None, **kwargs):
          msg = util.to_unicode(msg)
          if job_id:
              msg = "[%s] %s" % (job_id, msg)
          ctx.obj.echo_always(msg, err=(stream == sys.stderr), **kwargs)
  
      node = ctx.obj.graph[step_name]
  
      # Construct entrypoint CLI
      executable = ctx.obj.environment.executable(step_name, executable)
  
      # Set environment
      env = {}
      env_deco = [
          deco for deco in node.decorators if deco.name == "environment"
      ]
      if env_deco:
          env = env_deco[0].attributes["vars"]
  
      # Set input paths.
      input_paths = kwargs.get("input_paths")
      split_vars = None
      if input_paths:
          max_size = 30 * 1024
          split_vars = {
              "METAFLOW_INPUT_PATHS_%d"
              % (i // max_size): input_paths[i : i + max_size]
              for i in range(0, len(input_paths), max_size)
          }
          kwargs["input_paths"] = "".join(
              "${%s}" % s for s in split_vars.keys()
          )
          env.update(split_vars)
  
      # Set retry policy.
      retry_count = int(kwargs.get("retry_count", 0))
      retry_deco = [
          deco for deco in node.decorators if deco.name == "retry"
      ]
      minutes_between_retries = None
      if retry_deco:
          minutes_between_retries = int(
              retry_deco[0].attributes.get("minutes_between_retries", 2)
          )
      if retry_count:
          ctx.obj.echo_always(
              "Sleeping %d minutes before the next retry"
              % minutes_between_retries
          )
          time.sleep(minutes_between_retries * 60)
  
      step_cli = (
          "{entrypoint} {top_args} step {step} {step_args}".format(
              entrypoint="%s -u %s"
              % (executable, os.path.basename(sys.argv[0])),
              top_args=" ".join(
                  util.dict_to_cli_options(ctx.parent.parent.params)
              ),
              step=step_name,
              step_args=" ".join(util.dict_to_cli_options(kwargs)),
          )
      )
  
      # Set log tailing.
      ds = ctx.obj.flow_datastore.get_task_datastore(
          mode="w",
          run_id=kwargs["run_id"],
          step_name=step_name,
          task_id=kwargs["task_id"],
          attempt=int(retry_count),
      )
      stdout_location = ds.get_log_location(TASK_LOG_SOURCE, "stdout")
      stderr_location = ds.get_log_location(TASK_LOG_SOURCE, "stderr")
  
      # `node_selector` is a tuple of strings, convert it to a dictionary
      node_selector = parse_kube_keyvalue_list(node_selector)
  
      def _sync_metadata():
          if ctx.obj.metadata.TYPE == "local":
              sync_local_metadata_from_datastore(
                  DATASTORE_LOCAL_DIR,
                  ctx.obj.flow_datastore.get_task_datastore(
                      kwargs["run_id"], step_name, kwargs["task_id"]
                  ),
              )
  
      try:
          kubernetes = Kubernetes(
              datastore=ctx.obj.flow_datastore,
              metadata=ctx.obj.metadata,
              environment=ctx.obj.environment,
          )
          # Configure and launch Kubernetes job.
          with ctx.obj.monitor.measure(
              "metaflow.kubernetes.launch_job"
          ):
              kubernetes.launch_job(
                  flow_name=ctx.obj.flow.name,
                  run_id=kwargs["run_id"],
                  step_name=step_name,
                  task_id=kwargs["task_id"],
                  attempt=str(retry_count),
                  user=util.get_username(),
                  code_package_sha=code_package_sha,
                  code_package_url=code_package_url,
                  code_package_ds=ctx.obj.flow_datastore.TYPE,
                  step_cli=step_cli,
                  docker_image=image,
                  docker_image_pull_policy=image_pull_policy,
                  service_account=service_account,
                  secrets=secrets,
                  node_selector=node_selector,
                  namespace=k8s_namespace,
                  cpu=cpu,
                  disk=disk,
                  memory=memory,
                  gpu=gpu,
                  gpu_vendor=gpu_vendor,
                  use_tmpfs=use_tmpfs,
                  tmpfs_tempdir=tmpfs_tempdir,
                  tmpfs_size=tmpfs_size,
                  tmpfs_path=tmpfs_path,
                  run_time_limit=run_time_limit,
                  env=env,
                  persistent_volume_claims=persistent_volume_claims,
                  tolerations=tolerations,
              )
      except Exception as e:
          traceback.print_exc(chain=False)
          _sync_metadata()
          sys.exit(METAFLOW_EXIT_DISALLOW_RETRY)
      try:
          kubernetes.wait(stdout_location, stderr_location, echo=echo)
      except KubernetesKilledException:
          # don't retry killed tasks
          traceback.print_exc()
          sys.exit(METAFLOW_EXIT_DISALLOW_RETRY)
      finally:
          _sync_metadata()
  
file_path: metaflow/metaflow/plugins/kubernetes/kubernetes_cli.py
imports: import os
  import sys
  import time
  import traceback
  from metaflow import JSONTypeClass, util
  from metaflow._vendor import click
  from metaflow.exception import METAFLOW_EXIT_DISALLOW_RETRY, CommandException
  from metaflow.metadata.util import sync_local_metadata_from_datastore
  from metaflow.metaflow_config import DATASTORE_LOCAL_DIR, KUBERNETES_LABELS
  from metaflow.mflog import TASK_LOG_SOURCE
  import metaflow.tracing
  from .kubernetes import Kubernetes, KubernetesKilledException, parse_kube_keyvalue_list
  from .kubernetes_decorator import KubernetesDecorator
label_desc: Use image node_id 3 for calling a global function with example usage: step(ctx, step_name, code_package_sha, code_package_url, executable, image, image_pull_policy, service_account, secrets, node_selector, k8s_namespace, cpu, disk, memory, gpu, gpu_vendor, use_tmpfs, tmpfs_tempdir, tmpfs_size, tmpfs_path, run_time_limit, persistent_volume_claims, tolerations) without return types