Implement timeout for whole metrics send function

.github/workflows/build.yaml

@@ -91,9 +91,9 @@ jobs:
           tar -czvf "../releases/livepeer-catalyst-api-${GOOS}-${GOARCH}.tar.gz" .
 
       - name: Upload artifacts for cutting release
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
-          name: release-artifacts
+          name: release-artifacts-${{ matrix.platform.name }}-${{ matrix.arch }}
           path: releases/
 
   upload:
@@ -107,10 +107,11 @@ jobs:
       - build
     steps:
       - name: Download artifacts
-        uses: actions/download-artifact@v3
+        uses: actions/download-artifact@v4
         with:
-          name: release-artifacts
+          pattern: release-artifacts-*
           path: releases/
+          merge-multiple: true
 
       - name: Generate sha256 checksum and gpg signatures for release artifacts
         uses: livepeer/action-gh-checksum-and-gpg-sign@latest

.github/workflows/test.yaml

@@ -90,7 +90,7 @@ jobs:
           make generate integration-test
 
       - name: Upload logs
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         if: failure()
         with:
           name: cucumber-logs

balancer/catabalancer/catalyst_balancer.go

@@ -421,58 +421,76 @@ func StartMetricSending(nodeName string, latitude float64, longitude float64, mi
 	ticker := time.NewTicker(updateNodeStatsEvery)
 	go func() {
 		for range ticker.C {
-			start := time.Now()
-			sysusage, err := GetSystemUsage()
-			if err != nil {
-				log.LogNoRequestID("catabalancer failed to get sys usage", "err", err)
+			done := make(chan bool)
+
+			// use a separate goroutine to implement a timeout for the sendMetrics func
+			go func() {
+				sendMetrics(nodeName, latitude, longitude, mist, nodeStatsDB)
+				done <- true
+			}()
+
+			select {
+			case <-done:
+				continue
+			case <-time.After(updateNodeStatsEvery):
+				log.LogNoRequestID("catabalancer send metrics timed out")
 				continue
 			}
+		}
+	}()
+}
 
-			event := NodeUpdateEvent{
-				Resource: "nodeUpdate",
-				NodeID:   nodeName,
-				NodeMetrics: NodeMetrics{
-					CPUUsagePercentage:       sysusage.CPUUsagePercentage,
-					RAMUsagePercentage:       sysusage.RAMUsagePercentage,
-					BandwidthUsagePercentage: sysusage.BWUsagePercentage,
-					LoadAvg:                  sysusage.LoadAvg.Load5Min,
-					GeoLatitude:              latitude,
-					GeoLongitude:             longitude,
-					Timestamp:                time.Now(),
-				},
-			}
+func sendMetrics(nodeName string, latitude float64, longitude float64, mist clients.MistAPIClient, nodeStatsDB *sql.DB) {
+	start := time.Now()
+	sysusage, err := GetSystemUsage()
+	if err != nil {
+		log.LogNoRequestID("catabalancer failed to get sys usage", "err", err)
+		return
+	}
 
-			if mist != nil {
-				mistState, err := mist.GetState()
-				if err != nil {
-					log.LogNoRequestID("catabalancer failed to get mist state", "err", err)
-					continue
-				}
-
-				var nonIngestStreams, ingestStreams []string
-				for streamID := range mistState.ActiveStreams {
-					if mistState.IsIngestStream(streamID) {
-						ingestStreams = append(ingestStreams, streamID)
-					} else {
-						nonIngestStreams = append(nonIngestStreams, streamID)
-					}
-				}
-				event.SetStreams(nonIngestStreams, ingestStreams)
-			}
+	event := NodeUpdateEvent{
+		Resource: "nodeUpdate",
+		NodeID:   nodeName,
+		NodeMetrics: NodeMetrics{
+			CPUUsagePercentage:       sysusage.CPUUsagePercentage,
+			RAMUsagePercentage:       sysusage.RAMUsagePercentage,
+			BandwidthUsagePercentage: sysusage.BWUsagePercentage,
+			LoadAvg:                  sysusage.LoadAvg.Load5Min,
+			GeoLatitude:              latitude,
+			GeoLongitude:             longitude,
+			Timestamp:                time.Now(),
+		},
+	}
 
-			payload, err := json.Marshal(event)
-			if err != nil {
-				log.LogNoRequestID("catabalancer failed to marhsal node update", "err", err)
-				continue
-			}
-			sendMetrics(nodeStatsDB, nodeName, payload)
+	if mist != nil {
+		mistState, err := mist.GetState()
+		if err != nil {
+			log.LogNoRequestID("catabalancer failed to get mist state", "err", err)
+			return
+		}
 
-			metrics.Metrics.CatabalancerSendMetricDurationSec.Observe(time.Since(start).Seconds())
+		var nonIngestStreams, ingestStreams []string
+		for streamID := range mistState.ActiveStreams {
+			if mistState.IsIngestStream(streamID) {
+				ingestStreams = append(ingestStreams, streamID)
+			} else {
+				nonIngestStreams = append(nonIngestStreams, streamID)
+			}
 		}
-	}()
+		event.SetStreams(nonIngestStreams, ingestStreams)
+	}
+
+	payload, err := json.Marshal(event)
+	if err != nil {
+		log.LogNoRequestID("catabalancer failed to marhsal node update", "err", err)
+		return
+	}
+	sendMetricsToDB(nodeStatsDB, nodeName, payload)
+
+	metrics.Metrics.CatabalancerSendMetricDurationSec.Observe(time.Since(start).Seconds())
 }
 
-func sendMetrics(nodeStatsDB *sql.DB, nodeName string, payload []byte) {
+func sendMetricsToDB(nodeStatsDB *sql.DB, nodeName string, payload []byte) {
 	start := time.Now()
 	queryContext, cancel := context.WithTimeout(context.Background(), updateNodeStatsEvery)
 	defer cancel()

balancer/catabalancer/sysstats.go

@@ -108,7 +108,7 @@ func GetBandwidthUsage() (float64, error) {
 
 	// Parse json output
 	var data BandwidthData
-	err = json.Unmarshal([]byte(output), &data)
+	err = json.Unmarshal(output, &data)
 	if err != nil {
 		return -1, err
 	}