class_docstr: Specifies that this step should execute on Kubernetes.
  
  Parameters
  ----------
  cpu : int, default: 1
      Number of CPUs required for this step. If `@resources` is
      also present, the maximum value from all decorators is used.
  memory : int, default: 4096
      Memory size (in MB) required for this step. If
      `@resources` is also present, the maximum value from all decorators is
      used.
  disk : int, default: 10240
      Disk size (in MB) required for this step. If
      `@resources` is also present, the maximum value from all decorators is
      used.
  image : str, optional
      Docker image to use when launching on Kubernetes. If not specified, and
      METAFLOW_KUBERNETES_CONTAINER_IMAGE is specified, that image is used. If
      not, a default Docker image mapping to the current version of Python is used.
  image_pull_policy: str, default: KUBERNETES_IMAGE_PULL_POLICY
      If given, the imagePullPolicy to be applied to the Docker image of the step.
  service_account : str, default: METAFLOW_KUBERNETES_SERVICE_ACCOUNT
      Kubernetes service account to use when launching pod in Kubernetes.
  secrets : List[str], optional
      Kubernetes secrets to use when launching pod in Kubernetes. These
      secrets are in addition to the ones defined in `METAFLOW_KUBERNETES_SECRETS`
      in Metaflow configuration.
  namespace : str, default: METAFLOW_KUBERNETES_NAMESPACE
      Kubernetes namespace to use when launching pod in Kubernetes.
  gpu: int, optional
      Number of GPUs required for this step. A value of zero implies that
      the scheduled node should not have GPUs.
  gpu_vendor: str, default: KUBERNETES_GPU_VENDOR
      The vendor of the GPUs to be used for this step.
  tolerations : List[str], default: METAFLOW_KUBERNETES_TOLERATIONS
      Kubernetes tolerations to use when launching pod in Kubernetes.
  use_tmpfs: bool, default: False
      This enables an explicit tmpfs mount for this step.
  tmpfs_tempdir: bool, default: True
      sets METAFLOW_TEMPDIR to tmpfs_path if set for this step.
  tmpfs_size: int, optional
      The value for the size (in MiB) of the tmpfs mount for this step.
      This parameter maps to the `--tmpfs` option in Docker. Defaults to 50% of the
      memory allocated for this step.
  tmpfs_path: string, optional
      Path to tmpfs mount for this step. Defaults to /metaflow_temp.
  persistent_volume_claims: Dict[str, str], optional
      A map (dictionary) of persistent volumes to be mounted to the pod for this step. The map is from persistent
      volumes to the path to which the volume is to be mounted, e.g., `{'pvc-name': '/path/to/mount/on'}`.
