feat(publisher): add http liveness health check

Author: Matthew

controller/api/models.py

@@ -1159,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']:
@@ -1204,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')

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
 

publisher/server/publisher.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"net/http"
 	"regexp"
 	"strconv"
 	"sync"
@@ -115,6 +116,24 @@ 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/config/%s/", appName)
+				// check if the user specified a healthcheck URL
+				healthcheckURL := s.getEtcd(configKey + "healthcheck_url")
+				delay, err := strconv.Atoi(s.getEtcd(configKey + "healthcheck_initial_delay"))
+				if err != nil {
+					log.Println(err)
+					delay = 0
+				}
+				timeout, err := strconv.Atoi(s.getEtcd(configKey + "healthcheck_timeout"))
+				if err != nil {
+					log.Println(err)
+					timeout = 1
+				}
+				if healthcheckURL != "" {
+					if !s.HealthCheckOK("http://"+hostAndPort+healthcheckURL, delay, timeout) {
+						continue
+					}
+				}
 				s.setEtcd(keyPath, hostAndPort, uint64(ttl.Seconds()))
 				safeMap.Lock()
 				safeMap.data[container.ID] = appPath
@@ -169,6 +188,23 @@ func (s *Server) IsPortOpen(hostAndPort string) bool {
 	return portOpen
 }
 
+func (s *Server) HealthCheckOK(url string, delay, timeout int) bool {
+	// sleep for the initial delay
+	time.Sleep(time.Duration(delay) * time.Second)
+	client := http.Client{
+		Timeout: time.Duration(timeout) * time.Second,
+	}
+	resp, err := client.Get(url)
+	if err != nil {
+		log.Printf("an error occurred while performing a health check at %s (%v)\n", url, err)
+		return false
+	}
+	if resp.StatusCode != http.StatusOK {
+		log.Printf("healthcheck failed for %s (expected %d, got %d)\n", url, http.StatusOK, resp.StatusCode)
+	}
+	return resp.StatusCode == http.StatusOK
+}
+
 // 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 +249,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 {

publisher/server/publisher_test.go

@@ -1,7 +1,10 @@
 package server
 
 import (
+	"fmt"
 	"net"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 )
 
@@ -45,3 +48,25 @@ func TestIsPortOpen(t *testing.T) {
 		t.Errorf("Port should be closed")
 	}
 }
+
+func TestHealthCheckOK(t *testing.T) {
+	s := &Server{}
+
+	// good server
+	ts1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintln(w, "Hello, client")
+	}))
+	defer ts1.Close()
+	if !s.HealthCheckOK(ts1.URL, 0, 0) {
+		t.Errorf("healthcheck should be OK")
+	}
+
+	// bad server
+	ts2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.NotFound(w, r)
+	}))
+	defer ts2.Close()
+	if s.HealthCheckOK(ts2.URL, 0, 0) {
+		t.Errorf("healthcheck should be NOT OK")
+	}
+}

tests/Makefile

@@ -23,6 +23,7 @@ test-full: test-style
 	godep go test -tags integration -v -run TestBuilds
 	godep go test -tags integration -v -run TestConfig
 	godep go test -tags integration -v -run TestDomains
+	godep go test -tags integration -v -run TestHealthcheck
 	godep go test -tags integration -v -run TestKeys
 	godep go test -tags integration -v -run TestPerms
 	godep go test -tags integration -v -run TestPs

tests/healthcheck_test.go

@@ -0,0 +1,74 @@
+// +build integration
+
+package tests
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/deis/deis/tests/utils"
+)
+
+var (
+	healthcheckGoodCmd = "config:set HEALTHCHECK_URL=/ --app={{.AppName}}"
+)
+
+func TestHealthcheck(t *testing.T) {
+	client := utils.HTTPClient()
+	cfg := healthcheckSetup(t)
+	done := make(chan bool, 1)
+	url := fmt.Sprintf("http://%s.%s", cfg.AppName, cfg.Domain)
+
+	utils.Execute(t, healthcheckGoodCmd, cfg, false, "/")
+	go func() {
+		// there should never be any downtime during these health check operations
+		psScaleTest(t, cfg, psScaleCmd)
+		cfg.ProcessNum = "1"
+		psScaleTest(t, cfg, psScaleCmd)
+		// kill healthcheck goroutine
+		done <- true
+	}()
+
+	// run health checks in parallel while performing operations
+	fmt.Printf("starting health checks at %s\n", url)
+loop:
+	for {
+		select {
+		case <-done:
+			fmt.Println("done performing health checks")
+			break loop
+		default:
+			doHealthCheck(t, client, url)
+		}
+	}
+	utils.AppsDestroyTest(t, cfg)
+}
+
+func healthcheckSetup(t *testing.T) *utils.DeisTestConfig {
+	cfg := utils.GetGlobalConfig()
+	cfg.AppName = "healthchecksample"
+	utils.Execute(t, authLoginCmd, cfg, false, "")
+	utils.Execute(t, gitCloneCmd, cfg, false, "")
+	if err := utils.Chdir(cfg.ExampleApp); err != nil {
+		t.Fatal(err)
+	}
+	utils.Execute(t, appsCreateCmd, cfg, false, "")
+	utils.Execute(t, gitPushCmd, cfg, false, "")
+	utils.CurlApp(t, *cfg)
+	if err := utils.Chdir(".."); err != nil {
+		t.Fatal(err)
+	}
+	return cfg
+}
+
+func doHealthCheck(t *testing.T, client *http.Client, url string) {
+	response, err := client.Get(url)
+	if err != nil {
+		t.Fatalf("could not retrieve response from %s: %v\n", url, err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("app had some downtime while undergoing health checks (got %d response)", response.StatusCode)
+	}
+}

tests/utils/itutils.go

@@ -123,12 +123,17 @@ func GetGlobalConfig() *DeisTestConfig {
 	return &envCfg
 }
 
-func doCurl(url string) ([]byte, error) {
+// HTTPClient returns a client for use with the integration tests.
+func HTTPClient() *http.Client {
 	// disable security check for self-signed certificates
 	tr := &http.Transport{
 		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 	}
-	client := &http.Client{Transport: tr}
+	return &http.Client{Transport: tr}
+}
+
+func doCurl(url string) ([]byte, error) {
+	client := HTTPClient()
 	response, err := client.Get(url)
 	if err != nil {
 		return nil, err
@@ -300,6 +305,7 @@ func Execute(t *testing.T, cmd string, params interface{}, failFlag bool, expect
 
 // AppsDestroyTest destroys a Deis app and checks that it was successful.
 func AppsDestroyTest(t *testing.T, params *DeisTestConfig) {
+	fmt.Printf("destroying app %s...\n", params.ExampleApp)
 	cmd := "apps:destroy --app={{.AppName}} --confirm={{.AppName}}"
 	if err := Chdir(params.ExampleApp); err != nil {
 		t.Fatal(err)