From f955f7ef0dfe9a501d6e18e107e64ae9e4ca9372 Mon Sep 17 00:00:00 2001
From: Waldemar Quevedo <wally@nats.io>
Date: Thu, 6 Apr 2023 09:44:14 -0600
Subject: [PATCH] feat: show rates msgs and bytes per sec using spc

Signed-off-by: Waldemar Quevedo <wally@nats.io>
---
 nats-top.go      | 32 ++++++++++++++++++++++++++++++--
 util/toputils.go | 38 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 68 insertions(+), 2 deletions(-)

diff --git a/nats-top.go b/nats-top.go
index e7ec39c..f209b96 100644
--- a/nats-top.go
+++ b/nats-top.go
@@ -356,8 +356,30 @@ func generateParagraphPlainText(
 		}
 
 		connLineInfo = append(connLineInfo, conn.NumSubs)
-		connLineInfo = append(connLineInfo, top.Psize(*displayRawBytes, int64(conn.Pending)), top.Psize(*displayRawBytes, conn.OutMsgs), top.Psize(*displayRawBytes, conn.InMsgs))
-		connLineInfo = append(connLineInfo, top.Psize(*displayRawBytes, conn.OutBytes), top.Psize(*displayRawBytes, conn.InBytes))
+
+		connLineInfo = append(connLineInfo, top.Psize(*displayRawBytes, int64(conn.Pending)))
+
+		if !engine.ShowRates {
+			connLineInfo = append(connLineInfo, top.Psize(*displayRawBytes, conn.OutMsgs), top.Psize(*displayRawBytes, conn.InMsgs))
+			connLineInfo = append(connLineInfo, top.Psize(*displayRawBytes, conn.OutBytes), top.Psize(*displayRawBytes, conn.InBytes))
+		} else {
+			var (
+				inMsgsPerSec   float64
+				outMsgsPerSec  float64
+				inBytesPerSec  float64
+				outBytesPerSec float64
+			)
+			crate, wasConnected := stats.Rates.Connections[conn.Cid]
+			if wasConnected {
+				outMsgsPerSec = crate.OutMsgsRate
+				inMsgsPerSec = crate.InMsgsRate
+				outBytesPerSec = crate.OutBytesRate
+				inBytesPerSec = crate.InBytesRate
+			}
+			connLineInfo = append(connLineInfo, top.Psize(*displayRawBytes, int64(outMsgsPerSec)), top.Psize(*displayRawBytes, int64(inMsgsPerSec)))
+			connLineInfo = append(connLineInfo, top.Psize(*displayRawBytes, int64(outBytesPerSec)), top.Psize(*displayRawBytes, int64(inBytesPerSec)))
+		}
+
 		connLineInfo = append(connLineInfo, conn.Lang, conn.Version)
 		connLineInfo = append(connLineInfo, conn.Uptime, conn.LastActivity)
 
@@ -670,6 +692,10 @@ func StartUI(engine *top.Engine) {
 				fmt.Printf("%slimit   [%d]: %s", UI_HEADER_PREFIX, engine.Conns, optionBuf)
 			}
 
+			if e.Type == ui.EventKey && e.Key == ui.KeySpace {
+				engine.ShowRates = !engine.ShowRates
+			}
+
 			if e.Type == ui.EventKey && (e.Ch == 'q' || e.Key == ui.KeyCtrlC) {
 				close(engine.ShutdownCh)
 				cleanExit()
@@ -760,6 +786,8 @@ d                Toggle activating DNS address lookup for clients.
 
 b                Toggle displaying raw bytes.
 
+space            Toggle displaying rates per second in connections.
+
 q                Quit nats-top.
 
 Press any key to continue...
diff --git a/util/toputils.go b/util/toputils.go
index e3d6fb7..e4c76f2 100644
--- a/util/toputils.go
+++ b/util/toputils.go
@@ -28,6 +28,8 @@ type Engine struct {
 	ShutdownCh   chan struct{}
 	LastStats    *Stats
 	LastPollTime time.Time
+	ShowRates    bool
+	LastConnz    map[uint64]*server.ConnInfo
 }
 
 func NewEngine(host string, port int, conns int, delay int) *Engine {
@@ -38,6 +40,7 @@ func NewEngine(host string, port int, conns int, delay int) *Engine {
 		Delay:      delay,
 		StatsCh:    make(chan *Stats),
 		ShutdownCh: make(chan struct{}),
+		LastConnz:  make(map[uint64]*server.ConnInfo),
 	}
 }
 
@@ -185,6 +188,12 @@ func (engine *Engine) fetchStats() *Stats {
 	inBytesLastVal = inBytesVal
 	outBytesLastVal = outBytesVal
 
+	// Snapshot per sec metrics for connections.
+	connz := make(map[uint64]*server.ConnInfo)
+	for _, conn := range stats.Connz.Conns {
+		connz[conn.Cid] = conn
+	}
+
 	now := time.Now()
 	tdelta := now.Sub(engine.LastPollTime)
 
@@ -200,12 +209,33 @@ func (engine *Engine) fetchStats() *Stats {
 		OutMsgsRate:  outMsgsRate,
 		InBytesRate:  inBytesRate,
 		OutBytesRate: outBytesRate,
+		Connections:  make(map[uint64]*ConnRates),
 	}
+
+	// Measure per connection metrics.
+	for cid, conn := range connz {
+		cr := &ConnRates{
+			InMsgsRate:   0,
+			OutMsgsRate:  0,
+			InBytesRate:  0,
+			OutBytesRate: 0,
+		}
+		lconn, wasConnected := engine.LastConnz[cid]
+		if wasConnected {
+			cr.InMsgsRate = float64(conn.InMsgs - lconn.InMsgs)
+			cr.OutMsgsRate = float64(conn.OutMsgs - lconn.OutMsgs)
+			cr.InBytesRate = float64(conn.InBytes - lconn.InBytes)
+			cr.OutBytesRate = float64(conn.OutBytes - lconn.OutBytes)
+		}
+		rates.Connections[cid] = cr
+	}
+
 	stats.Rates = rates
 
 	// Snapshot stats.
 	engine.LastStats = stats
 	engine.LastPollTime = now
+	engine.LastConnz = connz
 
 	return stats
 }
@@ -265,6 +295,14 @@ type Rates struct {
 	OutMsgsRate  float64
 	InBytesRate  float64
 	OutBytesRate float64
+	Connections  map[uint64]*ConnRates
+}
+
+type ConnRates struct {
+	InMsgsRate   float64
+	OutMsgsRate  float64
+	InBytesRate  float64
+	OutBytesRate float64
 }
 
 const kibibyte = 1024