__init__.py

from datetime import datetime, timedelta
import json
import logging
import os
import string
import time
from urllib.parse import urljoin
import base64

from django.conf import settings
from docker import Client
from .states import JobState
import requests
from requests_toolbelt import user_agent
from .utils import dict_merge
from retrying import retry

from deis import __version__ as deis_version


logger = logging.getLogger(__name__)

# Used for one off command runs on pods
POD_BTEMPLATE = """\
{
  "kind": "Pod",
  "apiVersion": "$version",
  "metadata": {
    "name": "$id"
  },
  "spec": {
    "containers": [
      {
        "name": "$id",
        "image": "$slugimage",
        "env": [
        {
            "name":"PORT",
            "value":"5000"
        },
        {
            "name":"SLUG_URL",
            "value":"$image"
        },
        {
            "name": "BUILDER_STORAGE",
            "value":"$storagetype"
        },
        {
            "name": "DEIS_MINIO_SERVICE_HOST",
            "value":"$mHost"
        },
        {
            "name": "DEIS_MINIO_SERVICE_PORT",
            "value":"$mPort"
        }
        ],
        "volumeMounts":[
        {
            "name":"objectstorage-keyfile",
            "mountPath":"/var/run/secrets/deis/objectstore/creds",
            "readOnly":true
        }
        ]
      }
    ],
    "volumes":[
      {
        "name":"objectstorage-keyfile",
        "secret":{
        "secretName":"objectstorage-keyfile"
        }
      }
    ],
    "terminationGracePeriodSeconds": "$terminationGracePeriodSeconds",
    "restartPolicy": "Never"
  }
}
"""

POD_TEMPLATE = """\
{
  "kind": "Pod",
  "apiVersion": "$version",
  "metadata": {
    "name": "$id"
  },
  "spec": {
    "containers": [
      {
        "name": "$id",
        "image": "$image",
        "env": []
      }
    ],
    "terminationGracePeriodSeconds": "$terminationGracePeriodSeconds",
    "restartPolicy": "Never"
  }
}
"""

RCD_TEMPLATE = """\
{
  "kind": "ReplicationController",
  "apiVersion": "$version",
  "metadata": {
    "name": "$name",
    "labels": {
      "app": "$id",
      "version": "$appversion",
      "type": "$type",
      "heritage": "deis"
    }
  },
  "spec": {
    "replicas": $replicas,
    "selector": {
      "app": "$id",
      "version": "$appversion",
      "type": "$type",
      "heritage": "deis"
    },
    "template": {
      "metadata": {
        "labels": {
          "app": "$id",
          "version": "$appversion",
          "type": "$type",
          "heritage": "deis"
        }
      },
      "spec": {
        "terminationGracePeriodSeconds": "$terminationGracePeriodSeconds",
        "containers": [
          {
            "name": "$containername",
            "image": "$image",
            "imagePullPolicy": "$image_pull_policy",
            "env": [
            {
                "name":"DEIS_APP",
                "value":"$id"
            },
            {
                "name":"DEIS_RELEASE",
                "value":"$appversion"
            }
            ]
          }
        ],
        "nodeSelector": {}
      }
    }
  }
}
"""

RCB_TEMPLATE = """\
{
  "kind": "ReplicationController",
  "apiVersion": "$version",
  "metadata": {
    "name": "$name",
    "labels": {
      "app": "$id",
      "version": "$appversion",
      "type": "$type",
      "heritage": "deis"
    }
  },
  "spec": {
    "replicas": $replicas,
    "selector": {
      "app": "$id",
      "version": "$appversion",
      "type": "$type",
      "heritage": "deis"
    },
    "template": {
      "metadata": {
        "labels": {
          "app": "$id",
          "version": "$appversion",
          "type": "$type",
          "heritage": "deis"
        }
      },
      "spec": {
        "terminationGracePeriodSeconds": "$terminationGracePeriodSeconds",
        "containers": [
          {
            "name": "$containername",
            "image": "$slugimage",
            "imagePullPolicy": "$image_pull_policy",
            "env": [
            {
                "name":"PORT",
                "value":"5000"
            },
            {
                "name":"SLUG_URL",
                "value":"$image"
            },
            {
                "name":"DEIS_APP",
                "value":"$id"
            },
            {
                "name":"DEIS_RELEASE",
                "value":"$appversion"
            },
            {
                "name": "BUILDER_STORAGE",
                "value":"$storagetype"
            },
            {
                "name": "DEIS_MINIO_SERVICE_HOST",
                "value":"$mHost"
            },
            {
                "name": "DEIS_MINIO_SERVICE_PORT",
                "value":"$mPort"
            }
            ],
            "volumeMounts":[
              {
                "name":"objectstorage-keyfile",
                "mountPath":"/var/run/secrets/deis/objectstore/creds",
                "readOnly":true
              }
            ]
          }
        ],
        "nodeSelector": {},
        "volumes":[
        {
            "name":"objectstorage-keyfile",
            "secret":{
            "secretName":"objectstorage-keyfile"
            }
        }
        ]
      }
    }
  }
}
"""

# Ports and app type will be overwritten as required
SERVICE_TEMPLATE = """\
{
  "kind": "Service",
  "apiVersion": "$version",
  "metadata": {
    "name": "$name",
    "labels": {
      "app": "$name"
    },
    "annotations": {}
  },
  "spec": {
    "ports": [
      {
        "name": "http",
        "port": 80,
        "targetPort": 8080,
        "protocol": "TCP"
      }
    ],
    "selector": {
      "app": "$name",
      "heritage": "deis"
    }
  }
}
"""

SECRET_TEMPLATE = """\
{
  "kind": "Secret",
  "apiVersion": "$version",
  "metadata": {
    "name": "$name",
    "namespace": "$id",
    "labels": {
      "app": "$id"
    }
  },
  "type": "Opaque",
  "data": {}
}
"""


class KubeException(Exception):
    def __init__(self, *args, **kwargs):
        Exception.__init__(self, *args, **kwargs)


