class_docstr: An orchestrator implmentation which uses Kubernetes to deploy the components such as training and rollout workers
  and Redis Pub/Sub in Coach when used in the distributed mode.
class_docstr_tok: ["An","orchestrator","implmentation","which","uses","Kubernetes","to","deploy","the","components","such","as","training","and","rollout","workers","and","Redis","Pub\/Sub","in","Coach","when","used","in","the","distributed","mode","."]
docstr: Deploys the rollout worker(s) in Kubernetes.
docstr_tok: ["Deploys","the","rollout","worker","(","s",")","in","Kubernetes","."]
code: def deploy_worker(self):
      """
      Deploys the rollout worker(s) in Kubernetes.
      """
  
      worker_params = self.params.run_type_params.get(
          str(RunType.ROLLOUT_WORKER), None
      )
      if not worker_params:
          return False
  
      # At this point, the memory backend and data store have been deployed and in the process,
      # these parameters have been updated to include things like the hostname and port the
      # service can be found at.
      worker_params.command += [
          "--memory_backend_params",
          json.dumps(self.params.memory_backend_parameters.__dict__),
      ]
      worker_params.command += [
          "--data_store_params",
          json.dumps(self.params.data_store_params.__dict__),
      ]
      worker_params.command += [
          "--num_workers",
          "{}".format(worker_params.num_replicas),
      ]
  
      name = "{}-{}".format(worker_params.run_type, uuid.uuid4())
  
      # TODO: instead of defining each container and template spec from scratch, loaded default
      # configuration and modify them as necessary depending on the store type
      if self.params.data_store_params.store_type == "nfs":
          container = k8sclient.V1Container(
              name=name,
              image=worker_params.image,
              command=worker_params.command,
              args=worker_params.arguments,
              image_pull_policy="Always",
              volume_mounts=[
                  k8sclient.V1VolumeMount(
                      name="nfs-pvc",
                      mount_path=worker_params.checkpoint_dir,
                  )
              ],
              stdin=True,
              tty=True,
          )
          template = k8sclient.V1PodTemplateSpec(
              metadata=k8sclient.V1ObjectMeta(labels={"app": name}),
              spec=k8sclient.V1PodSpec(
                  containers=[container],
                  volumes=[
                      k8sclient.V1Volume(
                          name="nfs-pvc",
                          persistent_volume_claim=self.nfs_pvc,
                      )
                  ],
                  restart_policy="Never",
              ),
          )
      elif self.params.data_store_params.store_type == "s3":
          container = k8sclient.V1Container(
              name=name,
              image=worker_params.image,
              command=worker_params.command,
              args=worker_params.arguments,
              image_pull_policy="Always",
              env=[
                  k8sclient.V1EnvVar(
                      "ACCESS_KEY_ID", self.s3_access_key
                  ),
                  k8sclient.V1EnvVar(
                      "SECRET_ACCESS_KEY", self.s3_secret_key
                  ),
              ],
              stdin=True,
              tty=True,
          )
          template = k8sclient.V1PodTemplateSpec(
              metadata=k8sclient.V1ObjectMeta(labels={"app": name}),
              spec=k8sclient.V1PodSpec(
                  containers=[container], restart_policy="Never"
              ),
          )
      elif self.params.data_store_params.store_type == "redis":
          container = k8sclient.V1Container(
              name=name,
              image=worker_params.image,
              command=worker_params.command,
              args=worker_params.arguments,
              image_pull_policy="Always",
              stdin=True,
              tty=True,
              resources=k8sclient.V1ResourceRequirements(
                  limits={
                      "cpu": "4",
                      "memory": "4Gi",
                      # "nvidia.com/gpu": "0",
                  }
              ),
          )
          template = k8sclient.V1PodTemplateSpec(
              metadata=k8sclient.V1ObjectMeta(labels={"app": name}),
              spec=k8sclient.V1PodSpec(
                  containers=[container], restart_policy="Never"
              ),
          )
      else:
          raise ValueError(
              "unexpected store type {}".format(
                  self.params.data_store_params.store_type
              )
          )
  
      job_spec = k8sclient.V1JobSpec(
          completions=worker_params.num_replicas,
          parallelism=worker_params.num_replicas,
          template=template,
      )
  
      job = k8sclient.V1Job(
          api_version="batch/v1",
          kind="Job",
          metadata=k8sclient.V1ObjectMeta(name=name),
          spec=job_spec,
      )
  
      api_client = k8sclient.BatchV1Api()
      try:
          api_client.create_namespaced_job(self.params.namespace, job)
          worker_params.orchestration_params["job_name"] = name
          return True
      except k8sclient.rest.ApiException as e:
          screen.print("Got exception: %s\n while creating Job", e)
          return False
  
imports: import os
  import uuid
  import json
  import time
  import sys
  from enum import Enum
  from typing import List
  from configparser import ConfigParser, Error
  from multiprocessing import Process
  from rl_coach.base_parameters import RunType
  from rl_coach.orchestrators.deploy import Deploy, DeployParameters
  from kubernetes import client, config
  from rl_coach.memories.backend.memory import MemoryBackendParameters
  from rl_coach.memories.backend.memory_impl import get_memory_backend
  from rl_coach.data_stores.data_store import DataStoreParameters
  from rl_coach.data_stores.data_store_impl import get_data_store
  from rl_coach.logger import screen
label_desc: Use image node_id 4 for calling the Kubernetes obj's underlying member method code with example usage: obj.deploy_worker() and returns: False, True, False