Use SOH timestamp to calculate bandwidth in TM

CHANGELOG.md

@@ -57,6 +57,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 - [#7469](https://github.com/apache/trafficcontrol/pull/7469) *Traffic Ops* Changed logic to not report empty or missing cookies into TO error.log.
 
 ### Fixed
+- [#7539](https://github.com/apache/trafficcontrol/pull/7539) *Traffic Monitor* Use stats_over_http timestamp to calculate bandwidth for TM's health. 
 - [#7542](https://github.com/apache/trafficcontrol/pull/7542) *Traffic Ops* Fixed `CDN Locks` documentation to reflect the correct RFC3339 timestamps.
 - [#6340](https://github.com/apache/trafficcontrol/issues/6340) *Traffic Ops* Fixed alert messages for POST and PUT invalidation job APIs.
 - [#7511](https://github.com/apache/trafficcontrol/pull/7511) *Traffic Ops* Fixed the changelog registration message to include the username instead of duplicate email entry.

traffic_monitor/cache/cache.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"io"
 	"regexp"
+	"strconv"
 	"time"
 
 	"github.com/apache/trafficcontrol/lib/go-log"
@@ -306,6 +307,17 @@ func (handler Handler) Handle(id string, rdr io.Reader, format string, reqTime t
 		handler.resultChan <- result
 		return
 	}
+	if val, ok := miscStats["current_time_epoch_ms"]; ok {
+		valString := fmt.Sprintf("%s", val)
+		valInt, valErr := strconv.ParseInt(valString, 10, 64)
+		if valErr != nil {
+			log.Errorf("parse error '%v'", valErr)
+			result.Error = valErr
+			handler.resultChan <- result
+			return
+		}
+		result.Time = time.UnixMilli(valInt)
+	}
 	if value, ok := miscStats[rfc.Via]; ok {
 		result.ID = fmt.Sprintf("%v", value)
 	}

traffic_monitor/cache/cache_test.go

@@ -20,11 +20,15 @@ package cache
  */
 
 import (
-	"testing"
-
+	"bytes"
+	"fmt"
 	"github.com/apache/trafficcontrol/lib/go-tc"
 	"github.com/apache/trafficcontrol/lib/go-util"
+	"github.com/apache/trafficcontrol/traffic_monitor/poller"
 	"github.com/apache/trafficcontrol/traffic_monitor/todata"
+	"io/ioutil"
+	"net/http"
+	"testing"
 )
 
 func TestHandlerPrecompute(t *testing.T) {
@@ -95,3 +99,37 @@ func TestComputeStatGbps(t *testing.T) {
 		}
 	}
 }
+
+func TestParseAndDecode(t *testing.T) {
+	file, err := ioutil.ReadFile("stats_over_http.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pl := &poller.HTTPPollCtx{HTTPHeader: http.Header{}}
+	ctx := interface{}(pl)
+	ctx.(*poller.HTTPPollCtx).HTTPHeader.Set("Content-Type", "text/json")
+
+	decoder, err := GetDecoder("stats_over_http")
+	if err != nil {
+		t.Errorf("decoder error, expected: nil, got: %v", err)
+	}
+
+	_, miscStats, err := decoder.Parse("1", bytes.NewReader(file), ctx)
+	if err != nil {
+		t.Errorf("decoder parse error, expected: nil, got: %v", err)
+	}
+
+	if len(miscStats) < 1 {
+		t.Errorf("empty miscStats structure")
+	}
+
+	if val, ok := miscStats["current_time_epoch_ms"]; ok {
+		valString := fmt.Sprintf("%s", val)
+		if valString != "1684784878894" {
+			t.Errorf("unable to read `current_time_epoch_ms`")
+		}
+	} else {
+		t.Errorf("current_time_epoch_ms field was not found in the json file")
+	}
+}

traffic_monitor/cache/stats_over_http.json

@@ -533,6 +533,7 @@
 		"plugin.system_stats.net.docker0.rx_length_errors": "0",
 		"proxy.process.cache.volume_0.span.offline": "0",
 		"proxy.process.cache.volume_0.span.online": "0",
+		"current_time_epoch_ms": "1684784878894",
 		"server": "10.0.0"
 	}
 }

traffic_monitor/health/cache.go

@@ -60,6 +60,7 @@ func (t Threshold) String() string {
 
 // GetVitals Gets the vitals to decide health on in the right format
 func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *tc.TrafficMonitorConfigMap) {
+	var elapsedTimeInSecs float64
 	if newResult.Error != nil {
 		log.Errorf("cache_health.GetVitals() called with an errored Result!")
 		return
@@ -87,6 +88,14 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *tc.Traffic
 		return
 	}
 
+	if prevResult != nil {
+		elapsedTimeInSecs = float64(newResult.Time.UnixMilli()-prevResult.Time.UnixMilli()) / 1000
+		if elapsedTimeInSecs <= 0 {
+			*newResult = *prevResult
+			return
+		}
+	}
+
 	var monitoredInterfaces []tc.ServerInterfaceInfo
 	for _, srvrIfaceInfo := range mc.TrafficServer[newResult.ID].Interfaces {
 		if srvrIfaceInfo.Monitor {
@@ -115,7 +124,6 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *tc.Traffic
 		}
 
 		if prevResult != nil && prevResult.InterfaceVitals != nil && prevResult.InterfaceVitals[ifaceName].BytesOut != 0 {
-			elapsedTimeInSecs := float64(newResult.Time.UnixNano()-prevResult.Time.UnixNano()) / 1000000000
 			ifaceVitals.KbpsOut = int64(float64((ifaceVitals.BytesOut-prevResult.InterfaceVitals[ifaceName].BytesOut)*8/1000) / elapsedTimeInSecs)
 		}
 		newResult.InterfaceVitals[ifaceName] = ifaceVitals
@@ -128,7 +136,6 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *tc.Traffic
 	}
 
 	if prevResult != nil && prevResult.Vitals.BytesOut != 0 {
-		elapsedTimeInSecs := float64(newResult.Time.UnixNano()-prevResult.Time.UnixNano()) / 1000000000
 		newResult.Vitals.KbpsOut = int64(float64((newResult.Vitals.BytesOut-prevResult.Vitals.BytesOut)*8/1000) / elapsedTimeInSecs)
 	}
 

traffic_monitor/health/cache_test.go

@@ -310,6 +310,14 @@ func TestDualHomingMonitoredInterfacesGetVitals(t *testing.T) {
 	if firstResult.Vitals != expectedFirstVitals {
 		t.Errorf("Vitals do not match expected output. expected: %v actual: %v:", expectedFirstVitals, firstResult.Vitals)
 	}
+
+	//Test if elapsedTimeInSecs == 0
+	secondResult.Time = firstResult.Time
+	GetVitals(&secondResult, &firstResult, &tmcm)
+	if firstResult.Statistics.Interfaces["bond0"] != secondResult.Statistics.Interfaces["bond0"] {
+		t.Errorf("Load avg statistics do not match. expected: %v, got: %v", firstResult.Statistics.Interfaces["bond0"], secondResult.Statistics.Interfaces["bond0"])
+	}
+
 }
 
 func TestCalcAvailabilityThresholds(t *testing.T) {