Add stats server

consul/config.go

@@ -8,6 +8,7 @@ import (
 
 type Config struct {
 	ServiceName string
+	ServiceID   string
 	CAsPool     *x509.CertPool
 	Downstream  Downstream
 	Upstreams   []Upstream

consul/watcher.go

@@ -325,6 +325,7 @@ func (w *Watcher) genCfg() Config {
 
 	config := Config{
 		ServiceName: w.serviceName,
+		ServiceID:   w.service,
 		CAsPool:     w.certCAPool,
 		Downstream: Downstream{
 			LocalBindAddress: w.downstream.LocalBindAddress,

haproxy/haproxy.go

@@ -5,6 +5,8 @@ import (
 	"net"
 	"net/http"
 	"os/exec"
+	"strconv"
+	"sync"
 	"syscall"
 	"time"
 
@@ -13,6 +15,7 @@ import (
 	spoe "github.com/criteo/haproxy-spoe-go"
 	"github.com/haproxytech/models"
 	"github.com/hashicorp/consul/api"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	log "github.com/sirupsen/logrus"
 	"gopkg.in/mcuadros/go-syslog.v2"
 )
@@ -78,20 +81,25 @@ func (h *HAProxy) Run(sd *lib.Shutdown) error {
 		return err
 	}
 
-	//go (&Stats{h.dataplaneClient}).Run()
-
 	err = h.init()
 	if err != nil {
 		return err
 	}
 
+	var statsOnce sync.Once
 	for {
 		select {
 		case c := <-h.cfgC:
 			err := h.handleChange(c)
 			if err != nil {
 				log.Error(err)
 			}
+			statsOnce.Do(func() {
+				err := h.startStats()
+				if err != nil {
+					log.Error(err)
+				}
+			})
 		case <-sd.Stop:
 			return nil
 		}
@@ -260,3 +268,57 @@ func (h *HAProxy) startDataplane(sd *lib.Shutdown, haCmd *exec.Cmd) error {
 
 	return nil
 }
+
+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.currentCfg.ServiceID),
+				Name: fmt.Sprintf("%s-connect-stats", h.currentCfg.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()
+
+		for range time.Tick(time.Minute) {
+			reg()
+		}
+	}()
+	go (&Stats{
+		dpapi:   h.dataplaneClient,
+		service: h.currentCfg.ServiceName,
+	}).Run()
+	go func() {
+		http.Handle("/metrics", promhttp.Handler())
+
+		log.Infof("Starting stats server at %s", h.opts.StatsListenAddr)
+		http.ListenAndServe(h.opts.StatsListenAddr, nil)
+	}()
+
+	return nil
+}

haproxy/options.go

@@ -1,9 +1,11 @@
 package haproxy
 
 type Options struct {
-	HAProxyBin       string
-	DataplaneBin     string
-	ConfigBaseDir    string
-	SPOEAddress      string
-	EnableIntentions bool
+	HAProxyBin           string
+	DataplaneBin         string
+	ConfigBaseDir        string
+	SPOEAddress          string
+	EnableIntentions     bool
+	StatsListenAddr      string
+	StatsRegisterService bool
 }

haproxy/stats.go

@@ -1,27 +1,81 @@
 package haproxy
 
 import (
+	"strings"
 	"time"
 
-	"github.com/davecgh/go-spew/spew"
 	"github.com/haproxytech/models"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	log "github.com/sirupsen/logrus"
 )
 
 var (
-	opsProcessed = promauto.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "http_requests_total",
+	upMetric = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_up",
+		Help: "The total number of http requests",
+	}, []string{"service"})
+
+	reqOutRate = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_http_request_out_rate",
+		Help: "The total number of http requests",
+	}, []string{"service", "target"})
+	reqInRate = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_http_request_in_rate",
+		Help: "The total number of http requests",
+	}, []string{"service"})
+	resInTotal = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_http_response_in_total",
+		Help: "The total number of http requests",
+	}, []string{"service", "code"})
+	resOutTotal = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_http_response_out_total",
+		Help: "The total number of http requests",
+	}, []string{"service", "target", "code"})
+
+	resTimeIn = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_http_response_in_avg_time_second",
+		Help: "The total number of http requests",
+	}, []string{"service"})
+	resTimeOut = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_http_response_out_avg_time_second",
+		Help: "The total number of http requests",
+	}, []string{"service", "target"})
+
+	connOutCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_connection_out_rate",
+		Help: "The total number of http requests",
+	}, []string{"service", "target"})
+	connInCount = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_connection_in_count",
+		Help: "The total number of http requests",
+	}, []string{"service"})
+
+	bytesInOut = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_bytes_in_out_total",
+		Help: "The total number of http requests",
+	}, []string{"service", "target"})
+	bytesOutOut = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_bytes_out_out_total",
+		Help: "The total number of http requests",
+	}, []string{"service", "target"})
+	bytesInIn = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_bytes_in_in_total",
+		Help: "The total number of http requests",
+	}, []string{"service"})
+	bytesOutIn = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "haproxy_connect_bytes_out_in_total",
 		Help: "The total number of http requests",
 	}, []string{"service"})
 )
 
 type Stats struct {
-	dpapi *dataplaneClient
+	service string
+	dpapi   *dataplaneClient
 }
 
 func (s *Stats) Run() {
+	upMetric.WithLabelValues(s.service).Set(1)
 	for {
 		time.Sleep(time.Second)
 		stats, err := s.dpapi.Stats()
@@ -34,7 +88,65 @@ func (s *Stats) Run() {
 }
 
 func (s *Stats) handle(stats []models.NativeStat) {
-	for _, s := range stats {
-		spew.Dump(s)
+	for _, stats := range stats {
+		switch stats.Type {
+		case models.NativeStatTypeFrontend:
+			s.handleFrontend(stats)
+		case models.NativeStatTypeBackend:
+			s.handlebackend(stats)
+		case models.NativeStatTypeServer:
+			s.handleServer(stats)
+		}
+	}
+}
+
+func statVal(i *int64) float64 {
+	if i == nil {
+		return 0
+	}
+	return float64(*i)
+}
+
+func (s *Stats) handleFrontend(stats models.NativeStat) {
+	targetService := strings.TrimPrefix(stats.Name, "front_")
+
+	if targetService == "downstream" {
+		reqInRate.WithLabelValues(s.service).Set(statVal(stats.Stats.Rate))
+		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))
+	} else {
+		reqOutRate.WithLabelValues(s.service, targetService).Set(statVal(stats.Stats.Rate))
+		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))
 	}
 }