class_docstr_tok: ["Specifies","that","this","step","should","execute","on","Kubernetes",".","Parameters","--","--","--","--","--","cpu",":","int",",","default",":","1","Number","of","CPUs","required","for","this","step",".","If","`","@","resources","`","is","also","present",",","the","maximum","value","from","all","decorators","is","used",".","memory",":","int",",","default",":","4096","Memory","size","(","in","MB",")","required","for","this","step",".","If","`","@","resources","`","is","also","present",",","the","maximum","value","from","all","decorators","is","used",".","disk",":","int",",","default",":","10240","Disk","size","(","in","MB",")","required","for","this","step",".","If","`","@","resources","`","is","also","present",",","the","maximum","value","from","all","decorators","is","used",".","image",":","str",",","optional","Docker","image","to","use","when","launching","on","Kubernetes",".","If","not","specified",",","and","METAFLOW_KUBERNETES_CONTAINER_IMAGE","is","specified",",","that","image","is","used",".","If","not",",","a","default","Docker","image","mapping","to","the","current","version","of","Python","is","used",".","image_pull_policy",":","str",",","default",":","KUBERNETES_IMAGE_PULL_POLICY","If","given",",","the","imagePullPolicy","to","be","applied","to","the","Docker","image","of","the","step",".","service_account",":","str",",","default",":","METAFLOW_KUBERNETES_SERVICE_ACCOUNT","Kubernetes","service","account","to","use","when","launching","pod","in","Kubernetes",".","secrets",":","List","[","str","]",",","optional","Kubernetes","secrets","to","use","when","launching","pod","in","Kubernetes",".","These","secrets","are","in","addition","to","the","ones","defined","in","`","METAFLOW_KUBERNETES_SECRETS","`","in","Metaflow","configuration",".","namespace",":","str",",","default",":","METAFLOW_KUBERNETES_NAMESPACE","Kubernetes","namespace","to","use","when","launching","pod","in","Kubernetes",".","gpu",":","int",",","optional","Number","of","GPUs","required","for","this","step",".","A","value","of","zero","implies","that","the","scheduled","node","should","not","have","GPUs",".","gpu_vendor",":","str",",","default",":","KUBERNETES_GPU_VENDOR","The","vendor","of","the","GPUs","to","be","used","for","this","step",".","tolerations",":","List","[","str","]",",","default",":","METAFLOW_KUBERNETES_TOLERATIONS","Kubernetes","tolerations","to","use","when","launching","pod","in","Kubernetes",".","use_tmpfs",":","bool",",","default",":","False","This","enables","an","explicit","tmpfs","mount","for","this","step",".","tmpfs_tempdir",":","bool",",","default",":","True","sets","METAFLOW_TEMPDIR","to","tmpfs_path","if","set","for","this","step",".","tmpfs_size",":","int",",","optional","The","value","for","the","size","(","in","MiB",")","of","the","tmpfs","mount","for","this","step",".","This","parameter","maps","to","the","`","--","tmpfs","`","option","in","Docker",".","Defaults","to","50","%","of","the","memory","allocated","for","this","step",".","tmpfs_path",":","string",",","optional","Path","to","tmpfs","mount","for","this","step",".","Defaults","to","\/metaflow_temp",".","persistent_volume_claims",":","Dict","[","str",",","str","]",",","optional","A","map","(","dictionary",")","of","persistent","volumes","to","be","mounted","to","the","pod","for","this","step",".","The","map","is","from","persistent","volumes","to","the","path","to","which","the","volume","is","to","be","mounted",",","e.g.",",","`","{","'pvc-name","'",":","'\/path\/to\/mount\/on","'","}","`","."]
code: def __init__(self, attributes=None, statically_defined=False):
      super(KubernetesDecorator, self).__init__(
          attributes, statically_defined
      )
  
      if not self.attributes["namespace"]:
          self.attributes["namespace"] = KUBERNETES_NAMESPACE
      if not self.attributes["service_account"]:
          self.attributes[
              "service_account"
          ] = KUBERNETES_SERVICE_ACCOUNT
      if not self.attributes["gpu_vendor"]:
          self.attributes["gpu_vendor"] = KUBERNETES_GPU_VENDOR
      if (
          not self.attributes["node_selector"]
          and KUBERNETES_NODE_SELECTOR
      ):
          self.attributes["node_selector"] = KUBERNETES_NODE_SELECTOR
      if not self.attributes["tolerations"] and KUBERNETES_TOLERATIONS:
          self.attributes["tolerations"] = json.loads(
              KUBERNETES_TOLERATIONS
          )
      if (
          not self.attributes["persistent_volume_claims"]
          and KUBERNETES_PERSISTENT_VOLUME_CLAIMS
      ):
          self.attributes["persistent_volume_claims"] = json.loads(
              KUBERNETES_PERSISTENT_VOLUME_CLAIMS
          )
      if (
          not self.attributes["image_pull_policy"]
          and KUBERNETES_IMAGE_PULL_POLICY
      ):
          self.attributes[
              "image_pull_policy"
          ] = KUBERNETES_IMAGE_PULL_POLICY
  
      if isinstance(self.attributes["node_selector"], str):
          self.attributes["node_selector"] = parse_kube_keyvalue_list(
              self.attributes["node_selector"].split(",")
          )
  
      if self.attributes["tolerations"]:
          try:
              from kubernetes.client import V1Toleration
  
              for toleration in self.attributes["tolerations"]:
                  try:
                      invalid_keys = [
                          k
                          for k in toleration.keys()
                          if k not in V1Toleration.attribute_map.keys()
                      ]
                      if len(invalid_keys) > 0:
                          raise KubernetesException(
                              "Tolerations parameter contains invalid keys: %s"
                              % invalid_keys
                          )
                  except AttributeError:
                      raise KubernetesException(
                          "Unable to parse tolerations: %s"
                          % self.attributes["tolerations"]
                      )
          except (NameError, ImportError):
              pass
  
      # If no docker image is explicitly specified, impute a default image.
      if not self.attributes["image"]:
          # If metaflow-config specifies a docker image, just use that.
          if KUBERNETES_CONTAINER_IMAGE:
              self.attributes["image"] = KUBERNETES_CONTAINER_IMAGE
          # If metaflow-config doesn't specify a docker image, assign a
          # default docker image.
          else:
              # Default to vanilla Python image corresponding to major.minor
              # version of the Python interpreter launching the flow.
              self.attributes["image"] = "python:%s.%s" % (
                  platform.python_version_tuple()[0],
                  platform.python_version_tuple()[1],
              )
      # Assign docker registry URL for the image.
      if not get_docker_registry(self.attributes["image"]):
          if KUBERNETES_CONTAINER_REGISTRY:
              self.attributes["image"] = "%s/%s" % (
                  KUBERNETES_CONTAINER_REGISTRY.rstrip("/"),
                  self.attributes["image"],
              )
      # Check if TmpFS is enabled and set default tmpfs_size if missing.
      if self.attributes["use_tmpfs"] or (
          self.attributes["tmpfs_size"]
          and not self.attributes["use_tmpfs"]
      ):
          if not self.attributes["tmpfs_size"]:
              # default tmpfs behavior - https://man7.org/linux/man-pages/man5/tmpfs.5.html
              self.attributes["tmpfs_size"] = (
                  int(self.attributes["memory"]) // 2
              )
  
imports: import json
  import os
  import platform
  import sys
  from metaflow import current
  from metaflow.decorators import StepDecorator
  from metaflow.exception import MetaflowException
  from metaflow.metadata import MetaDatum
  from metaflow.metadata.util import sync_local_metadata_to_datastore
  from metaflow.metaflow_config import DATASTORE_LOCAL_DIR, KUBERNETES_CONTAINER_IMAGE, KUBERNETES_CONTAINER_REGISTRY, KUBERNETES_FETCH_EC2_METADATA, KUBERNETES_IMAGE_PULL_POLICY, KUBERNETES_GPU_VENDOR, KUBERNETES_NAMESPACE, KUBERNETES_NODE_SELECTOR, KUBERNETES_PERSISTENT_VOLUME_CLAIMS, KUBERNETES_TOLERATIONS, KUBERNETES_SERVICE_ACCOUNT
  from metaflow.plugins.resources_decorator import ResourcesDecorator
  from metaflow.plugins.timeout_decorator import get_run_time_limit_for_task
  from metaflow.sidecar import Sidecar
  from ..aws.aws_utils import get_docker_registry, get_ec2_instance_metadata
  from .kubernetes import KubernetesException, parse_kube_keyvalue_list
label_desc: Use image node_id 1 to create a new KubernetesDecorator object from inherited base classes: StepDecorator with example: obj = KubernetesDecorator(attributes, statically_defined)