agent: add per-endpoint proxy metrics (#220)

Fixes #216, where if the agent runs multiple endpoints, it attempts to
register the same metrics multiple times so panics.

This extends the proxy metrics to include an endpoint label, such as:
```
piko_agent_requests_total{endpoint="endpoint1",method="GET",status="200"} 1034
piko_agent_requests_total{endpoint="endpoint2",method="GET",status="200"} 1234
```

It also registers the metrics only once to avoid panicing.

agent/reverseproxy/metrics.go

@@ -0,0 +1,137 @@
+package reverseproxy
+
+import (
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type Metrics struct {
+	RequestsInFlight *prometheus.GaugeVec
+	RequestsTotal    *prometheus.CounterVec
+	RequestLatency   *prometheus.HistogramVec
+	RequestSize      *prometheus.HistogramVec
+	ResponseSize     *prometheus.HistogramVec
+}
+
+func NewMetrics(subsystem string) *Metrics {
+	sizeBuckets := prometheus.ExponentialBuckets(256, 4, 8)
+	return &Metrics{
+		RequestsInFlight: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Namespace: "piko",
+				Subsystem: subsystem,
+				Name:      "requests_in_flight",
+				Help:      "Number of requests currently handled by this server.",
+			},
+			[]string{"endpoint"},
+		),
+		RequestsTotal: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: "piko",
+				Subsystem: subsystem,
+				Name:      "requests_total",
+				Help:      "Total requests.",
+			},
+			[]string{"status", "method", "endpoint"},
+		),
+		RequestLatency: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: "piko",
+				Subsystem: subsystem,
+				Name:      "request_latency_seconds",
+				Help:      "Request latency.",
+				Buckets:   prometheus.DefBuckets,
+			},
+			[]string{"status", "method", "endpoint"},
+		),
+		RequestSize: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: "piko",
+				Subsystem: subsystem,
+				Name:      "request_size_bytes",
+				Help:      "Request size",
+				Buckets:   sizeBuckets,
+			},
+			[]string{"endpoint"},
+		),
+		ResponseSize: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: "piko",
+				Subsystem: subsystem,
+				Name:      "response_size_bytes",
+				Help:      "Response size",
+				Buckets:   sizeBuckets,
+			},
+			[]string{"endpoint"},
+		),
+	}
+}
+
+func (m *Metrics) Handler(endpoint string) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		m.RequestsInFlight.With(prometheus.Labels{
+			"endpoint": endpoint,
+		}).Inc()
+		defer m.RequestsInFlight.With(prometheus.Labels{
+			"endpoint": endpoint,
+		}).Dec()
+
+		start := time.Now()
+
+		// Process request.
+		c.Next()
+
+		m.RequestsTotal.With(prometheus.Labels{
+			"status":   strconv.Itoa(c.Writer.Status()),
+			"method":   c.Request.Method,
+			"endpoint": endpoint,
+		}).Inc()
+		m.RequestLatency.With(prometheus.Labels{
+			"status":   strconv.Itoa(c.Writer.Status()),
+			"method":   c.Request.Method,
+			"endpoint": endpoint,
+		}).Observe(float64(time.Since(start).Milliseconds()) / 1000)
+		m.RequestSize.With(prometheus.Labels{
+			"endpoint": endpoint,
+		}).Observe(float64(computeApproximateRequestSize(c.Request)))
+		m.ResponseSize.With(prometheus.Labels{
+			"endpoint": endpoint,
+		}).Observe(float64(c.Writer.Size()))
+	}
+}
+
+func (m *Metrics) Register(registry *prometheus.Registry) {
+	registry.MustRegister(
+		m.RequestsInFlight,
+		m.RequestsTotal,
+		m.RequestLatency,
+		m.RequestSize,
+		m.ResponseSize,
+	)
+}
+
+func computeApproximateRequestSize(r *http.Request) int {
+	s := 0
+	if r.URL != nil {
+		s += len(r.URL.String())
+	}
+
+	s += len(r.Method)
+	s += len(r.Proto)
+	for name, values := range r.Header {
+		s += len(name)
+		for _, value := range values {
+			s += len(value)
+		}
+	}
+	s += len(r.Host)
+
+	if r.ContentLength != -1 {
+		s += int(r.ContentLength)
+	}
+	return s
+}

agent/reverseproxy/server.go

@@ -7,7 +7,6 @@ import (
 	"net/http"
 
 	"github.com/gin-gonic/gin"
-	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
@@ -28,7 +27,7 @@ type Server struct {
 
 func NewServer(
 	conf config.ListenerConfig,
-	registry *prometheus.Registry,
+	metrics *Metrics,
 	logger log.Logger,
 ) *Server {
 	logger = logger.WithSubsystem("proxy.http")
@@ -50,11 +49,7 @@ func NewServer(
 
 	s.router.Use(middleware.NewLogger(conf.AccessLog, logger))
 
-	metrics := middleware.NewMetrics("agent")
-	if registry != nil {
-		metrics.Register(registry)
-	}
-	router.Use(metrics.Handler())
+	router.Use(metrics.Handler(conf.EndpointID))
 
 	s.router.NoRoute(s.proxyRoute)
 

cli/agent/command.go

@@ -121,6 +121,8 @@ func runAgent(conf *config.Config, logger log.Logger) error {
 	}
 
 	registry := prometheus.NewRegistry()
+	proxyMetrics := reverseproxy.NewMetrics("agent")
+	proxyMetrics.Register(registry)
 
 	var group rungroup.Group
 
@@ -138,7 +140,7 @@ func runAgent(conf *config.Config, logger log.Logger) error {
 		defer ln.Close()
 
 		if listenerConfig.Protocol == config.ListenerProtocolHTTP {
-			server := reverseproxy.NewServer(listenerConfig, registry, logger)
+			server := reverseproxy.NewServer(listenerConfig, proxyMetrics, logger)
 
 			// Listener handler.
 			group.Add(func() error {