Merge pull request #3874 from bacongobbler/expose-config-vars-to-etcd

feat(publisher): check app config for HTTP health check

Author: Matthew Fisher

controller/api/models.py

@@ -299,6 +299,45 @@ 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():
+                # check the app to see if it's healthy
+                try:
+                    self._do_healthcheck(config)
+                except Exception as e:
+                    log_event(self, str(e), logging.ERROR)
+                    self._destroy_containers(to_add)
+                    raise
+        except Config.DoesNotExist:
+            pass
+
+    def _do_healthcheck(self, config):
+        timeout = time.time() + 60  # 1 minute from now
+        while True:
+            if time.time() > timeout:
+                raise RuntimeError(
+                    'app failed to respond to health check within 60 seconds of launch')
+            try:
+                response = requests.get(
+                    'http://{}{}'.format(
+                        self.url,
+                        config.values['HEALTHCHECK_URL']),
+                    timeout=5)
+                expected_status_code = config.values['HEALTHCHECK_STATUS_CODE'] if \
+                    'HEALTHCHECK_STATUS_CODE' in config.values.keys() else \
+                    settings.DEIS_HEALTHCHECK_STATUS_CODE
+                if str(response.status_code) != str(expected_status_code):
+                    err = "aborting, app failed health check (got '{}', expected: '{}')".format(
+                        response.status_code,
+                        expected_status_code)
+                    raise RuntimeError(err)
+                break
+            except requests.exceptions.ConnectionError:
+                continue
+            except requests.exceptions.Timeout:
+                continue
 
     def _restart_containers(self, to_restart):
         """Restarts containers via the scheduler"""
@@ -1004,6 +1043,34 @@ def _etcd_purge_app(**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/services/{}/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/services/{}/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/services/{}/config'.format(config.app),
+                            prevExist=True, dir=True, recursive=True)
+    except KeyError:
+        pass
+
+
 def _etcd_publish_cert(**kwargs):
     cert = kwargs['instance']
     if kwargs['created']:
@@ -1069,3 +1136,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')

controller/api/tests/test_config.py

@@ -15,16 +15,38 @@
 from django.test import TransactionTestCase
 from rest_framework.authtoken.models import Token
 
-from api.models import Config
+from api.models import App, Config
 
 
-def mock_import_repository_task(*args, **kwargs):
+def mock_status_ok(*args, **kwargs):
     resp = requests.Response()
     resp.status_code = 200
     resp._content_consumed = True
     return resp
 
 
+def mock_status_not_found(*args, **kwargs):
+    resp = requests.Response()
+    resp.status_code = 404
+    resp._content_consumed = True
+    return resp
+
+
+def mock_request_timed_out(*args, **kwargs):
+    raise requests.exceptions.Timeout()
+
+
+def mock_request_connection_error(*args, **kwargs):
+    raise requests.exceptions.ConnectionError()
+
+
+def mock_time(*args, **kwargs):
+    if not hasattr(mock_time, "counter"):
+        mock_time.counter = 0  # it doesn't exist yet, so initialize it
+    mock_time.counter += 1
+    return mock_time.counter
+
+
 class ConfigTest(TransactionTestCase):
 
     """Tests setting and updating config values"""
@@ -35,7 +57,7 @@ def setUp(self):
         self.user = User.objects.get(username='autotest')
         self.token = Token.objects.get(user=self.user).key
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_config(self):
         """
         Test that config is auto-created for a new app and that
@@ -107,7 +129,7 @@ def test_config(self):
         self.assertEqual(response.status_code, 405)
         return config5
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_response_data(self):
         """Test that the serialized response contains only relevant data."""
         body = {'id': 'test'}
@@ -132,7 +154,7 @@ def test_response_data(self):
         }
         self.assertDictContainsSubset(expected, response.data)
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_config_set_same_key(self):
         """
         Test that config sets on the same key function properly
@@ -156,7 +178,7 @@ def test_config_set_same_key(self):
         self.assertIn('PORT', response.data['values'])
         self.assertEqual(response.data['values']['PORT'], '5001')
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_config_set_unicode(self):
         """
         Test that config sets with unicode values are accepted.
@@ -187,14 +209,14 @@ def test_config_set_unicode(self):
         self.assertIn('INTEGER', response.data['values'])
         self.assertEqual(response.data['values']['INTEGER'], 1)
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_config_str(self):
         """Test the text representation of a node."""
         config5 = self.test_config()
         config = Config.objects.get(uuid=config5['uuid'])
         self.assertEqual(str(config), "{}-{}".format(config5['app'], config5['uuid'][:7]))
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_admin_can_create_config_on_other_apps(self):
         """If a non-admin creates an app, an administrator should be able to set config
         values for that app.
