Merge pull request #396 from fnproject/add_prometheus_metrics

Add Prometheus statistics and an example to showcase them using Grafana

api/agent/agent.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
 	"sort"
 	"sync"
 	"time"
@@ -17,6 +18,7 @@ import (
 	"github.com/fnproject/fn/api/id"
 	"github.com/fnproject/fn/api/models"
 	"github.com/opentracing/opentracing-go"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/sirupsen/logrus"
 )
 
@@ -105,6 +107,9 @@ type Agent interface {
 	// Stats should be burned at the stake. adding so as to not ruffle feathers.
 	// TODO this should be derived from our metrics
 	Stats() Stats
+
+	// Return the http.Handler used to handle Prometheus metric requests
+	PromHandler() http.Handler
 }
 
 type agent struct {
@@ -131,6 +136,9 @@ type agent struct {
 	shutdown chan struct{}
 
 	stats // TODO kill me
+
+	// Prometheus HTTP handler
+	promHandler http.Handler
 }
 
 func New(ds models.Datastore, mq models.MessageQueue) Agent {
@@ -138,13 +146,14 @@ func New(ds models.Datastore, mq models.MessageQueue) Agent {
 	driver := docker.NewDocker(drivers.Config{})
 
 	a := &agent{
-		ds:       ds,
-		mq:       mq,
-		driver:   driver,
-		hot:      make(map[string]chan slot),
-		cond:     sync.NewCond(new(sync.Mutex)),
-		ramTotal: getAvailableMemory(),
-		shutdown: make(chan struct{}),
+		ds:          ds,
+		mq:          mq,
+		driver:      driver,
+		hot:         make(map[string]chan slot),
+		cond:        sync.NewCond(new(sync.Mutex)),
+		ramTotal:    getAvailableMemory(),
+		shutdown:    make(chan struct{}),
+		promHandler: promhttp.Handler(),
 	}
 
 	go a.asyncDequeue() // safe shutdown can nanny this fine

api/agent/prometheus_metrics.go

@@ -0,0 +1,9 @@
+package agent
+
+import (
+	"net/http"
+)
+
+func (a *agent) PromHandler() http.Handler {
+	return a.promHandler
+}

api/agent/stats.go

@@ -1,6 +1,9 @@
 package agent
 
-import "sync"
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"sync"
+)
 
 // TODO this should expose:
 // * hot containers active
@@ -44,6 +47,44 @@ type FunctionStats struct {
 	Failed   uint64
 }
 
+var (
+	fnQueued = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "fn_api_queued",
+			Help: "Queued requests by path",
+		},
+		[](string){"path"},
+	)
+	fnRunning = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "fn_api_running",
+			Help: "Running requests by path",
+		},
+		[](string){"path"},
+	)
+	fnCompleted = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "fn_api_completed",
+			Help: "Completed requests by path",
+		},
+		[](string){"path"},
+	)
+	fnFailed = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "fn_api_failed",
+			Help: "Failed requests by path",
+		},
+		[](string){"path"},
+	)
+)
+
+func init() {
+	prometheus.MustRegister(fnQueued)
+	prometheus.MustRegister(fnRunning)
+	prometheus.MustRegister(fnFailed)
+	prometheus.MustRegister(fnCompleted)
+}
+
 func (s *stats) getStatsForFunction(path string) *functionStats {
 	if s.functionStatsMap == nil {
 		s.functionStatsMap = make(map[string]*functionStats)
@@ -59,52 +100,78 @@ func (s *stats) getStatsForFunction(path string) *functionStats {
 
 func (s *stats) Enqueue(path string) {
 	s.mu.Lock()
+
 	s.queue++
 	s.getStatsForFunction(path).queue++
+	fnQueued.WithLabelValues(path).Inc()
+
 	s.mu.Unlock()
 }
 
 // Call when a function has been queued but cannot be started because of an error
 func (s *stats) Dequeue(path string) {
 	s.mu.Lock()
+
 	s.queue--
 	s.getStatsForFunction(path).queue--
+	fnQueued.WithLabelValues(path).Dec()
+
 	s.mu.Unlock()
 }
 
 func (s *stats) DequeueAndStart(path string) {
 	s.mu.Lock()
+
 	s.queue--
 	s.getStatsForFunction(path).queue--
+	fnQueued.WithLabelValues(path).Dec()
+
 	s.running++
 	s.getStatsForFunction(path).running++
+	fnRunning.WithLabelValues(path).Inc()
+
 	s.mu.Unlock()
 }
 
 func (s *stats) Complete(path string) {
 	s.mu.Lock()
+
 	s.running--
 	s.getStatsForFunction(path).running--
+	fnRunning.WithLabelValues(path).Dec()
+
 	s.complete++
 	s.getStatsForFunction(path).complete++
+	fnCompleted.WithLabelValues(path).Inc()
+
 	s.mu.Unlock()
 }
 
 func (s *stats) Failed(path string) {
 	s.mu.Lock()
+
 	s.running--
 	s.getStatsForFunction(path).running--
+	fnRunning.WithLabelValues(path).Dec()
+
 	s.failed++
 	s.getStatsForFunction(path).failed++
+	fnFailed.WithLabelValues(path).Inc()
+
 	s.mu.Unlock()
 }
 
 func (s *stats) DequeueAndFail(path string) {
 	s.mu.Lock()
+
 	s.queue--
 	s.getStatsForFunction(path).queue--
+	fnQueued.WithLabelValues(path).Dec()
+
 	s.failed++
 	s.getStatsForFunction(path).failed++
+	fnFailed.WithLabelValues(path).Inc()
+
 	s.mu.Unlock()
 }
 

api/server/prometheus_metrics.go

@@ -0,0 +1,9 @@
+package server
+
+import (
+	"github.com/gin-gonic/gin"
+)
+
+func (s *Server) handlePrometheusMetrics(c *gin.Context) {
+	s.Agent.PromHandler().ServeHTTP(c.Writer, c.Request)
+}

api/server/server.go

@@ -280,6 +280,7 @@ func (s *Server) bindHandlers(ctx context.Context) {
 	engine.GET("/", handlePing)
 	engine.GET("/version", handleVersion)
 	engine.GET("/stats", s.handleStats)
+	engine.GET("/metrics", s.handlePrometheusMetrics)
 
 	{
 		v1 := engine.Group("/v1")