Added basic metrics for streaming system

Author: Ian Pye

util/slog/slog.go

@@ -1,15 +1,20 @@
 package slog
 
 import (
+	json "encoding/json"
 	"fmt"
+	zmq "github.com/pebbe/zmq3"
 	"logger"
 	"os"
+	"stash.cloudflare.com/go-stream/util"
 	"strings"
+	"time"
 )
 
 var (
 	LogPrefix string
-	glog      *logger.Logger // the main logger object
+	glog      *logger.Logger         // the main logger object
+	Gm        *util.StreamingMetrics // Main metrics object
 )
 
 // fatal: outputs a fatal startup error to STDERR, logs it to the
@@ -30,7 +35,7 @@ func exit(code int, l *logger.Logger, format string, v ...interface{}) {
 	os.Exit(code)
 }
 
-func Init(logName *string, logLevel *string, logPrefix *string) {
+func Init(logName *string, logLevel *string, logPrefix *string, metrics *util.StreamingMetrics, metricsAddr *string) {
 	// Change logger level
 	if err := logger.SetLogName(*logName); err != nil {
 		fatal(nil, "Cannot set log name for program")
@@ -45,6 +50,9 @@ func Init(logName *string, logLevel *string, logPrefix *string) {
 			fatal(nil, "Cannot start logger")
 		}
 	}
+
+	Gm = metrics
+	go statsSender(metricsAddr, logPrefix)
 }
 
 func Logf(level logger.Level, format string, v ...interface{}) {
@@ -56,3 +64,45 @@ func Logf(level logger.Level, format string, v ...interface{}) {
 func Fatalf(format string, v ...interface{}) {
 	fatal(glog, format, v)
 }
+
+type statsPkg struct {
+	Name        string
+	TotalMsgs   int64
+	TotalErrors int64
+	UpTime      int64
+}
+
+func statsSender(metricsAddr *string, processName *string) {
+
+	rep, err := zmq.NewSocket(zmq.REP)
+	if err != nil {
+		Logf(logger.Levels.Error, "Stats Sender error: %v", err.Error())
+		return
+	}
+	defer rep.Close()
+	err = rep.Bind(*metricsAddr)
+	if err != nil {
+		Logf(logger.Levels.Error, "Stats Sender error: %v", err.Error())
+		return
+	}
+
+	Logf(logger.Levels.Info, "Stats sender, listening on %s", *metricsAddr)
+
+	// Loop, printing the stats on request
+	for {
+		_, err := rep.Recv(0)
+		if err != nil {
+			Logf(logger.Levels.Error, "%v", err.Error())
+		} else {
+			timestamp := time.Now().Unix() - Gm.StartTime
+			dBag := statsPkg{*processName, Gm.Total.Count(), Gm.Error.Count(), timestamp}
+			stats, err := json.Marshal(dBag)
+			if err == nil {
+				_, err = rep.SendBytes(stats, zmq.DONTWAIT)
+				if err != nil {
+					Logf(logger.Levels.Error, "%v", err.Error())
+				}
+			}
+		}
+	}
+}

util/util.go

@@ -1,8 +1,9 @@
 package util
 
-import "log"
-
-var _ = log.Printf
+import (
+	metrics "github.com/rcrowley/go-metrics"
+	"time"
+)
 
 type MemoryBuffer struct {
 	buf     [][]byte
@@ -165,3 +166,21 @@ func (buf *SequentialBufferChanImpl) Reset() [][]byte {
 	buf.seq = len(buf.chanbuf) + 1
 	return ret
 }
+
+type StreamingMetrics struct {
+	Reg       metrics.Registry
+	Total     metrics.Counter // total count of packets.
+	Current   metrics.Counter // packets in the last period
+	Error     metrics.Counter // total count of packets that are dropped
+	StartTime int64           // How long we've been running for
+}
+
+func NewStreamingMetrics(mReg metrics.Registry) *StreamingMetrics {
+	return &StreamingMetrics{
+		Reg:       mReg,
+		Total:     metrics.NewCounter(),
+		Current:   metrics.NewCounter(),
+		Error:     metrics.NewCounter(),
+		StartTime: time.Now().Unix(),
+	}
+}