Implement timeout for whole metrics send function (#1400)

* Implement timeout for whole metrics send function

I saw occasional slowness from the mist GetState function. The mist client has a 1 minute timeout and we need a much shorter timeout here as the loop runs every 5 seconds

* Fix build

* Fix build

* Use context.WithTimeout

* Reduce mist client timeout for catabalancer

* review comment: add comment

Author: mjh1

balancer/catabalancer/catalyst_balancer.go

@@ -411,70 +411,75 @@ func (c *CataBalancer) MistUtilLoadSource(ctx context.Context, streamID, lat, lo
 	return "", fmt.Errorf("catabalancer no node found for ingest stream: %s stale: false", streamID)
 }
 
-var updateNodeStatsEvery = 5 * time.Second
+var StatsUpdateInterval = 5 * time.Second
+var StatsUpdateTimeout = StatsUpdateInterval - 500*time.Millisecond // have the timeout sit within the update interval so we don't miss sending updates
 
 func isStale(timestamp time.Time, stale time.Duration) bool {
 	return time.Since(timestamp) >= stale
 }
 
 func StartMetricSending(nodeName string, latitude float64, longitude float64, mist clients.MistAPIClient, nodeStatsDB *sql.DB) {
-	ticker := time.NewTicker(updateNodeStatsEvery)
+	ticker := time.NewTicker(StatsUpdateInterval)
 	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)
-				continue
-			}
+			sendMetrics(nodeName, latitude, longitude, mist, nodeStatsDB)
+		}
+	}()
+}
 
-			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)
+	queryContext, cancel := context.WithTimeout(context.Background(), StatsUpdateTimeout)
 	defer cancel()
 	insertStatement := `insert into "node_stats"(
                             "node_id",

clients/mist_client.go

@@ -50,14 +50,21 @@ type MistClient struct {
 	HttpReqUrl string
 	configMu   sync.Mutex
 	cache      *cache.Cache
+	httpClient *http.Client
 }
 
-func NewMistAPIClient(user, password, host string, port int) MistAPIClient {
+const MistClientTimeout = 1 * time.Minute
+
+func NewMistAPIClient(user, password, host string, port int, timeout time.Duration) MistAPIClient {
+	if timeout == 0 {
+		timeout = MistClientTimeout
+	}
 	mist := &MistClient{
-		ApiUrl:   fmt.Sprintf("http://%s:%d", host, port),
-		Username: user,
-		Password: password,
-		cache:    cache.New(defaultCacheExpiration, cacheCleanupInterval),
+		ApiUrl:     fmt.Sprintf("http://%s:%d", host, port),
+		Username:   user,
+		Password:   password,
+		cache:      cache.New(defaultCacheExpiration, cacheCleanupInterval),
+		httpClient: newRetryableClient(&http.Client{Timeout: timeout}),
 	}
 	return mist
 }
@@ -238,10 +245,6 @@ type MistPushStats struct {
 	Tracks        []int `json:"tracks"`
 }
 
-const MIST_CLIENT_TIMEOUT = 1 * time.Minute
-
-var mistRetryableClient = newRetryableClient(&http.Client{Timeout: MIST_CLIENT_TIMEOUT})
-
 func (mc *MistClient) AddStream(streamName, sourceUrl string) error {
 	c := commandAddStream(streamName, sourceUrl)
 	return wrapErr(validateAddStream(mc.sendCommand(c)), streamName)
@@ -405,7 +408,7 @@ func (mc *MistClient) sendCommandToMist(command interface{}) (string, error) {
 		return "", err
 	}
 	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
-	resp, err := metrics.MonitorRequest(metrics.Metrics.MistClient, mistRetryableClient, req)
+	resp, err := metrics.MonitorRequest(metrics.Metrics.MistClient, mc.httpClient, req)
 	if err != nil {
 		return "", err
 	}
@@ -437,7 +440,7 @@ func (mc *MistClient) sendHttpRequest(streamName string) (string, error) {
 		return "", err
 	}
 	req.Header.Add("Content-Type", "application/json")
-	resp, err := metrics.MonitorRequest(metrics.Metrics.MistClient, mistRetryableClient, req)
+	resp, err := metrics.MonitorRequest(metrics.Metrics.MistClient, mc.httpClient, req)
 	if err != nil {
 		return "", err
 	}

main.go

@@ -227,7 +227,7 @@ func main() {
 	}
 
 	if cli.MistEnabled {
-		mist = clients.NewMistAPIClient(cli.MistUser, cli.MistPassword, cli.MistHost, cli.MistPort)
+		mist = clients.NewMistAPIClient(cli.MistUser, cli.MistPassword, cli.MistHost, cli.MistPort, 0)
 	}
 
 	catabalancerEnabled := balancer.CombinedBalancerEnabled(cli.CataBalancer)
@@ -267,6 +267,7 @@ func main() {
 
 		if catabalancerEnabled && nodeStatsDB != nil {
 			if cli.Tags["node"] == "media" { // don't announce load balancing availability for testing nodes
+				mist := clients.NewMistAPIClient(cli.MistUser, cli.MistPassword, cli.MistHost, cli.MistPort, catabalancer.StatsUpdateTimeout)
 				catabalancer.StartMetricSending(cli.NodeName, cli.NodeLatitude, cli.NodeLongitude, mist, nodeStatsDB)
 			}
 		}