@@ -214,7 +236,7 @@ def test_admin_can_create_config_on_other_apps(self):
         self.assertIn('PORT', response.data['values'])
         return response
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_limit_memory(self):
         """
         Test that limit is auto-created for a new app and that
@@ -302,7 +324,7 @@ def test_limit_memory(self):
         self.assertEqual(response.status_code, 405)
         return limit4
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_limit_cpu(self):
         """
         Test that CPU limits can be set
@@ -373,7 +395,7 @@ def test_limit_cpu(self):
         self.assertEqual(response.status_code, 405)
         return limit4
 
-    @mock.patch('requests.post', mock_import_repository_task)
+    @mock.patch('requests.post', mock_status_ok)
     def test_tags(self):
         """
         Test that tags can be set on an application
@@ -477,3 +499,73 @@ def test_unauthorized_user_cannot_modify_config(self):
         response = self.client.post(url, json.dumps(body), content_type='application/json',
                                     HTTP_AUTHORIZATION='token {}'.format(unauthorized_token))
         self.assertEqual(response.status_code, 403)
+
+    def _test_app_healthcheck(self):
+        url = '/v1/apps'
+        response = self.client.post(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 201)
+        app_id = response.data['id']
+        # post a new build, expecting it to pass as usual
+        url = "/v1/apps/{app_id}/builds".format(**locals())
+        body = {'image': 'autotest/example'}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 201)
+        # set an initial healthcheck url.
+        url = "/v1/apps/{app_id}/config".format(**locals())
+        body = {'values': json.dumps({'HEALTHCHECK_URL': '/'})}
+        return self.client.post(url, json.dumps(body), content_type='application/json',
+                                HTTP_AUTHORIZATION='token {}'.format(self.token))
+
+    @mock.patch('requests.get', mock_status_not_found)
+    def test_app_healthcheck_good(self):
+        """
+        If a user deploys an app with a config value set for HEALTHCHECK_URL, the controller
+        should check that the application is up. If it's down, the app should be rolled back.
+        """
+        response = self._test_app_healthcheck()
+        self.assertEqual(response.status_code, 503)
+        self.assertEqual(
+            response.data,
+            {
+                'detail':
+                "aborting, app failed health check (got '404', expected: '200')"
+            })
+        # add in the expected app healthcheck status, which will result in a successful deployment
+        app_id = App.objects.all()[0]
+        url = "/v1/apps/{app_id}/config".format(**locals())
+        body = {'values': json.dumps({'HEALTHCHECK_STATUS_CODE': '404'})}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 201)
+
+    @mock.patch('requests.get', mock_request_timed_out)
+    @mock.patch('time.time', mock_time)
+    def test_app_healthcheck_timeout(self):
+        """
+        If a user deploys an app with a config value set for HEALTHCHECK_URL but the app
+        times out, the controller should continue checking until either the app
+        responds or the app fails to respond within the timeout.
+        """
+        response = self._test_app_healthcheck()
+        self.assertEqual(response.status_code, 503)
+        self.assertEqual(
+            response.data,
+            {'detail': 'app failed to respond to health check within 60 seconds of launch'})
+
+    @mock.patch('requests.get', mock_request_connection_error)
+    @mock.patch('time.time', mock_time)
+    def test_app_healthcheck_connection_error(self):
+        """
+        If a user deploys an app with a config value set for HEALTHCHECK_URL but the app
+        returns a connection error, the controller should continue checking until either the app
+        responds or the app fails to respond within the timeout.
+
+        NOTE (bacongobbler): the Docker userland proxy listens for connections and returns a
+        ConnectionError, hence the unit test.
+        """
+        response = self._test_app_healthcheck()
+        self.assertEqual(response.status_code, 503)
+        self.assertEqual(
+            response.data,
+            {'detail': 'app failed to respond to health check within 60 seconds of launch'})

controller/deis/settings.py

@@ -290,6 +290,9 @@
 # names which apps cannot reserve for routing
 DEIS_RESERVED_NAMES = ['deis']
 
+# the default expected status code for healthchecks
+DEIS_HEALTHCHECK_STATUS_CODE = '200'
+
 # default scheduler settings
 SCHEDULER_MODULE = 'scheduler.mock'
 SCHEDULER_TARGET = ''  # path to scheduler endpoint (e.g. /var/run/fleet.sock)

docs/managing_deis/backing_up_data.rst

@@ -212,6 +212,7 @@ use in the ``export`` command should correspond to the IP of the host machine wh
     [a.save() for a in App.objects.all()]
     [d.save() for d in Domain.objects.all()]
     [c.save() for c in Certificate.objects.all()]
+    [c.save() for c in Config.objects.all()]
     EOF
     $ exit
 

docs/using_deis/config-application.rst

@@ -91,6 +91,27 @@ appname to the old one:
     an application, however you can work around this by pointing the @ record to the
     address of the load balancer (if any).
 
+Custom Health Checks
+--------------------
+
+By default, Deis checks if the container is running for health checks. You can change both
+the healthcheck URL and the expected status code by setting custom config:
+
+.. code-block:: console
+
+    $ deis config:set HEALTHCHECK_URL=/404.html
+    === peachy-waxworks
+    HEALTHCHECK_URL: /404.html
+    $ deis config:set HEALTHCHECK_STATUS_CODE=404
+    === peachy-waxworks
+    HEALTHCHECK_STATUS_CODE: 404
+    HEALTHCHECK_URL: /404.html
+
+If a new release does not pass the healthcheck, the application will be rolled back to the previous
+release. Beyond that, if an application container responds to a heartbeat check with a different
+status than is expected, the :ref:`router` will mark that container as down and stop sending
+requests to that container.
+
 Track Changes
 -------------
 Each time a build or config change is made to your application, a new :ref:`release` is created.

publisher/server/publisher.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"net/http"
 	"regexp"
 	"strconv"
 	"sync"
@@ -115,6 +116,19 @@ func (s *Server) publishContainer(container *docker.APIContainers, ttl time.Dura
 			port := strconv.Itoa(int(p.PublicPort))
 			hostAndPort := s.host + ":" + port
 			if s.IsPublishableApp(containerName) && s.IsPortOpen(hostAndPort) {
+				configKey := fmt.Sprintf("/deis/services/%s/config/", appName)
+				// check if the user specified an upcheck URL
+				healthcheckURL := s.getEtcd(configKey + "healthcheck_url")
+				var healthcheckStatusCode int
+				healthcheckStatusCode, err := strconv.Atoi(s.getEtcd(configKey + "healthcheck_status_code"))
+				if err != nil {
+					healthcheckStatusCode = 200
+				}
+				if healthcheckURL != "" {
+					if !s.HealthCheckOK("http://"+hostAndPort+healthcheckURL, healthcheckStatusCode) {
+						continue
+					}
+				}
 				s.setEtcd(keyPath, hostAndPort, uint64(ttl.Seconds()))
 				safeMap.Lock()
 				safeMap.data[container.ID] = appPath
@@ -169,6 +183,19 @@ func (s *Server) IsPortOpen(hostAndPort string) bool {
 	return portOpen
 }
 
+func (s *Server) HealthCheckOK(url string, expectedStatusCode int) bool {
+	resp, err := http.Get(url)
+	if err != nil {
+		log.Printf("healthcheck failed for %s (expected %d, got %v)\n", url, expectedStatusCode, err)
+		return false
+	}
+	if resp.StatusCode == expectedStatusCode {
+		return true
+	}
+	log.Printf("healthcheck failed for %s (expected %d, got %d)\n", url, expectedStatusCode, resp.StatusCode)
+	return false
+}
+
 // latestRunningVersion retrieves the highest version of the application published
 // to etcd. If no app has been published, returns 0.
 func latestRunningVersion(client *etcd.Client, appName string) int {
@@ -213,6 +240,21 @@ func max(n []int) int {
 	return val
 }
 
+// getEtcd retrieves the etcd key's value. Returns an empty string if the key was not found.
+func (s *Server) getEtcd(key string) string {
+	if s.logLevel == "debug" {
+		log.Println("get", key)
+	}
+	resp, err := s.EtcdClient.Get(key, false, false)
+	if err != nil {
+		return ""
+	}
+	if resp != nil && resp.Node != nil {
+		return resp.Node.Value
+	}
+	return ""
+}
+
 // setEtcd sets the corresponding etcd key with the value and ttl
 func (s *Server) setEtcd(key, value string, ttl uint64) {
 	if _, err := s.EtcdClient.Set(key, value, ttl); err != nil {