class KubeHTTPException(KubeException):
    def __init__(self, *args, **kwargs):
        self.response = kwargs.pop('response', object)
        KubeException.__init__(self, *args, **kwargs)


def error(response, errmsg, *args):
    errmsg = errmsg.format(*args)
    errmsg = "failed to {}: {} {}\n{}".format(
        errmsg,
        response.status_code,
        response.reason,
        response.json()
    )

    raise KubeHTTPException(errmsg, response=response)


def unhealthy(status_code):
    if not 200 <= status_code <= 299:
        return True

    return False


class KubeHTTPClient(object):
    apiversion = "v1"

    def __init__(self):
        self.url = settings.SCHEDULER_URL
        self.registry = settings.REGISTRY_URL

        with open('/var/run/secrets/kubernetes.io/serviceaccount/token') as token_file:
            token = token_file.read()

        session = requests.Session()
        session.headers = {
            'Authorization': 'Bearer ' + token,
            'Content-Type': 'application/json',
            'User-Agent': user_agent('Deis Controller', deis_version)
        }
        # TODO: accessing the k8s api server by IP address rather than hostname avoids
        # TODO look at https://toolbelt.readthedocs.org/en/latest/adapters.html#fingerprintadapter
        # intermittent DNS errors, but at the price of disabling cert verification.
        # session.verify = '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt'
        session.verify = False
        self.session = session

    def deploy(self, namespace, name, image, command, **kwargs):  # noqa
        logger.debug('deploy {}, img {}, params {}, cmd "{}"'.format(name, image, kwargs, command))
        app_type = kwargs.get('app_type')
        build_type = kwargs.get('build_type')
        routable = kwargs.get('routable', False)
        port = None

        try:
            if routable:
                if build_type == "buildpack":
                    logger.debug("Using default port 5000 for build pack app {}".format(name))
                    port = 5000
                else:
                    port = self._get_port(image)
                    if port is None:
                        raise Exception("Expose a port or make the app non routable by changing"
                                        " the process type")
        except Exception as e:
            raise KubeException('{} (scheduler::deploy): {}'.format(name, e))

        # Fetch old RC and create the new one for a release
        old_rc = self._get_old_rc(namespace, app_type)

        # If an RC already exists then stop processing of the deploy
        try:
            self._get_rc(namespace, name)
            logger.debug('RC {} already exists under Namespace {}. Stopping deploy'.format(name, namespace))  # noqa
            return
        except KubeHTTPException:
            new_rc = self._create_rc(namespace, name, image, command, **kwargs)

        # Get the desired number to scale to
        if old_rc:
            desired = int(old_rc["spec"]["replicas"])
        else:
            desired = kwargs['replicas']
            logger.debug('No prior RC could be found for {}-{}'.format(namespace, app_type))

        # see if application or global deploy batches are defined
        if not kwargs.get('batches', None):
            # figure out how many nodes the application can go on
            tags = kwargs.get('tags', {})
            steps = len(self._get_nodes(labels=tags).json()['items'])
        else:
            steps = int(kwargs.get('batches'))

        # figure out what kind of batches the deploy is done in - 1 in, 1 out or higher
        if desired < steps:
            # do it all in one go
            batches = [desired]
        else:
            # figure out the stepped deploy count and then see if there is a leftover
            batches = [steps for n in set(range(1, (desired + 1))) if n % steps == 0]
            if desired - sum(batches) > 0:
                batches.append(desired - sum(batches))

        try:
            count = 0
            new_name = new_rc["metadata"]["name"]
            for batch in batches:
                count += batch
                logger.debug('scaling release {} to {} out of final {}'.format(
                    new_name, count, desired
                ))
                self._scale_rc(namespace, new_name, count)

                if old_rc:
                    old_name = old_rc["metadata"]["name"]
                    logger.debug('scaling old release {} from original {} to {}'.format(
                        old_name, desired, (desired-count))
                    )
                    self._scale_rc(namespace, old_name, (desired-count))
        except Exception as e:
            # New release is broken. Clean up
            logger.error('Could not scale {} to {}. Deleting and going back to old release'.format(
                new_rc["metadata"]["name"], desired)
            )

            # Remove new release of the RC
            self._scale_rc(namespace, new_rc["metadata"]["name"], 0)
            self._delete_rc(namespace, new_rc["metadata"]["name"])

            # Bring back old release if available of the RC
            if old_rc:
                self._scale_rc(namespace, old_rc["metadata"]["name"], desired)

            raise KubeException('{} (scheduler::deploy): {}'.format(name, e))

        # New release is live and kicking. Clean up old release
        if old_rc:
            self._scale_rc(namespace, old_rc["metadata"]["name"], 0)
            self._delete_rc(namespace, old_rc["metadata"]["name"])

        # Make sure the application is routable and uses the correct port
        # Done after the fact to let initial deploy settle before routing
        # traffic to the application
        self._update_application_service(namespace, name, app_type, port, routable)

    def _update_application_service(self, namespace, name, app_type, port, routable=False):
        """Update application service with all the various required information"""
        try:
            # Fetch service
            service = self._get_service(namespace, namespace).json()
            old_service = service.copy()  # in case anything fails for rollback

            # Update service information
            if routable:
                service['metadata']['labels']['router.deis.io/routable'] = 'true'

            # Set app type if there is not one available
            if 'type' not in service['spec']['selector']:
                service['spec']['selector']['type'] = app_type

            # Find if target port exists already, update / create as required
            if routable:
                for pos, item in enumerate(service['spec']['ports']):
                    if item['port'] == 80 and port != item['targetPort']:
                        # port 80 is the only one we care about right now
                        service['spec']['ports'][pos]['targetPort'] = port

            self._update_service(namespace, namespace, data=service)
        except Exception as e:
            # Fix service to old port and app type
            self._update_service(namespace, namespace, data=old_service)
            raise KubeException('{} (scheduler::deploy::service_update): {}'.format(name, e))

    def scale(self, namespace, name, image, command, **kwargs):
        logger.debug('scale {}, img {}, params {}, cmd "{}"'.format(name, image, kwargs, command))
        replicas = kwargs.pop('replicas')
        if unhealthy(self._get_rc_status(namespace, name)):
            # add RC if it is missing for the namespace
            try:
                # Create RC with scale as 0 and then scale to get pod monitoring
                kwargs['replicas'] = 0
                self._create_rc(namespace, name, image, command, **kwargs)
            except KubeException as e:
                logger.debug("Creating RC failed because of: {}".format(str(e)))
                raise KubeException('{} (RC): {}'.format(name, e))

        try:
            self._scale_rc(namespace, name, replicas)
        except KubeException as e:
            logger.debug("Scaling failed because of: {}".format(str(e)))
            old = self._get_rc(namespace, name).json()
            self._scale_rc(namespace, name, old['spec']['replicas'])
            raise KubeException('{} (Scale): {}'.format(name, e))

    def create(self, namespace, **kwargs):
        """Create a basic structure for an application in k8s"""
        logger.debug('create {}'.format(namespace))
        try:
            # Create essential resources
            try:
                self._get_namespace(namespace)
            except KubeException:
                self._create_namespace(namespace)

            try:
                self._get_service(namespace, namespace)
            except KubeException:
                self._create_service(namespace, namespace)
        except KubeException as e:
            # Blow it all away only if something horrible happens
            logger.debug(e)
            self._delete_namespace(namespace)
            raise

    def destroy(self, namespace):
        """Destroy a application by deleting its namespace."""
        logger.debug("destroy {}".format(namespace))
        self._delete_namespace(namespace)

        # wait 30 seconds for termination
        for _ in range(30):
            try:
                self._get_namespace(namespace).json()
            except KubeException:
                break

    def run(self, namespace, name, image, entrypoint, command, **kwargs):
        """Run a one-off command."""
        logger.debug('run {}, img {}, entrypoint {}, cmd "{}"'.format(
            name, image, entrypoint, command)
        )

        imgurl = self.registry + '/' + image
        POD = POD_TEMPLATE

        l = {
            'id': name,
            'version': self.apiversion,
            'image': imgurl,
            'image_pull_policy': settings.DOCKER_BUILDER_IMAGE_PULL_POLICY,
            'storagetype': os.getenv("APP_STORAGE"),
            'terminationGracePeriodSeconds': settings.KUBERNETES_POD_TERMINATION_GRACE_PERIOD_SECONDS  # noqa
        }

        if entrypoint == '/runner/init':
            POD = POD_BTEMPLATE
            l["image"] = image
            l['image_pull_policy'] = settings.SLUG_BUILDER_IMAGE_PULL_POLICY
            l["slugimage"] = settings.SLUGRUNNER_IMAGE
            l["mHost"] = os.getenv("DEIS_MINIO_SERVICE_HOST")
            l["mPort"] = os.getenv("DEIS_MINIO_SERVICE_PORT")

        template = json.loads(string.Template(POD).substitute(l))

        if command.startswith('-c '):
            args = command.split(' ', 1)
            args[1] = args[1][1:-1]
        else:
            args = [command[1:-1]]

        containers = template['spec']['containers'][0]
        containers['command'] = [entrypoint]
        containers['args'] = args

        self._set_environment(containers, namespace, **kwargs)

        url = self._api("/namespaces/{}/pods", namespace)
        response = self.session.post(url, json=template)
        if unhealthy(response.status_code):
            error(response, 'create Pod in Namespace "{}"', namespace)

        data = ''
        duration = 30
        iteration = 1
        while (iteration < duration):
            try:
                response = self._get_pod(namespace, name)
                data = response.text
                pod = response.json()
                if pod['status']['phase'] == 'Succeeded':
                    response = self._pod_log(namespace, name)
                    response.encoding = 'utf-8'  # defaults to "ISO-8859-1" otherwise...
                    log = response.text
                    self._delete_pod(namespace, name)
                    return 0, log

                elif pod['status']['phase'] == 'Running':
                    if iteration > 28:
                        duration += 1

                elif pod['status']['phase'] == 'Failed':
                    pod_state = pod['status']['containerStatuses'][0]['state']
                    err_code = pod_state['terminated']['exitCode']
                    self._delete_pod(namespace, name)
                    return err_code, data

            except KubeException:
                break

            iteration += 1
            time.sleep(1)

        if iteration >= duration:
            error(response, 'Pod start took more than 30 seconds', namespace)
            return 0, data

        return 0, data

    def _set_environment(self, data, namespace, **kwargs):
        app_type = kwargs.get('app_type')
        mem = kwargs.get('memory', {}).get(app_type)
        cpu = kwargs.get('cpu', {}).get(app_type)
        env = kwargs.get('envs', {})

        # create env list if missing
        if 'env' not in data:
            data['env'] = []

        if env:
            # env vars are stored in secrets and mapped to env in k8s
            try:
                # secrets use dns labels for keys, map those properly here
                secrets_env = {}
                for key, value in env.items():
                    secrets_env[key.lower().replace('_', '-')] = str(value)

                secret_name = "{}-{}-env".format(namespace, kwargs.get('version'))
                self._get_secret(namespace, secret_name)
            except KubeHTTPException:
                labels = {
                    'version': kwargs.get('version'),
                    'type': 'env'
                }
                self._create_secret(namespace, secret_name, secrets_env, labels)
            else:
                self._update_secret(namespace, secret_name, secrets_env)

            for key in env.keys():
                data["env"].append({
                    "name": key,
                    "valueFrom": {
                        "secretKeyRef": {
                            "name": secret_name,
                            # k8s doesn't allow _ so translate to -, see above
                            "key": key.lower().replace('_', '-')
                        }
                    }
                })

        # Inject debugging if workflow is in debug mode
        if os.environ.get("DEBUG", False):
            data["env"].append({
                "name": "DEBUG",
                "value": "1"
            })

        if mem or cpu:
            data["resources"] = {"limits": {}}

        if mem:
            if mem[-2:-1].isalpha() and mem[-1].isalpha():
                mem = mem[:-1]

            mem = mem + "i"
            data["resources"]["limits"]["memory"] = mem

        if cpu:
            data["resources"]["limits"]["cpu"] = cpu

    def resolve_state(self, pod):
        # See "Pod Phase" at http://kubernetes.io/v1.1/docs/user-guide/pod-states.html
        if pod is None:
            return JobState.destroyed

        states = {
            'Pending': JobState.initializing.name,
            'ContainerCreating': JobState.creating.name,
            'Starting': JobState.starting.name,
            'Running': JobState.up.name,
            'Terminating': JobState.terminating.name,
            'Succeeded': JobState.down.name,
            'Failed': JobState.crashed.name,
            'Unknown': JobState.error.name,
        }

        # being in a Pending state can mean different things, introspecting app container first
        if pod['status']['phase'] == 'Pending':
            pod_state = self._pod_pending_status(pod)
        # being in a running state can mean a pod is starting, actually running or terminating
        elif pod['status']['phase'] == 'Running':
            # is the readiness probe passing?
            pod_state = self._pod_readiness_status(pod)
            if pod_state in ['Starting', 'Terminating']:
                return states[pod_state]
            elif pod_state == 'Running' and self._pod_liveness_status(pod):
                # is the pod ready to serve requests?
                return states[pod_state]
        else:
            # if no match was found for deis mapping then passthrough the real state
            pod_state = pod['status']['phase']

        return states.get(pod_state, pod_state)

    @retry(stop_max_attempt_number=3, wait_fixed=1000)
    def _get_port(self, image):
        # try thrice to find the port before raising exception as docker-py is flaky
        imagepath = self.registry + '/' + image
        repo = imagepath.split(":")
        # image already includes the tag, so we split it out here
        docker_cli = Client(version="auto")
        docker_cli.pull(repo[0]+":"+repo[1], tag=repo[2], insecure_registry=True)
        image_info = docker_cli.inspect_image(imagepath)
        if 'ExposedPorts' not in image_info['Config']:
            return None
        port = int(list(image_info['Config']['ExposedPorts'].keys())[0].split("/")[0])
        return port

    def _api(self, tmpl, *args):
        """Return a fully-qualified Kubernetes API URL from a string template with args."""
        url = "/api/{}".format(self.apiversion) + tmpl.format(*args)
        return urljoin(self.url, url)

    def _selectors(self, **kwargs):
        query = {}

        # labels and fields are encoded slightly differently than python-requests can do
        labels = kwargs.get('labels', {})
        if labels:
            # http://kubernetes.io/v1.1/docs/user-guide/labels.html#list-and-watch-filtering
            labels = ['{}={}'.format(key, value) for key, value in labels.items()]
            query['labelSelector'] = ','.join(labels)

        fields = kwargs.get('fields', {})
        if fields:
            fields = ['{}={}'.format(key, value) for key, value in fields.items()]
            query['fieldSelector'] = ','.join(fields)

        # Which resource version to start from. Otherwise starts from the beginning
        resource_version = kwargs.get('resourceVersion', None)
        if resource_version:
            query['resourceVersion'] = resource_version

        # If output should pretty print, only True / False allowed
        pretty = bool(kwargs.get('pretty', False))
        if pretty:
            query['pretty'] = pretty

        return query

    # NAMESPACE #

    def _get_namespace_events(self, namespace, **kwargs):
        url = self._api("/namespaces/{}/events", namespace)
        response = self.session.get(url, params=self._selectors(**kwargs))
        if unhealthy(response.status_code):
            error(response, "get Events in Namespace {}", namespace)

        return response

    def _get_namespace(self, namespace):
        url = self._api("/namespaces/{}/", namespace)
        response = self.session.get(url)
        if unhealthy(response.status_code):
            error(response, 'get Namespace "{}"', namespace)

        return response

    def _get_namespaces(self, **kwargs):
        url = self._api("/namespaces")
        response = self.session.get(url, params=self._selectors(**kwargs))
        if unhealthy(response.status_code):
            error(response, 'get Namespaces')

        return response

    def _create_namespace(self, namespace):
        url = self._api("/namespaces")
        data = {
            "kind": "Namespace",
            "apiVersion": self.apiversion,
            "metadata": {
                "name": namespace
            }
        }

        response = self.session.post(url, json=data)
        if not response.status_code == 201:
            error(response, "create Namespace {}".format(namespace))

        return response

    def _delete_namespace(self, namespace):
        url = self._api("/namespaces/{}", namespace)
        response = self.session.delete(url)
        if response.status_code == 404:
            logger.warn('delete Namespace "{}": not found'.format(namespace))
        elif response.status_code != 200:
            error(response, 'delete Namespace "{}"', namespace)

        return response

    # REPLICATION CONTROLLER #

    def _get_old_rc(self, namespace, app_type):
        labels = {
            'app': namespace,
            'type': app_type
        }
        controllers = self._get_rcs(namespace, labels=labels).json()
        if len(controllers['items']) == 0:
            return False

        return controllers['items'][0]

    def _get_rc_status(self, namespace, name):
        url = self._api("/namespaces/{}/replicationcontrollers/{}", namespace, name)
        resp = self.session.get(url)
        return resp.status_code

    def _get_rc(self, namespace, name):
        url = self._api("/namespaces/{}/replicationcontrollers/{}", namespace, name)
        response = self.session.get(url)
        if unhealthy(response.status_code):
            error(response, 'get ReplicationController "{}" in Namespace "{}"', name, namespace)

        return response

    def _get_rcs(self, namespace, **kwargs):
        url = self._api("/namespaces/{}/replicationcontrollers", namespace)
        response = self.session.get(url, params=self._selectors(**kwargs))
        if unhealthy(response.status_code):
            error(response, 'get ReplicationControllers in Namespace "{}"', namespace)

        return response

    def _wait_until_pods_terminate(self, namespace, labels, current, desired):
        """Wait until all the desired pods are terminated"""
        # http://kubernetes.io/docs/api-reference/v1/definitions/#_v1_podspec
        # https://github.com/kubernetes/kubernetes/blob/release-1.2/docs/devel/api-conventions.md#metadata
        # http://kubernetes.io/docs/user-guide/pods/#termination-of-pods

        timeout = settings.KUBERNETES_POD_TERMINATION_GRACE_PERIOD_SECONDS
        delta = current - desired
        logger.debug("waiting for {} pods in {} namespace to be terminated ({}s timeout)".format(delta, namespace, timeout))  # noqa
        for waited in range(timeout):
            pods = self._get_pods(namespace, labels=labels).json()
            count = len(pods['items'])

            # see if any pods are past their terminationGracePeriodsSeconds (as in stuck)
            # seems to be a problem in k8s around that:
            # https://github.com/kubernetes/kubernetes/search?q=terminating&type=Issues
            # these will be eventually GC'ed by k8s, ignoring them for now
            for pod in pods['items']:
                # remove pod if it is passed the graceful termination period
                if self._pod_deleted(pod):
                    count -= 1

            # stop when all pods are terminated as expected
            if count == desired:
                break

            if waited > 0 and (waited % 10) == 0:
                logger.debug("waited {}s and {} pods out of {} are fully terminated".format(waited, (delta - count), delta))  # noqa

            time.sleep(1)

        logger.debug("{} pods in namespace {} are terminated".format(delta, namespace))  # noqa

    def _get_pod_ready_status(self, namespace, controller, labels, desired):
        # If desired is 0 then there is no ready state to check on
        if desired == 0:
            return

        waited = 0
        timeout = 120  # 2 minutes
        # If there is initial delay on the readiness check then timeout needs to be higher
        # this is to account for kubernetes having readiness check report as failure until
        # the initial delay period is up
        delay = 0
        container_name = '{}-{}'.format(
            controller['metadata']['labels']['app'],
            controller['metadata']['labels']['type']
        )
        # get health info from spec
        for container in controller['spec']['template']['spec']['containers']:
            if container['name'] != container_name or 'readinessProbe' not in container:
                continue

            delay = int(container['readinessProbe']['initialDelaySeconds'])
            logger.debug("adding {}s on to the original {}s timeout to account for the initial delay specified in the readiness probe for the RC".format(delay, timeout, controller['metadata']['name']))  # noqa
            timeout += delay

        logger.debug("waiting for {} pods in {} namespace to be in services ({} timeout)".format(desired, namespace, timeout))  # noqa

        # has timeout been increased or not within the loop
        timeout_padded = False
        # Ensure the minimum desired number of pods are available
        while True:
            # timed out, time to bail
            if waited > timeout:
                logger.debug('timed out waiting for pods to come up in namespace {}'.format(namespace))  # noqa
                break

            count = 0  # ready pods
            pods = self._get_pods(namespace, labels=labels).json()
            for pod in pods['items']:
                # If pulling an image is taking long then increase the timeout
                if (
                    pod['status']['phase'] == 'Pending' and
                    self._pod_pending_status(pod) == 'Pulling' and
                    not timeout_padded
                ):
                    # last event should be Pulling in this case
                    event = self._pod_events(pod).pop()
                    # see if pull operation has been happening for over 1 minute
                    start = datetime.strptime(
                        event['firstTimestamp'],
                        settings.DEIS_DATETIME_FORMAT
                    )

                    seconds = 60
                    if (start + timedelta(seconds=seconds)) < datetime.utcnow():
                        # add 10 minutes to timeout to allow a pull image operation to finish
                        logger.debug('Kubernetes has been pulling the image for {} seconds'.format(seconds))  # noqa
                        logger.debug('Increasing timeout by 10 minutes to allow a pull image operation to finish for pods in namespace {}'.format(namespace))  # noqa
                        timeout += (60 * 10)
                        timeout_padded = True

                # now that state is running time to see if probes are passing
                if self._pod_ready(pod):
                    count += 1

            if count == desired:
                break

            if waited > 0 and (waited % 10) == 0:
                logger.debug("waited {}s and {} pods are in service".format(waited, count))

            # increase wait time without dealing with jitters from above code
            waited += 1
            time.sleep(1)

        logger.debug("{} out of {} pods in namespace {} are in service".format(count, desired, namespace))  # noqa

    def _scale_rc(self, namespace, name, desired):
        rc = self._get_rc(namespace, name).json()

        # get the current replica count by querying for pods instead of introspecting RC
        labels = {
            'app': rc['spec']['selector']['app'],
            'type': rc['spec']['selector']['type'],
            'version': rc['spec']['selector']['version']
        }

        # Are there any pods running (and verified as ready) available?
        pods = self._get_pods(namespace, labels=labels).json()['items']
        current = 0
        for pod in pods:
            if self._pod_ready(pod):
                current += 1

        if desired == current:
            logger.debug("Not scaling RC {} in Namespace {} to {} replicas. Already at desired replicas".format(name, namespace, desired))  # noqa
            return
        elif desired != rc['spec']['replicas']:  # RC needs new replica count
            # Set the new desired replica count
            rc['spec']['replicas'] = desired

            logger.debug("scaling RC {} in Namespace {} from {} to {} replicas".format(name, namespace, current, desired))  # noqa

            self._update_rc(namespace, name, rc)

            resource_ver = rc['metadata']['resourceVersion']
            logger.debug("waiting for RC {} to get a newer resource version than {} (30s timeout)".format(name, resource_ver))  # noqa
            for waited in range(30):
                js_template = self._get_rc(namespace, name).json()
                if js_template["metadata"]["resourceVersion"] != resource_ver:
                    break

                if waited > 0 and (waited % 10) == 0:
                    logger.debug("waited {}s so far for a new resource version".format(waited))

                time.sleep(1)

            logger.debug("RC {} has a new resource version {}".format(name, js_template["metadata"]["resourceVersion"]))  # noqa

        # Double check enough pods are in the required state to service the application
        self._get_pod_ready_status(namespace, rc, labels, desired)

        # if it was a scale down operation, wait until terminating pods are done
        if int(desired) < int(current):
            self._wait_until_pods_terminate(namespace, labels, current, desired)

    def _create_rc(self, namespace, name, image, command, **kwargs):  # noqa
        app_type = kwargs.get('app_type')
        container_name = namespace + '-' + app_type
        args = command.split()
        imgurl = self.registry + "/" + image
        storageType = os.getenv("APP_STORAGE")
        TEMPLATE = RCD_TEMPLATE

        l = {
            "name": name,
            "id": namespace,
            "appversion": kwargs.get("version"),
            "version": self.apiversion,
            "image": imgurl,
            'image_pull_policy': settings.DOCKER_BUILDER_IMAGE_PULL_POLICY,
            "replicas": kwargs.get("replicas", 0),
            "containername": container_name,
            "type": app_type,
            "storagetype": storageType,
            "mHost": os.getenv("DEIS_MINIO_SERVICE_HOST"),
            "mPort": os.getenv("DEIS_MINIO_SERVICE_PORT"),
            "terminationGracePeriodSeconds": settings.KUBERNETES_POD_TERMINATION_GRACE_PERIOD_SECONDS  # noqa
        }

        # Check if it is a slug builder image.
        if kwargs.get('build_type') == "buildpack":
            # only buildpack apps need access to object storage
            try:
                self._get_secret(namespace, 'objectstorage-keyfile')
            except KubeException:
                secret = self._get_secret('deis', 'objectstorage-keyfile').json()
                self._create_secret(namespace, 'objectstorage-keyfile', secret['data'])

            l["image"] = image
            l['image_pull_policy'] = settings.SLUG_BUILDER_IMAGE_PULL_POLICY
            l["slugimage"] = settings.SLUGRUNNER_IMAGE
            TEMPLATE = RCB_TEMPLATE

        template = json.loads(string.Template(TEMPLATE).substitute(l))

        # apply tags as needed
        tags = kwargs.get('tags', {})
        template["spec"]["template"]["spec"]["nodeSelector"] = tags

        # Deal with container information
        container = template["spec"]["template"]["spec"]["containers"][0]
        container['args'] = args

        self._set_environment(container, namespace, **kwargs)

        # add in healthchecks
        if kwargs.get('healthcheck'):
            template = self._healthcheck(template, kwargs['routable'], **kwargs['healthcheck'])
        else:
            template = self._default_readiness_probe(template, image, kwargs.get('build_type'))

        url = self._api("/namespaces/{}/replicationcontrollers", namespace)
        resp = self.session.post(url, json=template)
        if unhealthy(resp.status_code):
            error(resp, 'create ReplicationController "{}" in Namespace "{}"', name, namespace)
            logger.debug('template used: {}'.format(json.dumps(template, indent=4)))

        for _ in range(30):
            try:
                rc = self._get_rc(namespace, name).json()
                if (
                    "observedGeneration" in rc["status"] and
                    rc["metadata"]["generation"] == rc["status"]["observedGeneration"]
                ):
                    break

                time.sleep(1)
            except KubeHTTPException as e:
                if e.response.status_code == 404:
                    time.sleep(1)

        return resp.json()

    def _update_rc(self, namespace, name, data):
        url = self._api("/namespaces/{}/replicationcontrollers/{}", namespace, name)
        response = self.session.put(url, json=data)
        if unhealthy(response.status_code):
            error(response, 'scale ReplicationController "{}"', name)

        return response

    def _delete_rc(self, namespace, name):
        url = self._api("/namespaces/{}/replicationcontrollers/{}", namespace, name)
        response = self.session.delete(url)
        if unhealthy(response.status_code):
            error(response, 'delete ReplicationController "{}" in Namespace "{}"',
                  name, namespace)

        return response

    def _healthcheck(self, controller, routable=False, path='/', port=5000, delay=30, timeout=5,
                     period_seconds=1, success_threshold=1, failure_threshold=3):  # noqa
        """
        Apply HTTP GET healthcehck to the application container

        http://kubernetes.io/docs/user-guide/walkthrough/k8s201/#health-checking
        http://kubernetes.io/docs/user-guide/pod-states/#container-probes
        http://kubernetes.io/docs/user-guide/liveness/
        """
        if not routable:
            return controller

        namespace = controller['spec']['selector']['app']
        # Inspect if a PORT env is already defined, make sure that's the port used
        try:
            service = self._get_service(namespace, namespace).json()
            port = service['spec']['ports'][0]['targetPort']
        except:
            pass

        # Only support HTTP checks for now
        # http://kubernetes.io/v1.1/docs/user-guide/pod-states.html#container-probes
        healthcheck = {
            # defines the health checking
            'livenessProbe': {
                # an http probe
                'httpGet': {
                    'path': path,
                    'port': port
                },
                # length of time to wait for a pod to initialize
                # after pod startup, before applying health checking
                'initialDelaySeconds': delay,
                'timeoutSeconds': timeout,
                'periodSeconds': period_seconds,
                'successThreshold': success_threshold,
                'failureThreshold': failure_threshold,
            },
            'readinessProbe': {
                # an http probe
                'httpGet': {
                    'path': path,
                    'port': port
                },
                # length of time to wait for a pod to initialize
                # after pod startup, before applying health checking
                'initialDelaySeconds': delay,
                'timeoutSeconds': timeout,
                'periodSeconds': period_seconds,
                'successThreshold': success_threshold,
                'failureThreshold': failure_threshold,
            },
        }

        # Update only the application container with the health check
        app_type = controller['spec']['selector']['type']
        container_name = '{}-{}'.format(namespace, app_type)
        containers = controller['spec']['template']['spec']['containers']
        for container in containers:
            if container['name'] == container_name:
                container.update(healthcheck)

        return controller

    def _default_readiness_probe(self, controller, image, build_type):
        if build_type == "buildpack":
            readinessprobe = self._default_buildpack_readiness_probe()
        else:
            readinessprobe = self._default_dockerapp_readiness_probe(image)
        if readinessprobe is None:
            return controller
        # Update only the application container with the health check
        app_type = controller['spec']['selector']['type']
        namespace = controller['spec']['selector']['app']
        container_name = '{}-{}'.format(namespace, app_type)
        containers = controller['spec']['template']['spec']['containers']
        for container in containers:
            if container['name'] == container_name:
                container.update(readinessprobe)

        return controller

    '''
    Applies exec readiness probe to the slugrunner container.
    http://kubernetes.io/docs/user-guide/pod-states/#container-probes

    /runner/init is the entry point of the slugrunner.
    https://github.com/deis/slugrunner/blob/01eac53f1c5f1d1dfa7570bbd6b9e45c00441fea/rootfs/Dockerfile#L20
    Once it downloads the slug it starts running using `exec` which means the pid 1
    will point to the slug/application command instead of entry point once the application has
    started.
    https://github.com/deis/slugrunner/blob/01eac53f1c5f1d1dfa7570bbd6b9e45c00441fea/rootfs/runner/init#L90

    This should be added only for the build pack apps when a custom liveness probe is not set to
    make sure that the pod is ready only when the slug is downloaded and started running.
    '''
    def _default_buildpack_readiness_probe(self, delay=30, timeout=5, period_seconds=5,
                                           success_threshold=1, failure_threshold=1):
        readinessprobe = {
            'readinessProbe': {
                # an exec probe
                'exec': {
                    "command": [
                        "bash",
                        "-c",
                        "[[ '$(ps -p 1 -o args)' != *'bash /runner/init'* ]]"
                    ]
                },
                # length of time to wait for a pod to initialize
                # after pod startup, before applying health checking
                'initialDelaySeconds': delay,
                'timeoutSeconds': timeout,
                'periodSeconds': period_seconds,
                'successThreshold': success_threshold,
                'failureThreshold': failure_threshold,
            },
        }
        return readinessprobe

    '''
    Applies tcp socket readiness probe to the docker app container only if some port is exposed
    by the docker image.
    '''
    def _default_dockerapp_readiness_probe(self, image, delay=5, timeout=5, period_seconds=5,
                                           success_threshold=1, failure_threshold=1):
        try:
            port = self._get_port(image)
            if port is None:
                return None
        except Exception:
            return None
        readinessprobe = {
            'readinessProbe': {
                # an exec probe
                'tcpSocket': {
                    "port": port
                },
                # length of time to wait for a pod to initialize
                # after pod startup, before applying health checking
                'initialDelaySeconds': delay,
                'timeoutSeconds': timeout,
                'periodSeconds': period_seconds,
                'successThreshold': success_threshold,
                'failureThreshold': failure_threshold,
            },
        }
        return readinessprobe

    # SECRETS #
    # http://kubernetes.io/v1.1/docs/api-reference/v1/definitions.html#_v1_secret
    def _get_secret(self, namespace, name):
        url = self._api("/namespaces/{}/secrets/{}", namespace, name)
        response = self.session.get(url)
        if unhealthy(response.status_code):
            error(response, 'get Secret "{}" in Namespace "{}"', name, namespace)

        # decode the base64 data
        secrets = response.json()
        for key, value in secrets['data'].items():
            secrets['data'][key] = base64.b64decode(value).decode(encoding='UTF-8')

        # tell python-requests it actually hasn't consumed the data
        response._content = bytes(json.dumps(secrets), 'UTF-8')

        return response

    def _get_secrets(self, namespace, **kwargs):
        url = self._api('/namespaces/{}/secrets', namespace)
        response = self.session.get(url, params=self._selectors(**kwargs))
        if unhealthy(response.status_code):
            error(response, 'get Secrets in Namespace "{}"', namespace)

        return response

    def _create_secret(self, namespace, name, data, labels={}):
        template = json.loads(string.Template(SECRET_TEMPLATE).substitute({
            "version": self.apiversion,
            "id": namespace,
            "name": name
        }))

        # add in any additional label info
        template['metadata']['labels'].update(labels)

        for key, value in data.items():
            value = value if isinstance(value, bytes) else bytes(value, 'UTF-8')
            item = base64.b64encode(value).decode(encoding='UTF-8')
            template["data"].update({key: item})

        url = self._api("/namespaces/{}/secrets", namespace)
        response = self.session.post(url, json=template)
        if unhealthy(response.status_code):
            error(response, 'failed to create Secret "{}" in Namespace "{}"', name, namespace)

        return response

    def _update_secret(self, namespace, name, data):
        template = json.loads(string.Template(SECRET_TEMPLATE).substitute({
            "version": self.apiversion,
            "id": namespace,
            "name": name
        }))

        for key, value in data.items():
            value = value if isinstance(value, bytes) else bytes(value, 'UTF-8')
            item = base64.b64encode(value).decode(encoding='UTF-8')
            template["data"].update({key: item})

        url = self._api("/namespaces/{}/secrets/{}", namespace, name)
        response = self.session.put(url, json=template)
        if unhealthy(response.status_code):
            error(response, 'failed to update Secret "{}" in Namespace "{}"', name, namespace)

        return response

    def _delete_secret(self, namespace, name):
        url = self._api("/namespaces/{}/secrets/{}", namespace, name)
        response = self.session.delete(url)
        if unhealthy(response.status_code):
            error(response, 'delete Secret "{}" in Namespace "{}"', name, namespace)

        return response

    # SERVICES #

    def _get_service(self, namespace, name):
        url = self._api("/namespaces/{}/services/{}", namespace, name)
        response = self.session.get(url)
        if unhealthy(response.status_code):
            error(response, 'get Service "{}" in Namespace "{}"', name, namespace)

        return response

    def _get_services(self, namespace, **kwargs):
        url = self._api('/namespaces/{}/services', namespace)
        response = self.session.get(url, params=self._selectors(**kwargs))
        if unhealthy(response.status_code):
            error(response, 'get Services in Namespace "{}"', namespace)

        return response

    def _create_service(self, namespace, name, data={}, **kwargs):
        l = {
            "version": self.apiversion,
            "name": namespace,
        }

        # Merge external data on to the prefined template
        template = json.loads(string.Template(SERVICE_TEMPLATE).substitute(l))
        data = dict_merge(template, data)
        url = self._api("/namespaces/{}/services", namespace)
        response = self.session.post(url, json=data)
        if unhealthy(response.status_code):
            error(response, 'create Service "{}" in Namespace "{}"', namespace, namespace)

        return response

    def _update_service(self, namespace, name, data):
        url = self._api("/namespaces/{}/services/{}", namespace, name)
        response = self.session.put(url, json=data)
        if unhealthy(response.status_code):
            error(response, 'update Service "{}" in Namespace "{}"', namespace, name)

        return response

    def _delete_service(self, namespace, name):
        url = self._api("/namespaces/{}/services/{}", namespace, name)
        response = self.session.delete(url)
        if unhealthy(response.status_code):
            error(response, 'delete Service "{}" in Namespace "{}"', name, namespace)

        return response

    # PODS #

    def _get_pod(self, namespace, name):
        url = self._api("/namespaces/{}/pods/{}", namespace, name)
        response = self.session.get(url)
        if unhealthy(response.status_code):
            error(response, 'get Pod "{}" in Namespace "{}"', name, namespace)

        return response

    def _get_pods(self, namespace, **kwargs):
        url = self._api('/namespaces/{}/pods', namespace)
        response = self.session.get(url, params=self._selectors(**kwargs))
        if unhealthy(response.status_code):
            error(response, 'get Pods in Namespace "{}"', namespace)

        return response

    def _delete_pod(self, namespace, name):
        url = self._api("/namespaces/{}/pods/{}", namespace, name)
        resp = self.session.delete(url)
        if unhealthy(resp.status_code):
            error(resp, 'delete Pod "{}" in Namespace "{}"', name, namespace)

        # Verify the pod has been deleted
        # Only wait as long as the grace period is - k8s will eventually GC
        for _ in range(settings.KUBERNETES_POD_TERMINATION_GRACE_PERIOD_SECONDS):
            try:
                pod = self._get_pod(namespace, name).json()
                # hide pod if it is passed the graceful termination period
                if self._pod_deleted(pod):
                    return
            except KubeHTTPException as e:
                if e.response.status_code == 404:
                    break

            time.sleep(1)

    def _pod_log(self, namespace, name):
        url = self._api("/namespaces/{}/pods/{}/log", namespace, name)
        response = self.session.get(url)
        if unhealthy(response.status_code):
            error(response, 'get logs for Pod "{}" in Namespace "{}"', name, namespace)

        return response

    def _pod_pending_status(self, pod):
        """Introspect the pod containers when pod is in Pending state"""
        if 'containerStatuses' not in pod['status']:
            return 'Pending'

        name = '{}-{}'.format(pod['metadata']['labels']['app'], pod['metadata']['labels']['type'])
        for container in pod['status']['containerStatuses']:
            # find the right container in case there are many on the pod
            if container['name'] != name:
                continue

            if 'waiting' in container['state']:
                reason = container['state']['waiting']['reason']
                if reason == 'ContainerCreating':
                    # get the last event
                    event = self._pod_events(pod).pop()
                    return event['reason']

        # Return Pending if nothing else can be found
        return 'Pending'

    def _pod_events(self, pod):
        """Process events for a given Pod to find if Pulling is happening, among other events"""
        # fetch all events for this pod
        fields = {
            'involvedObject.name': pod['metadata']['name'],
            'involvedObject.namespace': pod['metadata']['namespace'],
            'involvedObject.uid': pod['metadata']['uid']
        }
        events = self._get_namespace_events(pod['metadata']['namespace'], fields=fields).json()
        # make sure that events are sorted
        events['items'].sort(key=lambda x: x['lastTimestamp'])
        return events['items']

    def _pod_readiness_status(self, pod):
        """Check if the pod container have passed the readiness probes"""
        name = '{}-{}'.format(pod['metadata']['labels']['app'], pod['metadata']['labels']['type'])
        for container in pod['status']['containerStatuses']:
            # find the right container in case there are many on the pod
            if container['name'] != name:
                continue

            if not container['ready']:
                if 'running' in container['state'].keys():
                    return 'Starting'

                if (
                    'terminated' in container['state'].keys() or
                    'deletionTimestamp' in pod['metadata']
                ):
                    return 'Terminating'
            else:
                # See if k8s is in Terminating state
                if 'deletionTimestamp' in pod['metadata']:
                    return 'Terminating'

                return 'Running'

        # Seems like the most sensible default
        return 'Unknown'

    def _pod_liveness_status(self, pod):
        """Check if the pods liveness probe status has passed all checks"""
        for condition in pod['status']['conditions']:
            # type = Ready is the only binary type right now
            if condition['type'] == 'Ready' and condition['status'] != 'True':
                return False

        return True

    def _pod_ready(self, pod):
        """Combines various checks to see if the pod is considered up or not by checking probes"""
        return (
            pod['status']['phase'] == 'Running' and
            # is the readiness probe passing?
            self._pod_readiness_status(pod) == 'Running' and
            # is the pod ready to serve requests?
            self._pod_liveness_status(pod)
        )

    def _pod_deleted(self, pod):
        """Checks if a pod is deleted and past its graceful termination period"""
        # https://github.com/kubernetes/kubernetes/blob/release-1.2/docs/devel/api-conventions.md#metadata
        # http://kubernetes.io/docs/user-guide/pods/#termination-of-pods
        if 'deletionTimestamp' in pod['metadata']:
            deletion = datetime.strptime(
                pod['metadata']['deletionTimestamp'],
                settings.DEIS_DATETIME_FORMAT
            )

            # past the graceful deletion period
            if deletion < datetime.utcnow():
                return True

        return False

    # NODES #

    def _get_nodes(self, **kwargs):
        url = self._api('/nodes')
        response = self.session.get(url, params=self._selectors(**kwargs))
        if unhealthy(response.status_code):
            error(response, 'get Nodes')

        return response

    def _get_node(self, name, **kwargs):
        url = self._api('/nodes/{}'.format(name))
        response = self.session.get(url)
        if unhealthy(response.status_code):
            error(response, 'get Node {} in Nodes'.format(name))

        return response


SchedulerClient = KubeHTTPClient