Merge pull request #2116 from gravitl/GRA-1321

GRA-1321: Metrics Fix

logic/metrics/metrics.go

@@ -10,7 +10,7 @@ import (
 )
 
 // Collect - collects metrics
-func Collect(iface, server, network string, peerMap models.PeerMap) (*models.Metrics, error) {
+func Collect(iface, server, network string, peerMap models.PeerMap, proxy bool) (*models.Metrics, error) {
 	var metrics models.Metrics
 	metrics.Connectivity = make(map[string]models.Metric)
 	var wgclient, err = wgctrl.New()
@@ -45,6 +45,7 @@ func Collect(iface, server, network string, peerMap models.PeerMap) (*models.Met
 		newMetric.TotalSent = int64(proxyMetrics.TrafficSent)
 		newMetric.Latency = int64(proxyMetrics.LastRecordedLatency)
 		newMetric.Connected = proxyMetrics.NodeConnectionStatus[id]
+		newMetric.CollectedByProxy = proxy
 		if newMetric.Connected {
 			newMetric.Uptime = 1
 		}

logic/peers.go

@@ -90,6 +90,7 @@ func GetProxyUpdateForHost(ctx context.Context, host *models.Host) (models.Proxy
 				currPeerConf = models.PeerConf{
 					Proxy:            peerHost.ProxyEnabled,
 					PublicListenPort: int32(GetPeerListenPort(peerHost)),
+					ProxyListenPort:  GetProxyListenPort(peerHost),
 				}
 			}
 
@@ -278,6 +279,7 @@ func GetPeerUpdateForHost(ctx context.Context, network string, host *models.Host
 						Address:         peer.PrimaryAddress(),
 						Name:            peerHost.Name,
 						Network:         peer.Network,
+						ProxyListenPort: GetProxyListenPort(peerHost),
 					}
 					nodePeer = peerConfig
 				} else {
@@ -289,6 +291,7 @@ func GetPeerUpdateForHost(ctx context.Context, network string, host *models.Host
 						Address:         peer.PrimaryAddress(),
 						Name:            peerHost.Name,
 						Network:         peer.Network,
+						ProxyListenPort: GetProxyListenPort(peerHost),
 					}
 					nodePeer = hostPeerUpdate.Peers[peerIndexMap[peerHost.PublicKey.String()]]
 				}
@@ -402,6 +405,15 @@ func GetPeerListenPort(host *models.Host) int {
 	return peerPort
 }
 
+// GetProxyListenPort - fetches the proxy listen port
+func GetProxyListenPort(host *models.Host) int {
+	proxyPort := host.ProxyListenPort
+	if host.PublicListenPort != 0 {
+		proxyPort = host.PublicListenPort
+	}
+	return proxyPort
+}
+
 func getExtPeers(node *models.Node) ([]wgtypes.PeerConfig, []models.IDandAddr, error) {
 	var peers []wgtypes.PeerConfig
 	var idsAndAddr []models.IDandAddr

models/metrics.go

@@ -24,6 +24,7 @@ type Metric struct {
 	ActualUptime     time.Duration `json:"actualuptime" bson:"actualuptime" yaml:"actualuptime"`
 	PercentUp        float64       `json:"percentup" bson:"percentup" yaml:"percentup"`
 	Connected        bool          `json:"connected" bson:"connected" yaml:"connected"`
+	CollectedByProxy bool          `json:"collected_by_proxy" bson:"collected_by_proxy" yaml:"collected_by_proxy"`
 }
 
 // IDandAddr - struct to hold ID and primary Address

models/proxy.go

@@ -37,6 +37,7 @@ type RelayedConf struct {
 type PeerConf struct {
 	Proxy            bool         `json:"proxy"`
 	PublicListenPort int32        `json:"public_listen_port"`
+	ProxyListenPort  int          `json:"proxy_listen_port"`
 	IsExtClient      bool         `json:"is_ext_client"`
 	Address          net.IP       `json:"address"`
 	ExtInternalIp    net.IP       `json:"ext_internal_ip"`

mq/handlers.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -362,6 +363,21 @@ func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) boo
 		oldMetric := oldMetrics.Connectivity[k]
 		currMetric.TotalTime += oldMetric.TotalTime
 		currMetric.Uptime += oldMetric.Uptime // get the total uptime for this connection
+		if currMetric.CollectedByProxy {
+			currMetric.TotalReceived += oldMetric.TotalReceived
+			currMetric.TotalSent += oldMetric.TotalSent
+		} else {
+			if currMetric.TotalReceived < oldMetric.TotalReceived {
+				currMetric.TotalReceived += oldMetric.TotalReceived
+			} else {
+				currMetric.TotalReceived += int64(math.Abs(float64(currMetric.TotalReceived) - float64(oldMetric.TotalReceived)))
+			}
+			if currMetric.TotalSent < oldMetric.TotalSent {
+				currMetric.TotalSent += oldMetric.TotalSent
+			} else {
+				currMetric.TotalSent += int64(math.Abs(float64(currMetric.TotalSent) - float64(oldMetric.TotalSent)))
+			}
+		}
 		if currMetric.Uptime == 0 || currMetric.TotalTime == 0 {
 			currMetric.PercentUp = 0
 		} else {