code: def __init__(
      self,
      kube_api_client=None,
      image=None,
      namespace=None,
      supervisor_num=1,
      supervisor_cpu=1,
      supervisor_mem="4G",
      supervisor_mem_limit_ratio=None,
      worker_num=1,
      worker_cpu=None,
      worker_mem=None,
      worker_spill_paths=None,
      worker_cache_mem=None,
      min_worker_num=None,
      worker_min_cache_mem=None,
      worker_mem_limit_ratio=None,
      web_port=None,
      service_name=None,
      service_type=None,
      timeout=None,
      **kwargs,
  ):
      from kubernetes import client as kube_client
  
      if worker_cpu is None or worker_mem is None:  # pragma: no cover
          raise TypeError(
              "`worker_cpu` and `worker_mem` must be specified"
          )
  
      self._api_client = kube_api_client
      self._core_api = kube_client.CoreV1Api(kube_api_client)
  
      self._namespace = namespace
      self._image = image
      self._timeout = timeout
      self._service_name = service_name or "marsservice"
      self._service_type = service_type or "NodePort"
      self._extra_volumes = kwargs.pop("extra_volumes", ())
      self._pre_stop_command = kwargs.pop("pre_stop_command", None)
      self._log_when_fail = kwargs.pop("log_when_fail", False)
  
      extra_modules = kwargs.pop("extra_modules", None) or []
      extra_modules = (
          extra_modules.split(",")
          if isinstance(extra_modules, str)
          else extra_modules
      )
      extra_envs = kwargs.pop("extra_env", None) or dict()
      extra_labels = kwargs.pop("extra_labels", None) or dict()
      service_port = (
          kwargs.pop("service_port", None) or self._default_service_port
      )
  
      def _override_modules(updates):
          modules = set(extra_modules)
          updates = (
              updates.split(",")
              if isinstance(updates, str)
              else updates
          )
          modules.update(updates)
          return sorted(modules)
  
      def _override_dict(d, updates):
          updates = updates or dict()
          ret = d.copy()
          ret.update(updates)
          return ret
  
      _override_envs = functools.partial(_override_dict, extra_envs)
      _override_labels = functools.partial(_override_dict, extra_labels)
  
      self._supervisor_num = supervisor_num
      self._supervisor_cpu = supervisor_cpu
      self._supervisor_mem = calc_size_by_str(supervisor_mem, None)
      self._supervisor_mem_limit_ratio = supervisor_mem_limit_ratio
      self._supervisor_extra_modules = _override_modules(
          kwargs.pop("supervisor_extra_modules", [])
      )
      self._supervisor_extra_env = _override_envs(
          kwargs.pop("supervisor_extra_env", None)
      )
      self._supervisor_extra_labels = _override_labels(
          kwargs.pop("supervisor_extra_labels", None)
      )
      self._supervisor_service_port = (
          kwargs.pop("supervisor_service_port", None) or service_port
      )
      self._web_port = web_port or self._default_web_port
      self._external_web_endpoint = None
  
      self._worker_num = worker_num
      self._worker_cpu = worker_cpu
      self._worker_mem = calc_size_by_str(worker_mem, None)
      self._worker_mem_limit_ratio = worker_mem_limit_ratio
      self._worker_spill_paths = worker_spill_paths
      self._worker_cache_mem = worker_cache_mem
      self._worker_min_cache_men = worker_min_cache_mem
      self._min_worker_num = min_worker_num
      self._worker_extra_modules = _override_modules(
          kwargs.pop("worker_extra_modules", [])
      )
      self._worker_extra_env = _override_envs(
          kwargs.pop("worker_extra_env", None)
      )
      self._worker_extra_labels = _override_labels(
          kwargs.pop("worker_extra_labels", None)
      )
      self._worker_service_port = (
          kwargs.pop("worker_service_port", None) or service_port
      )
  
imports: import asyncio
  import functools
  import logging
  import random
  import time
  import uuid
  from urllib.parse import urlparse
  from ...lib.aio import new_isolation, stop_isolation
  from ...services.cluster.api import WebClusterAPI
  from ...session import new_session
  from ...utils import calc_size_by_str
  from ..utils import wait_services_ready
  from .config import NamespaceConfig, RoleConfig, RoleBindingConfig, ServiceConfig, MarsSupervisorsConfig, MarsWorkersConfig
label_desc: Use image node_id 1 to create a new KubernetesCluster object  with example: obj = KubernetesCluster(kube_api_client, image, namespace, supervisor_num, supervisor_cpu, supervisor_mem, supervisor_mem_limit_ratio, worker_num, worker_cpu, worker_mem, worker_spill_paths, worker_cache_mem, min_worker_num, worker_min_cache_mem, worker_mem_limit_ratio, web_port, service_name, service_type, timeout)