From 3e5381840cb9f08a1352884be8daafd8550bf83c Mon Sep 17 00:00:00 2001
From: Ivan Danyliuk <ivan.daniluk@gmail.com>
Date: Thu, 24 May 2018 18:51:16 +0300
Subject: [PATCH] Add Prometheus metrics

---
 main.go   | 11 +++++-----
 sender.go | 15 ++++++--------
 stats.go  | 62 ++++++++++++++++++++++++++++++++-----------------------
 3 files changed, 48 insertions(+), 40 deletions(-)

diff --git a/main.go b/main.go
index 1b9afc2..c840f6b 100644
--- a/main.go
+++ b/main.go
@@ -12,10 +12,11 @@ import (
 
 func main() {
 	var (
-		name     = flag.String("name", "randomstring", "Public chat name used for this health bots")
-		interval = flag.Duration("interval", 5*time.Second, "Interval for health check")
-		rpcHost  = flag.String("rpc", "http://localhost:8545", "Host:port to statusd's RPC endpoint")
-		isSender = flag.Bool("send", true, "Select bot role, sender or responder")
+		name      = flag.String("name", "randomstring", "Public chat name used for this health bots")
+		interval  = flag.Duration("interval", 5*time.Second, "Interval for health check")
+		rpcHost   = flag.String("rpc", "http://localhost:8545", "Host:port to statusd's RPC endpoint")
+		statsPort = flag.String("statsPort", ":8080", "Host:port to bind to for exposed Prometheus metrics")
+		isSender  = flag.Bool("send", true, "Select bot role, sender or responder")
 	)
 	flag.Parse()
 
@@ -37,7 +38,7 @@ func main() {
 	}
 
 	if *isSender {
-		startSender(ch, *interval)
+		startSender(ch, *interval, *statsPort)
 	} else {
 		startReceiver(ch)
 		select {}
diff --git a/sender.go b/sender.go
index 2540481..d5583c7 100644
--- a/sender.go
+++ b/sender.go
@@ -7,13 +7,12 @@ import (
 	"github.com/status-im/status-go-sdk"
 )
 
-func startSender(ch *sdk.Channel, interval time.Duration) {
+func startSender(ch *sdk.Channel, interval time.Duration, statsPort string) {
 	var (
-		counter     int
-		ticker      = time.NewTicker(interval)
-		statsTicker = time.NewTicker(10 * time.Second)
-		pending     = make(map[int]time.Time)
-		recvCh      = make(chan Msg, 1000)
+		counter int
+		ticker  = time.NewTicker(interval)
+		pending = make(map[int]time.Time)
+		recvCh  = make(chan Msg, 1000)
 	)
 
 	if _, err := ch.Subscribe(func(m *sdk.Msg) {
@@ -30,7 +29,7 @@ func startSender(ch *sdk.Channel, interval time.Duration) {
 		log.Fatal(err)
 	}
 
-	stats := NewStats()
+	stats := NewStats(statsPort)
 
 	for {
 		select {
@@ -58,8 +57,6 @@ func startSender(ch *sdk.Channel, interval time.Duration) {
 			delete(pending, c)
 			dur := time.Since(start)
 			stats.AddRountrip(dur)
-		case <-statsTicker.C:
-			stats.Print()
 		}
 	}
 }
diff --git a/stats.go b/stats.go
index 772bbfa..17b3e3e 100644
--- a/stats.go
+++ b/stats.go
@@ -1,47 +1,57 @@
 package main
 
 import (
-	"fmt"
-	"sync"
+	"log"
+	"net/http"
 	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
+var (
+	msgsSent = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "msgs_sent",
+		Help: "Messages sent by bot",
+	})
+	msgsReceived = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "msgs_received",
+		Help: "Message responses received by bot",
+	})
+	msgsLatencies = prometheus.NewSummary(prometheus.SummaryOpts{
+		Name:       "msgs_responses_latency",
+		Help:       "Latencies of responses to bot messages",
+		Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+	})
+)
+
+func init() {
+	// Metrics have to be registered to be exposed
+	prometheus.MustRegister(msgsSent)
+	prometheus.MustRegister(msgsReceived)
+	prometheus.MustRegister(msgsLatencies)
+}
+
 // Stats represents messages' statistics.
 type Stats struct {
-	mx       sync.RWMutex
-	sent     int
-	received int
-	delays   []time.Duration
 }
 
 // NewStats returns new empty Stats object.
-func NewStats() *Stats {
+func NewStats(statsPort string) *Stats {
+	go func() {
+		http.Handle("/metrics", promhttp.Handler())
+		log.Fatal(http.ListenAndServe(statsPort, nil))
+	}()
 	return &Stats{}
 }
 
 // AddSent adds information about sent messages.
 func (s *Stats) AddSent() {
-	s.mx.Lock()
-	defer s.mx.Unlock()
-	s.sent++
+	msgsSent.Inc()
 }
 
 // AddRoundtrip adds information about successful message roundtrip.
 func (s *Stats) AddRountrip(d time.Duration) {
-	s.mx.Lock()
-	defer s.mx.Unlock()
-	s.received++
-	s.delays = append(s.delays, d)
-}
-
-// Print dumps stats to the console.
-func (s *Stats) Print() {
-	s.mx.RLock()
-	defer s.mx.RUnlock()
-	fmt.Println("-------------------------")
-	fmt.Println("Time:", time.Now())
-	fmt.Println("Sent:", s.sent)
-	fmt.Println("Received:", s.received)
-	fmt.Println("Delays:", s.delays)
-	fmt.Println("-------------------------")
+	msgsReceived.Inc()
+	msgsLatencies.Observe(float64(d / time.Millisecond))
 }