Improve rate calcuations for graphs

Here we check if the value from Nats the we use to in the
server, stream and consumer graph actions is > 0 before
using it to calculate rate. This will prevent us from
storing negative rates and breaking the graphs.

cli/consumer_command.go

@@ -500,8 +500,8 @@ func (c *consumerCmd) graphAction(_ *fisk.ParseContext) error {
 				continue
 			}
 
-			deliveredRates = append(deliveredRates, float64(nfo.Delivered.Stream-lastDeliveredSeq)/time.Since(lastStateTs).Seconds())
-			ackedRates = append(ackedRates, float64(nfo.AckFloor.Stream-lastAckedSeq)/time.Since(lastStateTs).Seconds())
+			deliveredRates = append(deliveredRates, calculateRate(float64(nfo.Delivered.Stream), float64(lastDeliveredSeq), time.Since(lastStateTs)))
+			ackedRates = append(ackedRates, calculateRate(float64(nfo.AckFloor.Stream), float64(lastAckedSeq), time.Since(lastStateTs)))
 			unprocessedMessages = append(unprocessedMessages, float64(nfo.NumPending))
 			outstandingMessages = append(outstandingMessages, float64(nfo.NumAckPending))
 			lastDeliveredSeq = nfo.Delivered.Stream

cli/server_graph_command.go

@@ -258,8 +258,9 @@ func (c *SrvGraphCmd) graphServer() error {
 		memUsed = c.resizeData(memUsed, width, float64(vz.Mem)/1024/1024)
 		connections = c.resizeData(connections, width, float64(vz.Connections))
 		subscriptions = c.resizeData(subscriptions, width, float64(vz.Subscriptions))
-		messagesRate = c.resizeData(messagesRate, width, (float64(vz.InMsgs+vz.OutMsgs)-lastMessages)/time.Since(lastStateTs).Seconds())
-		bytesRate = c.resizeData(bytesRate, width, (float64(vz.InBytes+vz.OutBytes)-lastByes)/time.Since(lastStateTs).Seconds())
+
+		messagesRate = c.resizeData(messagesRate, width, calculateRate(float64(vz.InMsgs+vz.OutMsgs), lastMessages, time.Since(lastStateTs)))
+		bytesRate = c.resizeData(bytesRate, width, calculateRate(float64(vz.InBytes+vz.OutBytes), lastByes, time.Since(lastStateTs)))
 
 		lastMessages = float64(vz.InMsgs + vz.OutMsgs)
 		lastByes = float64(vz.InBytes + vz.OutBytes)

cli/stream_command.go

@@ -490,8 +490,8 @@ func (c *streamCmd) graphAction(_ *fisk.ParseContext) error {
 			}
 
 			messagesStored = append(messagesStored, float64(nfo.Msgs))
-			messageRates = append(messageRates, float64(nfo.LastSeq-lastLastSeq)/time.Since(lastStateTs).Seconds())
-			limitedRates = append(limitedRates, float64(nfo.FirstSeq-lastFirstSeq)/time.Since(lastStateTs).Seconds())
+			messageRates = append(messageRates, calculateRate(float64(nfo.LastSeq), float64(lastLastSeq), time.Since(lastStateTs)))
+			limitedRates = append(limitedRates, calculateRate(float64(nfo.FirstSeq), float64(lastFirstSeq), time.Since(lastStateTs)))
 
 			lastStateTs = time.Now()
 			lastLastSeq = nfo.LastSeq

cli/util.go

@@ -1229,3 +1229,12 @@ func currentActiveServers(nc *nats.Conn) (int, error) {
 
 	return expect, err
 }
+
+func calculateRate(new, last float64, since time.Duration) float64 {
+	// If new == 0 we have missed a data point from nats.
+	// Return the previous calculation so that it doesn't break graphs
+	if new == 0 {
+		return last
+	}
+	return (new - last) / since.Seconds()
+}