MEDIUM: Add /health and /ready endpoint to stats server (#69)

In some setup, we'd like to wait until connect is ready before running
the app. This PR adds a /ready endpoint that blocks until haproxy is
setup. A /health endpoint is also added to monitor that consul connect
is running.

Author: ShimmerGlass

haproxy/config.go

@@ -14,13 +14,7 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-const (
-	dataplaneUser = "haproxy"
-)
-
-var dataplanePass string
-
-var baseCfgTmpl = `
+const baseCfgTmpl = `
 global
 	master-worker
     stats socket {{.SocketPath}} mode 600 level admin expose-fd listeners
@@ -71,6 +65,8 @@ type haConfig struct {
 	StatsSock               string
 	DataplaneSock           string
 	DataplaneTransactionDir string
+	DataplaneUser           string
+	DataplanePass           string
 	LogsSock                string
 }
 
@@ -116,14 +112,15 @@ func newHaConfig(baseDir string, sd *lib.Shutdown) (*haConfig, error) {
 		}
 	}()
 
-	dataplanePass = createRandomString()
+	cfg.DataplanePass = createRandomString()
+	cfg.DataplaneUser = "hapeoxy"
 
 	err = tmpl.Execute(cfgFile, baseParams{
 		NbThread:      runtime.GOMAXPROCS(0),
 		SocketPath:    cfg.StatsSock,
 		LogsPath:      cfg.LogsSock,
-		DataplaneUser: dataplaneUser,
-		DataplanePass: dataplanePass,
+		DataplaneUser: cfg.DataplaneUser,
+		DataplanePass: cfg.DataplanePass,
 	})
 	if err != nil {
 		sd.Done()

haproxy/haproxy.go

@@ -3,18 +3,15 @@ package haproxy
 import (
 	"fmt"
 	"net"
-	"net/http"
-	"strconv"
-	"time"
 
 	spoe "github.com/criteo/haproxy-spoe-go"
 	"github.com/haproxytech/haproxy-consul-connect/consul"
 	"github.com/haproxytech/haproxy-consul-connect/haproxy/dataplane"
 	"github.com/haproxytech/haproxy-consul-connect/haproxy/haproxy_cmd"
 	"github.com/haproxytech/haproxy-consul-connect/haproxy/state"
+	"github.com/haproxytech/haproxy-consul-connect/haproxy/stats"
 	"github.com/haproxytech/haproxy-consul-connect/lib"
 	"github.com/hashicorp/consul/api"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	log "github.com/sirupsen/logrus"
 	"gopkg.in/mcuadros/go-syslog.v2"
 )
@@ -50,16 +47,16 @@ func New(consulClient *api.Client, cfg chan consul.Config, opts Options) *HAProx
 }
 
 func (h *HAProxy) Run(sd *lib.Shutdown) error {
-	return h.watch(sd)
-}
-
-func (h *HAProxy) start(sd *lib.Shutdown) error {
 	hc, err := newHaConfig(h.opts.ConfigBaseDir, sd)
 	if err != nil {
 		return err
 	}
 	h.haConfig = hc
 
+	return h.watch(sd)
+}
+
+func (h *HAProxy) start(sd *lib.Shutdown) error {
 	if h.opts.LogRequests {
 		err := h.startLogger()
 		if err != nil {
@@ -74,19 +71,19 @@ func (h *HAProxy) start(sd *lib.Shutdown) error {
 		}
 	}
 
-	dpc, err := haproxy_cmd.Start(sd, haproxy_cmd.Config{
+	var err error
+	h.dataplaneClient, err = haproxy_cmd.Start(sd, haproxy_cmd.Config{
 		HAProxyPath:             h.opts.HAProxyBin,
-		HAProxyConfigPath:       hc.HAProxy,
+		HAProxyConfigPath:       h.haConfig.HAProxy,
 		DataplanePath:           h.opts.DataplaneBin,
-		DataplaneTransactionDir: hc.DataplaneTransactionDir,
-		DataplaneSock:           hc.DataplaneSock,
-		DataplaneUser:           dataplaneUser,
-		DataplanePass:           dataplanePass,
+		DataplaneTransactionDir: h.haConfig.DataplaneTransactionDir,
+		DataplaneSock:           h.haConfig.DataplaneSock,
+		DataplaneUser:           h.haConfig.DataplaneUser,
+		DataplanePass:           h.haConfig.DataplanePass,
 	})
 	if err != nil {
 		return err
 	}
-	h.dataplaneClient = dpc
 
 	err = h.startStats()
 	if err != nil {
@@ -145,53 +142,22 @@ func (h *HAProxy) startStats() error {
 	if h.opts.StatsListenAddr == "" {
 		return nil
 	}
-	go func() {
-		if !h.opts.StatsRegisterService {
-			return
-		}
-
-		_, portStr, err := net.SplitHostPort(h.opts.StatsListenAddr)
-		if err != nil {
-			log.Errorf("cannot parse stats listen addr: %s", err)
-		}
-		port, _ := strconv.Atoi(portStr)
-
-		reg := func() {
-			err = h.consulClient.Agent().ServiceRegister(&api.AgentServiceRegistration{
-				ID:   fmt.Sprintf("%s-connect-stats", h.currentConsulConfig.ServiceID),
-				Name: fmt.Sprintf("%s-connect-stats", h.currentConsulConfig.ServiceName),
-				Port: port,
-				Checks: api.AgentServiceChecks{
-					&api.AgentServiceCheck{
-						HTTP:                           fmt.Sprintf("http://localhost:%d/metrics", port),
-						Interval:                       (10 * time.Second).String(),
-						DeregisterCriticalServiceAfter: time.Minute.String(),
-					},
-				},
-				Tags: []string{"connect-stats"},
-			})
-			if err != nil {
-				log.Errorf("cannot register stats service: %s", err)
-			}
-		}
 
-		reg()
+	s := stats.New(
+		h.consulClient,
+		h.dataplaneClient,
+		h.Ready,
+		stats.Config{
+			RegisterService: h.opts.StatsRegisterService,
+			ListenAddr:      h.opts.StatsListenAddr,
+			ServiceName:     h.currentConsulConfig.ServiceName,
+			ServiceID:       h.currentConsulConfig.ServiceID,
+		})
 
-		for range time.Tick(time.Minute) {
-			reg()
-		}
-	}()
-	go (&Stats{
-		dpapi:   h.dataplaneClient,
-		service: h.currentConsulConfig.ServiceName,
-	}).Run()
 	go func() {
-		http.Handle("/metrics", promhttp.Handler())
-
-		log.Infof("Starting stats server at %s", h.opts.StatsListenAddr)
-		err := http.ListenAndServe(h.opts.StatsListenAddr, nil)
+		err := s.Run()
 		if err != nil {
-			log.Errorf("error starting stats server: %s", err)
+			log.Error(err)
 		}
 	}()
 

haproxy/state.go

@@ -25,6 +25,7 @@ func (h *HAProxy) watch(sd *lib.Shutdown) error {
 	var currentConfig consul.Config
 	dirty := false
 	started := false
+	ready := false
 
 	waitAndRetry := func() {
 		time.Sleep(retryBackoff)
@@ -69,7 +70,6 @@ func (h *HAProxy) watch(sd *lib.Shutdown) error {
 				return err
 			}
 			started = true
-			close(h.Ready)
 		}
 
 		if dirty {
@@ -122,6 +122,11 @@ func (h *HAProxy) watch(sd *lib.Shutdown) error {
 			continue
 		}
 
+		if !ready {
+			close(h.Ready)
+			ready = true
+		}
+
 		currentState = newState
 		log.Info("state applied")
 	}

haproxy/stats.go renamed to haproxy/stats/metrics.go

@@ -1,10 +1,9 @@
-package haproxy
+package stats
 
 import (
 	"strings"
 	"time"
 
-	"github.com/haproxytech/haproxy-consul-connect/haproxy/dataplane"
 	"github.com/haproxytech/models"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -70,13 +69,8 @@ var (
 	}, []string{"service"})
 )
 
-type Stats struct {
-	service string
-	dpapi   *dataplane.Dataplane
-}
-
-func (s *Stats) Run() {
-	upMetric.WithLabelValues(s.service).Set(1)
+func (s *Stats) runMetrics() {
+	upMetric.WithLabelValues(s.cfg.ServiceName).Set(1)
 	for {
 		time.Sleep(time.Second)
 		stats, err := s.dpapi.Stats()
@@ -114,42 +108,42 @@ func (s *Stats) handleFrontend(stats *models.NativeStat) {
 	targetService := strings.TrimPrefix(stats.Name, "front_")
 
 	if targetService == "downstream" {
-		reqInRate.WithLabelValues(s.service).Set(statVal(stats.Stats.ReqRate))
-		connInCount.WithLabelValues(s.service).Set(statVal(stats.Stats.Scur))
-		bytesInIn.WithLabelValues(s.service).Set(statVal(stats.Stats.Bin))
-		bytesOutIn.WithLabelValues(s.service).Set(statVal(stats.Stats.Bout))
-
-		resInTotal.WithLabelValues(s.service, "1xx").Set(statVal(stats.Stats.Hrsp1xx))
-		resInTotal.WithLabelValues(s.service, "2xx").Set(statVal(stats.Stats.Hrsp2xx))
-		resInTotal.WithLabelValues(s.service, "3xx").Set(statVal(stats.Stats.Hrsp3xx))
-		resInTotal.WithLabelValues(s.service, "4xx").Set(statVal(stats.Stats.Hrsp4xx))
-		resInTotal.WithLabelValues(s.service, "5xx").Set(statVal(stats.Stats.Hrsp5xx))
-		resInTotal.WithLabelValues(s.service, "other").Set(statVal(stats.Stats.HrspOther))
+		reqInRate.WithLabelValues(s.cfg.ServiceName).Set(statVal(stats.Stats.ReqRate))
+		connInCount.WithLabelValues(s.cfg.ServiceName).Set(statVal(stats.Stats.Scur))
+		bytesInIn.WithLabelValues(s.cfg.ServiceName).Set(statVal(stats.Stats.Bin))
+		bytesOutIn.WithLabelValues(s.cfg.ServiceName).Set(statVal(stats.Stats.Bout))
+
+		resInTotal.WithLabelValues(s.cfg.ServiceName, "1xx").Set(statVal(stats.Stats.Hrsp1xx))
+		resInTotal.WithLabelValues(s.cfg.ServiceName, "2xx").Set(statVal(stats.Stats.Hrsp2xx))
+		resInTotal.WithLabelValues(s.cfg.ServiceName, "3xx").Set(statVal(stats.Stats.Hrsp3xx))
+		resInTotal.WithLabelValues(s.cfg.ServiceName, "4xx").Set(statVal(stats.Stats.Hrsp4xx))
+		resInTotal.WithLabelValues(s.cfg.ServiceName, "5xx").Set(statVal(stats.Stats.Hrsp5xx))
+		resInTotal.WithLabelValues(s.cfg.ServiceName, "other").Set(statVal(stats.Stats.HrspOther))
 	} else {
-		reqOutRate.WithLabelValues(s.service, targetService).Set(statVal(stats.Stats.ReqRate))
-		connOutCount.WithLabelValues(s.service, targetService).Set(statVal(stats.Stats.Scur))
-		bytesInOut.WithLabelValues(s.service, targetService).Set(statVal(stats.Stats.Bin))
-		bytesOutOut.WithLabelValues(s.service, targetService).Set(statVal(stats.Stats.Bout))
-
-		resOutTotal.WithLabelValues(s.service, targetService, "1xx").Set(statVal(stats.Stats.Hrsp1xx))
-		resOutTotal.WithLabelValues(s.service, targetService, "2xx").Set(statVal(stats.Stats.Hrsp2xx))
-		resOutTotal.WithLabelValues(s.service, targetService, "3xx").Set(statVal(stats.Stats.Hrsp3xx))
-		resOutTotal.WithLabelValues(s.service, targetService, "4xx").Set(statVal(stats.Stats.Hrsp4xx))
-		resOutTotal.WithLabelValues(s.service, targetService, "5xx").Set(statVal(stats.Stats.Hrsp5xx))
-		resOutTotal.WithLabelValues(s.service, targetService, "other").Set(statVal(stats.Stats.HrspOther))
+		reqOutRate.WithLabelValues(s.cfg.ServiceName, targetService).Set(statVal(stats.Stats.ReqRate))
+		connOutCount.WithLabelValues(s.cfg.ServiceName, targetService).Set(statVal(stats.Stats.Scur))
+		bytesInOut.WithLabelValues(s.cfg.ServiceName, targetService).Set(statVal(stats.Stats.Bin))
+		bytesOutOut.WithLabelValues(s.cfg.ServiceName, targetService).Set(statVal(stats.Stats.Bout))
+
+		resOutTotal.WithLabelValues(s.cfg.ServiceName, targetService, "1xx").Set(statVal(stats.Stats.Hrsp1xx))
+		resOutTotal.WithLabelValues(s.cfg.ServiceName, targetService, "2xx").Set(statVal(stats.Stats.Hrsp2xx))
+		resOutTotal.WithLabelValues(s.cfg.ServiceName, targetService, "3xx").Set(statVal(stats.Stats.Hrsp3xx))
+		resOutTotal.WithLabelValues(s.cfg.ServiceName, targetService, "4xx").Set(statVal(stats.Stats.Hrsp4xx))
+		resOutTotal.WithLabelValues(s.cfg.ServiceName, targetService, "5xx").Set(statVal(stats.Stats.Hrsp5xx))
+		resOutTotal.WithLabelValues(s.cfg.ServiceName, targetService, "other").Set(statVal(stats.Stats.HrspOther))
 	}
 }
 
 func (s *Stats) handlebackend(stats *models.NativeStat) {
 	targetService := strings.TrimPrefix(stats.Name, "back_")
 
 	if targetService == "downstream" {
-		resTimeIn.WithLabelValues(s.service).Set(statVal(stats.Stats.Ttime) / 1000)
+		resTimeIn.WithLabelValues(s.cfg.ServiceName).Set(statVal(stats.Stats.Ttime) / 1000)
 	} else {
-		resTimeOut.WithLabelValues(s.service, targetService).Set(statVal(stats.Stats.Ttime) / 1000)
+		resTimeOut.WithLabelValues(s.cfg.ServiceName, targetService).Set(statVal(stats.Stats.Ttime) / 1000)
 	}
 }
 
 func (s *Stats) handleServer(stats *models.NativeStat) {
-	resTimeOut.WithLabelValues(s.service, stats.Name).Set(statVal(stats.Stats.Ttime) / 1000)
+	resTimeOut.WithLabelValues(s.cfg.ServiceName, stats.Name).Set(statVal(stats.Stats.Ttime) / 1000)
 }