def _BuildPodManifest()

in tfx/components/infra_validator/model_server_runners/kubernetes_runner.py [0:0]
57 lines of code
12 McCabe index (conditional complexity)

  def _BuildPodManifest(self) -> k8s_client.V1Pod:
    annotations = {}
    env_vars = []
    resources = None

    if isinstance(self._serving_binary, serving_bins.TensorFlowServing):
      env_vars_dict = self._serving_binary.MakeEnvVars(
          model_path=self._model_path)
      env_vars.extend(
          k8s_client.V1EnvVar(name=key, value=value)
          for key, value in env_vars_dict.items())

    if self._config.serving_pod_overrides:
      overrides = self._config.serving_pod_overrides
      if overrides.annotations:
        annotations.update(overrides.annotations)
      if overrides.env:
        env_vars.extend(_convert_to_kube_env(env) for env in overrides.env)
      if overrides.resources:
        resources = _convert_to_resource_requirements(overrides.resources)

    service_account_name = (self._config.service_account_name or
                            self._executor_pod.spec.service_account_name)
    active_deadline_seconds = (self._config.active_deadline_seconds or
                               _DEFAULT_ACTIVE_DEADLINE_SEC)
    if active_deadline_seconds < 0:
      raise ValueError(
          'active_deadline_seconds should be > 0, but got '
          f'{active_deadline_seconds}.')

    result = k8s_client.V1Pod(
        metadata=k8s_client.V1ObjectMeta(
            generate_name=_MODEL_SERVER_POD_NAME_PREFIX,
            annotations=annotations,
            labels=self._label_dict,
            # Resources with ownerReferences are automatically deleted once all
            # its owners are deleted.
            owner_references=[
                k8s_client.V1OwnerReference(
                    api_version=self._executor_pod.api_version,
                    kind=self._executor_pod.kind,
                    name=self._executor_pod.metadata.name,
                    uid=self._executor_pod.metadata.uid,
                ),
            ],
        ),
        spec=k8s_client.V1PodSpec(
            containers=[
                k8s_client.V1Container(
                    name=_MODEL_SERVER_CONTAINER_NAME,
                    image=self._serving_binary.image,
                    env=env_vars,
                    resources=resources,
                    volume_mounts=[],
                ),
            ],
            service_account_name=service_account_name,
            # No retry in case model server container failed. Retry will happen
            # at the outermost loop (executor.py).
            restart_policy=_RestartPolicy.NEVER.value,
            # This is a hard deadline for the model server container to ensure
            # the Pod is properly cleaned up even with an unexpected termination
            # of an infra validator. After the deadline, container will be
            # removed but Pod resource won't. This makes the Pod log visible
            # after the termination.
            active_deadline_seconds=active_deadline_seconds,
            volumes=[],
            # TODO(b/152002076): Add TTL controller once it graduates Beta.
            # ttl_seconds_after_finished=,
        ))

    self._SetupModelVolumeIfNeeded(result)

    return result