Adding more extensive metrics

Author: Ian Pye

cluster/manager.go

@@ -5,6 +5,7 @@ import "time"
 /* There should be a separate manager for each service */
 type Manager interface {
 	GetEra(t time.Time) Era
+	GetCurrentEra() Era
 }
 
 type StaticManager struct {
@@ -15,6 +16,10 @@ func (c *StaticManager) GetEra(t time.Time) Era {
 	return c.E
 }
 
+func (c *StaticManager) GetCurrentEra() Era {
+	return c.E
+}
+
 func NewStaticManager(e Era) *StaticManager {
 	return &StaticManager{e}
 }

stream/chain.go

@@ -170,6 +170,11 @@ func (c *inChain) In() chan Object {
 	return ops[0].(In).In()
 }
 
+func (c *inChain) GetInDepth() int {
+	ops := c.Operators()
+	return ops[0].(In).GetInDepth()
+}
+
 func (c *inChain) SetIn(ch chan Object) {
 	ops := c.Operators()
 	ops[0].(In).SetIn(ch)

stream/operator.go

@@ -42,6 +42,7 @@ type Out interface {
 type In interface {
 	In() chan Object
 	SetIn(c chan Object)
+	GetInDepth() int
 }
 
 type InOutOperator interface {

stream/util.go

@@ -21,6 +21,10 @@ func (o *BaseIn) In() chan Object {
 	return o.in
 }
 
+func (o *BaseIn) GetInDepth() int {
+	return len(o.in)
+}
+
 func (o *BaseIn) SetIn(c chan Object) {
 	o.in = c
 }

transport/client.go

@@ -72,6 +72,15 @@ func (src *Client) Run() error {
 	defer func() {
 		src.running = false
 	}()
+
+	slog.Gm.Register(stream.Name(src))
+	go func(op string, s *Client) { // Update the queue depth on input for each phase
+		for {
+			slog.Gm.Update(&op, s.GetInDepth())
+			time.Sleep(1 * time.Second)
+		}
+	}(stream.Name(src), src)
+
 	for src.retries < 3 {
 		err := src.connect()
 		if err == nil {
@@ -163,6 +172,7 @@ func (src *Client) connect() error {
 	closing := false
 
 	//defer log.Println("Exiting client loop")
+	opName := stream.Name(src)
 	writesNotCompleted := uint(0)
 	for {
 		upstreamCh := src.In()
@@ -189,6 +199,7 @@ func (src *Client) connect() error {
 				}
 				sendData(sndChData, bytes, seq)
 				writesNotCompleted += 1
+				slog.Gm.Event(&opName) // These are bactched
 			}
 		case cnt := <-writeNotifier.NotificationChannel():
 			writesNotCompleted -= cnt

transport/server.go

@@ -73,6 +73,7 @@ func (src Server) Run() error {
 		hardCloseListener(src.StopNotifier, scl, ln)
 	}()
 
+	slog.Gm.Register(stream.Name(src))
 	for {
 		conn, err := ln.Accept()
 		if err != nil {
@@ -121,6 +122,7 @@ func (src Server) handleConnection(conn net.Conn) {
 	wg_sub := &sync.WaitGroup{}
 	defer wg_sub.Wait()
 
+	opName := stream.Name(src)
 	sndChData := make(chan stream.Object, 100)
 	sndChCloseNotifier := make(chan bool, 1)
 	defer close(sndChData)
@@ -168,6 +170,7 @@ func (src Server) handleConnection(conn net.Conn) {
 				return
 			}
 			command, seq, payload, err := parseMsg(obj.([]byte))
+			slog.Gm.Event(&opName)
 
 			if err == nil {
 				if command == DATA {

util/slog/slog.go

@@ -66,10 +66,9 @@ func Fatalf(format string, v ...interface{}) {
 }
 
 type statsPkg struct {
-	Name        string
-	TotalMsgs   int64
-	TotalErrors int64
-	UpTime      int64
+	Name      string
+	UpTime    int64
+	OpMetrics map[string]interface{}
 }
 
 func statsSender(metricsAddr *string, processName *string) {
@@ -95,7 +94,10 @@ func statsSender(metricsAddr *string, processName *string) {
 			Logf(logger.Levels.Error, "%v", err.Error())
 		} else {
 			timestamp := time.Now().Unix() - Gm.StartTime
-			dBag := statsPkg{*processName, Gm.Total.Count(), Gm.Error.Count(), timestamp}
+			dBag := statsPkg{*processName, timestamp, map[string]interface{}{}}
+			for k, v := range Gm.OpGroups {
+				dBag.OpMetrics[k] = map[string]int64{"Events": v.Events.Count(), "Errors": v.Errors.Count(), "Queue": v.QueueLength.Value()}
+			}
 			stats, err := json.Marshal(dBag)
 			if err == nil {
 				_, err = rep.SendBytes(stats, zmq.DONTWAIT)

util/util.go

@@ -167,20 +167,38 @@ func (buf *SequentialBufferChanImpl) Reset() [][]byte {
 	return ret
 }
 
+type MetricsGroup struct {
+	Events      metrics.Counter
+	Errors      metrics.Counter
+	QueueLength metrics.Gauge
+}
+
 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
+	OpGroups  map[string]MetricsGroup // Each Op can have an associated metrics group
+	StartTime int64                   // How long we've been running for
+}
+
+func (m *StreamingMetrics) Event(op *string) {
+	m.OpGroups[*op].Events.Inc(1)
+}
+
+func (m *StreamingMetrics) Error(op *string) {
+	m.OpGroups[*op].Errors.Inc(1)
+}
+
+func (m *StreamingMetrics) Update(op *string, v int) {
+	m.OpGroups[*op].QueueLength.Update(int64(v))
+}
+
+func (m *StreamingMetrics) Register(op string) {
+	m.OpGroups[op] = MetricsGroup{metrics.NewCounter(), metrics.NewCounter(), metrics.NewGauge()}
 }
 
 func NewStreamingMetrics(mReg metrics.Registry) *StreamingMetrics {
 	return &StreamingMetrics{
 		Reg:       mReg,
-		Total:     metrics.NewCounter(),
-		Current:   metrics.NewCounter(),
-		Error:     metrics.NewCounter(),
+		OpGroups:  make(map[string]MetricsGroup),
 		StartTime: time.Now().Unix(),
 	}
 }