+
+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)
+	} else {
+		resTimeOut.WithLabelValues(s.service, 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)
+}

main.go

@@ -21,6 +21,8 @@ func main() {
 	haproxyBin := flag.String("haproxy", "haproxy", "Haproxy binary path")
 	dataplaneBin := flag.String("dataplane", "dataplane-api", "Dataplane binary path")
 	haproxyCfgBasePath := flag.String("haproxy-cfg-base-path", "/tmp", "Haproxy binary path")
+	statsListenAddr := flag.String("stats-addr", "", "Listen addr for stats server")
+	statsServiceRegister := flag.Bool("stats-service-register", false, "Register a consul service for connect stats")
 	enableIntentions := flag.Bool("enable-intentions", false, "Enable Connect intentions")
 	token := flag.String("token", "", "Consul ACL token")
 	flag.Parse()
@@ -45,10 +47,12 @@ func main() {
 	}()
 
 	hap := haproxy.New(consulClient, watcher.C, haproxy.Options{
-		HAProxyBin:       *haproxyBin,
-		DataplaneBin:     *dataplaneBin,
-		ConfigBaseDir:    *haproxyCfgBasePath,
-		EnableIntentions: *enableIntentions,
+		HAProxyBin:           *haproxyBin,
+		DataplaneBin:         *dataplaneBin,
+		ConfigBaseDir:        *haproxyCfgBasePath,
+		EnableIntentions:     *enableIntentions,
+		StatsListenAddr:      *statsListenAddr,
+		StatsRegisterService: *statsServiceRegister,
 	})
 	sd.Add(1)
 	go func() {