class_docstr: A model server runner that launches model server in kubernetes cluster.
class_docstr_tok: ["A","model","server","runner","that","launches","model","server","in","kubernetes","cluster","."]
docstr: Create a kubernetes model server runner.
  
  Args:
    model_path: An IV-flavored model path. (See model_path_utils.py)
    serving_binary: A ServingBinary to run.
    serving_spec: A ServingSpec instance.
docstr_tok: ["Create","a","kubernetes","model","server","runner",".","Args",":","model_path",":","An","IV-flavored","model","path",".","(","See","model_path_utils.py",")","serving_binary",":","A","ServingBinary","to","run",".","serving_spec",":","A","ServingSpec","instance","."]
code: def __init__(
      self,
      model_path: str,
      serving_binary: serving_bins.ServingBinary,
      serving_spec: infra_validator_pb2.ServingSpec,
  ):
      """Create a kubernetes model server runner.
  
      Args:
        model_path: An IV-flavored model path. (See model_path_utils.py)
        serving_binary: A ServingBinary to run.
        serving_spec: A ServingSpec instance.
      """
      assert (
          serving_spec.WhichOneof("serving_platform") == "kubernetes"
      ), "ServingSpec configuration mismatch."
      self._config = serving_spec.kubernetes
  
      self._model_path = model_path
      self._serving_binary = serving_binary
      self._serving_spec = serving_spec
      self._k8s_core_api = kube_utils.make_core_v1_api()
      if not kube_utils.is_inside_kfp():
          raise NotImplementedError(
              "KubernetesRunner should be running inside KFP."
          )
      self._executor_pod = kube_utils.get_current_kfp_pod(
          self._k8s_core_api
      )
      self._executor_container = _get_container_or_error(
          self._executor_pod,
          container_name=kube_utils.ARGO_MAIN_CONTAINER_NAME,
      )
      self._namespace = kube_utils.get_kfp_namespace()
      self._label_dict = {
          _APP_KEY: _MODEL_SERVER_APP_LABEL,
      }
      # Pod name would be populated once creation request sent.
      self._pod_name = None
      # Endpoint would be populated once the Pod is running.
      self._endpoint = None
  
imports: import datetime
  import os
  import time
  from typing import Optional
  from absl import logging
  from apache_beam.utils import retry
  from kubernetes import client
  from kubernetes.client import rest
  from tfx.components.infra_validator import error_types
  from tfx.components.infra_validator import serving_bins
  from tfx.components.infra_validator.model_server_runners import base_runner
  from tfx.proto import infra_validator_pb2
  from tfx.utils import kube_utils
label_desc: Use image node_id 1 to create a new KubernetesRunner object from inherited base classes: base_runner with example: obj = KubernetesRunner(model_path, serving_binary, serving_spec)