code: def _start_kube_cluster(use_test_docker_file=True, **kwargs):
      _load_docker_env()
      image_name = _build_docker_images(
          use_test_docker_file=use_test_docker_file
      )
  
      temp_spill_dir = tempfile.mkdtemp(prefix="test-mars-k8s-")
      api_client = k8s_config.new_client_from_config()
      kube_api = k8s_client.CoreV1Api(api_client)
  
      cluster_client = None
      try:
          if use_test_docker_file:
              extra_volumes = [
                  HostPathVolumeConfig(
                      "mars-src-path", "/mnt/mars", MARS_ROOT
                  )
              ]
              pre_stop_command = ["rm", "/tmp/stopping.tmp"]
          else:
              extra_volumes = []
              pre_stop_command = None
  
          cluster_client = new_cluster(
              api_client,
              image=image_name,
              worker_spill_paths=[temp_spill_dir],
              extra_volumes=extra_volumes,
              pre_stop_command=pre_stop_command,
              timeout=600,
              log_when_fail=True,
              **kwargs,
          )
  
          assert cluster_client.endpoint is not None
  
          pod_items = kube_api.list_namespaced_pod(
              cluster_client.namespace
          ).to_dict()
  
          log_processes = []
          for item in pod_items["items"]:
              log_processes.append(
                  subprocess.Popen(
                      [
                          "kubectl",
                          "logs",
                          "-f",
                          "-n",
                          cluster_client.namespace,
                          item["metadata"]["name"],
                      ]
                  )
              )
  
          yield
  
          if use_test_docker_file:
              # turn off service processes with grace to get coverage data
              procs = []
              pod_items = kube_api.list_namespaced_pod(
                  cluster_client.namespace
              ).to_dict()
              for item in pod_items["items"]:
                  p = subprocess.Popen(
                      [
                          "kubectl",
                          "exec",
                          "-n",
                          cluster_client.namespace,
                          item["metadata"]["name"],
                          "--",
                          "/srv/graceful_stop.sh",
                      ]
                  )
                  procs.append(p)
              for p in procs:
                  p.wait()
  
          [p.terminate() for p in log_processes]
      finally:
          shutil.rmtree(temp_spill_dir)
          if cluster_client:
              try:
                  cluster_client.stop(wait=True, timeout=20)
              except TimeoutError:
                  pass
          _collect_coverage()
          _remove_docker_image(image_name, False)
  
imports: import glob
  import os
  import shutil
  import subprocess
  import tempfile
  import uuid
  from contextlib import contextmanager
  from distutils.spawn import find_executable
  import numpy
  import pytest
  from ....None import tensor
  from ....tests.core import mock
  from ..None import new_cluster
  from ..config import HostPathVolumeConfig
label_desc: Use image node_id 5 for calling a global function with example usage: _start_kube_cluster(use_test_docker_file) without return types