Merge pull request #4127 from bacongobbler/3813-http-healthchecks

feat(platform): add http liveness and readiness health checks

Author: Matthew Fisher

controller/api/exceptions.py

@@ -0,0 +1,3 @@
+class HealthcheckException(Exception):
+    """Exception class used for when the application's health check fails"""
+    pass

controller/api/models.py

@@ -31,7 +31,7 @@
 import requests
 from rest_framework.authtoken.models import Token
 
-from api import fields, utils
+from api import fields, utils, exceptions
 from registry import publish_release
 from utils import dict_diff, fingerprint
 
@@ -118,6 +118,20 @@ def validate_certificate(value):
         raise ValidationError('Could not load certificate: {}'.format(e))
 
 
+def get_etcd_client():
+    if not hasattr(get_etcd_client, "client"):
+        # wire up etcd publishing if we can connect
+        try:
+            get_etcd_client.client = etcd.Client(
+                host=settings.ETCD_HOST,
+                port=int(settings.ETCD_PORT))
+            get_etcd_client.client.get('/deis')
+        except etcd.EtcdException:
+            logger.log(logging.WARNING, 'Cannot synchronize with etcd cluster')
+            get_etcd_client.client = None
+    return get_etcd_client.client
+
+
 class AuditedModel(models.Model):
     """Add created and updated fields to a model."""
 
@@ -353,6 +367,58 @@ def _start_containers(self, to_add):
         if set([c.state for c in to_add]) != set(['up']):
             err = 'warning, some containers failed to start'
             log_event(self, err, logging.WARNING)
+        # if the user specified a health check, try checking to see if it's running
+        try:
+            config = self.config_set.latest()
+            if 'HEALTHCHECK_URL' in config.values.keys():
+                self._healthcheck(to_add, config.values)
+        except Config.DoesNotExist:
+            pass
+
+    def _healthcheck(self, containers, config):
+        # if at first it fails, back off and try again at 10%, 50% and 100% of INITIAL_DELAY
+        intervals = [1.0, 0.1, 0.5, 1.0]
+        # HACK (bacongobbler): we need to wait until publisher has a chance to publish each
+        # service to etcd, which can take up to 20 seconds.
+        time.sleep(20)
+        for i in range(0, 4):
+            delay = int(config.get('HEALTHCHECK_INITIAL_DELAY', 0))
+            try:
+                # sleep until the initial timeout is over
+                if delay > 0:
+                    time.sleep(delay * intervals[i])
+                self._do_healthcheck(containers, config)
+                break
+            except exceptions.HealthcheckException as e:
+                try:
+                    new_delay = delay * intervals[i+1]
+                    msg = "{}; trying again in {} seconds".format(e, new_delay)
+                    log_event(self, msg, logging.WARNING)
+                except IndexError:
+                    log_event(self, e, logging.WARNING)
+        else:
+            self._destroy_containers(containers)
+            msg = "aborting, app containers failed to respond to health check"
+            log_event(self, msg, logging.ERROR)
+            raise RuntimeError(msg)
+
+    def _do_healthcheck(self, containers, config):
+        path = config.get('HEALTHCHECK_URL', '/')
+        timeout = int(config.get('HEALTHCHECK_TIMEOUT', 1))
+        if not _etcd_client:
+            raise exceptions.HealthcheckException('no etcd client available')
+        for container in containers:
+            try:
+                key = "/deis/services/{self}/{container.job_id}".format(**locals())
+                url = "http://{}{}".format(_etcd_client.get(key).value, path)
+                response = requests.get(url, timeout=timeout)
+                if response.status_code != requests.codes.OK:
+                    raise exceptions.HealthcheckException(
+                        "app failed health check (got '{}', expected: '200')".format(
+                            response.status_code))
+            except (requests.Timeout, requests.ConnectionError, KeyError) as e:
+                raise exceptions.HealthcheckException(
+                    'failed to connect to container ({})'.format(e))
 
     def _restart_containers(self, to_restart):
         """Restarts containers via the scheduler"""
@@ -452,7 +518,7 @@ def _default_scale(self, user, release):
 
         self.scale(user, structure)
 
-    def logs(self, log_lines):
+    def logs(self, log_lines=str(settings.LOG_LINES)):
         """Return aggregated log data for this application."""
         path = os.path.join(settings.DEIS_LOG_DIR, self.id + '.log')
         if not os.path.exists(path):
@@ -513,7 +579,7 @@ def _scheduler(self):
 
     @property
     def state(self):
-        return self._scheduler.state(self._job_id).name
+        return self._scheduler.state(self.job_id).name
 
     def short_name(self):
         return "{}.{}.{}".format(self.app.id, self.type, self.num)
@@ -526,15 +592,10 @@ class Meta:
         get_latest_by = '-created'
         ordering = ['created']
 
-    def _get_job_id(self):
-        app = self.app.id
-        release = self.release
-        version = "v{}".format(release.version)
-        num = self.num
-        job_id = "{app}_{version}.{self.type}.{num}".format(**locals())
-        return job_id
-
-    _job_id = property(_get_job_id)
+    @property
+    def job_id(self):
+        version = "v{}".format(self.release.version)
+        return "{self.app.id}_{version}.{self.type}.{self.num}".format(**locals())
 
     def _get_command(self):
         try:
@@ -566,45 +627,41 @@ def create(self):
         kwargs = {'memory': self.release.config.memory,
                   'cpu': self.release.config.cpu,
                   'tags': self.release.config.tags}
-        job_id = self._job_id
         try:
             self._scheduler.create(
-                name=job_id,
+                name=self.job_id,
                 image=image,
                 command=self._command,
                 **kwargs)
         except Exception as e:
-            err = '{} (create): {}'.format(job_id, e)
+            err = '{} (create): {}'.format(self.job_id, e)
             log_event(self.app, err, logging.ERROR)
             raise
 
     @close_db_connections
     def start(self):
-        job_id = self._job_id
         try:
-            self._scheduler.start(job_id)
+            self._scheduler.start(self.job_id)
         except Exception as e:
-            err = '{} (start): {}'.format(job_id, e)
+            err = '{} (start): {}'.format(self.job_id, e)
             log_event(self.app, err, logging.WARNING)
             raise
 
     @close_db_connections
     def stop(self):
-        job_id = self._job_id
         try:
-            self._scheduler.stop(job_id)
+            self._scheduler.stop(self.job_id)
         except Exception as e:
-            err = '{} (stop): {}'.format(job_id, e)
+            err = '{} (stop): {}'.format(self.job_id, e)
             log_event(self.app, err, logging.ERROR)
             raise
 
     @close_db_connections
     def destroy(self):
-        job_id = self._job_id
         try:
-            self._scheduler.destroy(job_id)
+            self._scheduler.destroy(self.job_id)
         except Exception as e:
-            err = '{} (destroy): {}'.format(job_id, e)
+            err = '{} (destroy): {}'.format(self.job_id, e)
             log_event(self.app, err, logging.ERROR)
             raise
 
@@ -614,7 +671,6 @@ def run(self, command):
             raise EnvironmentError('No build associated with this release '
                                    'to run this command')
         image = self.release.image
-        job_id = self._job_id
         entrypoint = '/bin/bash'
         # if this is a procfile-based app, switch the entrypoint to slugrunner's default
         # FIXME: remove slugrunner's hardcoded entrypoint
@@ -626,10 +682,10 @@ def run(self, command):
         else:
             command = "-c '{}'".format(command)
         try:
-            rc, output = self._scheduler.run(job_id, image, entrypoint, command)
+            rc, output = self._scheduler.run(self.job_id, image, entrypoint, command)
             return rc, output
         except Exception as e:
-            err = '{} (run): {}'.format(job_id, e)
+            err = '{} (run): {}'.format(self.job_id, e)
             log_event(self.app, err, logging.ERROR)
             raise
 
@@ -1103,6 +1159,34 @@ def _etcd_purge_cert(**kwargs):
         pass
 
 
+def _etcd_publish_config(**kwargs):
+    config = kwargs['instance']
+    # we purge all existing config when adding the newest instance. This is because
+    # deis config:unset would remove an existing value, but not delete the
+    # old config object
+    try:
+        _etcd_client.delete('/deis/config/{}'.format(config.app),
+                            prevExist=True, dir=True, recursive=True)
+    except KeyError:
+        pass
+    if kwargs['created']:
+        for k, v in config.values.iteritems():
+            _etcd_client.write(
+                '/deis/config/{}/{}'.format(
+                    config.app,
+                    unicode(k).encode('utf-8').lower()),
+                unicode(v).encode('utf-8'))
+
+
+def _etcd_purge_config(**kwargs):
+    config = kwargs['instance']
+    try:
+        _etcd_client.delete('/deis/config/{}'.format(config.app),
+                            prevExist=True, dir=True, recursive=True)
+    except KeyError:
+        pass
+
+
 def _etcd_publish_domains(**kwargs):
     domain = kwargs['instance']
     if kwargs['created']:
@@ -1134,13 +1218,9 @@ def create_auth_token(sender, instance=None, created=False, **kwargs):
     if created:
         Token.objects.create(user=instance)
 
-# wire up etcd publishing if we can connect
-try:
-    _etcd_client = etcd.Client(host=settings.ETCD_HOST, port=int(settings.ETCD_PORT))
-    _etcd_client.get('/deis')
-except etcd.EtcdException:
-    logger.log(logging.WARNING, 'Cannot synchronize with etcd cluster')
-    _etcd_client = None
+
+_etcd_client = get_etcd_client()
+
 
 if _etcd_client:
     post_save.connect(_etcd_publish_key, sender=Key, dispatch_uid='api.models')
@@ -1152,3 +1232,5 @@ def create_auth_token(sender, instance=None, created=False, **kwargs):
     post_delete.connect(_etcd_purge_app, sender=App, dispatch_uid='api.models')
     post_save.connect(_etcd_publish_cert, sender=Certificate, dispatch_uid='api.models')
     post_delete.connect(_etcd_purge_cert, sender=Certificate, dispatch_uid='api.models')
+    post_save.connect(_etcd_publish_config, sender=Config, dispatch_uid='api.models')
+    post_delete.connect(_etcd_purge_config, sender=Config, dispatch_uid='api.models')