Merge pull request #230 from StackVista/STAC-23531

[STAC-23531]: feat: mark k8s watch api conn with a specific label

Author: Andreagit97

checks/net.go

@@ -742,9 +742,10 @@ const (
 	httpResponseTime  metricName = "http_response_time_seconds"
 	httpRequestsDelta metricName = "http_requests_delta"
 
-	httpStatusCodeTag = "code"
-	httpPathTag       = "path"
-	httpMethodTag     = "method"
+	httpStatusCodeTag  = "code"
+	httpPathTag        = "path"
+	httpMethodTag      = "method"
+	httpHighLatencyTag = "known_high_latency"
 
 	mongoResponseTime  metricName = "mongo_response_time_seconds"
 	mongoRequestsDelta metricName = "mongo_requests_delta"
@@ -795,25 +796,31 @@ func (k aggStatsKey) toMap() map[string]string {
 // We need to transform the nanoseconds to seconds here
 const nsToS float64 = 0.000000001
 
-func aggregateStats(stats []http.RequestStat) (int, *ddsketch.DDSketch) {
+func aggregateStats(stats []http.RequestStat) (int, *ddsketch.DDSketch, bool) {
 	requestCount := 0
 	latencies := emptySketch()
+	highLatency := false
 
 	for _, stat := range stats {
-		requestCount += stat.Count
 		if stat.Count == 0 {
 			continue
-		} else if stat.Count == 1 {
+		}
+
+		if stat.IsWatchAPI() {
+			highLatency = true
+		}
+
+		requestCount += stat.Count
+
+		if stat.Count == 1 {
 			latencies.Add(stat.FirstLatencySample * nsToS)
-		} else {
-			if stat.Latencies != nil {
-				var scaled = emptySketch()
-				scaled = stat.Latencies.ChangeMapping(scaled.IndexMapping, scaled.GetPositiveValueStore(), scaled.GetNegativeValueStore(), nsToS)
-				latencies.MergeWith(scaled)
-			}
+		} else if stat.Latencies != nil {
+			var scaled = emptySketch()
+			scaled = stat.Latencies.ChangeMapping(scaled.IndexMapping, scaled.GetPositiveValueStore(), scaled.GetNegativeValueStore(), nsToS)
+			latencies.MergeWith(scaled)
 		}
 	}
-	return requestCount, latencies
+	return requestCount, latencies, highLatency
 }
 
 func aggregateHTTPTraceObservations(httpObservations []http.TransactionObservation) map[connKey][]*model.HTTPTraceObservation {
@@ -1015,6 +1022,8 @@ func aggregateHTTPStats(httpStats map[http.Key]*http.RequestStats, sendForPath b
 		return connStats
 	}
 
+	// httpStats has the following structure: (coming from datadog tracer)
+	// Key: struct{Tuple, Path, Method} -> Value: map[code(100/200/300/...)]struct{Latency, Count, StaticTags}
 	for statKey, statsByCode := range httpStats {
 		for statusCodeClass := uint16(100); statusCodeClass <= 500; statusCodeClass += 100 {
 			stat := statsByCode.Data[statusCodeClass]
@@ -1024,14 +1033,12 @@ func aggregateHTTPStats(httpStats map[http.Key]*http.RequestStats, sendForPath b
 			}
 
 			statusCodeGroup := statusCodeClassToString(statusCodeClass)
-			if statusCodeGroup == "" {
-				continue
-			}
 			connKey := getConnectionKeyForHTTPStats(statKey)
 			connStats := regroupedStats[connKey]
 
 			connStats = appendStatsForStatusGroup(connStats, statusCodeGroup, statKey, stat)
 
+			// Key: struct{Tuple} -> Value: map[struct{Path(optional), Method, StatusCode}][]struct{Latency, Count, StaticTags}
 			regroupedStats[connKey] = connStats
 		}
 	}
@@ -1041,7 +1048,10 @@ func aggregateHTTPStats(httpStats map[http.Key]*http.RequestStats, sendForPath b
 	for connKey, statsByTags := range regroupedStats {
 		for tagsKey, stats := range statsByTags {
 			data := tagsKey.toMap()
-			requestCount, latencies := aggregateStats(stats)
+			requestCount, latencies, highLatency := aggregateStats(stats)
+			if highLatency {
+				data[httpHighLatencyTag] = "true"
+			}
 			result[connKey] = append(result[connKey],
 				makeConnectionMetricWithNumber(
 					httpRequestsDelta, data,

checks/net_test.go

@@ -623,6 +623,54 @@ func msToNs(ms float64) float64 {
 	return ms * 1000000
 }
 
+func TestHTTPAggregation_HighLatency(t *testing.T) {
+	// this is not a watch API
+	conn1 := http.NewKey(
+		util.AddressFromString("10.0.0.1"), util.AddressFromString("192.168.1.1"), 12345, 80,
+		[]byte("/api/v1/namespaces/default/pods?limit=500"), true, http.MethodGet, 0)
+	stats1 := http.NewRequestStats()
+	stats1.AddRequest(200, msToNs(6.0), 0, nil)
+
+	// this is a watch API but on a different dest port, so we have a different conn tuple (source port)
+	conn2 := http.NewKey(
+		util.AddressFromString("10.0.0.1"), util.AddressFromString("192.168.1.1"), 12346, 80,
+		[]byte("/api/v1/namespaces/default/pods?resourceVersion=418&watch=true"), true, http.MethodGet, 0)
+	stats2 := http.NewRequestStats()
+	stats2.AddRequest(200, msToNs(60000.0), http.WatchAPI, nil)
+
+	metrics := aggregateHTTPStats(map[http.Key]*http.RequestStats{
+		conn1: stats1,
+		conn2: stats2,
+	}, false)
+
+	assert.Len(t, metrics, 2)
+
+	// conn1 shouldn't have high latency
+	conn1Metrics := metrics[getConnectionKeyForHTTPStats(conn1)]
+	// for each metric we have 5 status groups (1xx, 2xx, 3xx, 4xx, 5xx)
+	// we have 2 metrics:
+	// - http_requests_delta
+	// - http_response_time_seconds
+	assert.Len(t, conn1Metrics, 10)
+	// all the metrics should have the httpHighLatencyTag
+	for _, metric := range conn1Metrics {
+		assert.NotContains(t, metric.Tags, httpHighLatencyTag)
+	}
+
+	// conn2 should have high latency
+	conn2Metrics := metrics[getConnectionKeyForHTTPStats(conn2)]
+	assert.Len(t, conn2Metrics, 10)
+	for _, metric := range conn2Metrics {
+		// only metrics with 2xx status should have the httpHighLatencyTag
+		// because are the only one with at least a stat with high latency
+		if metric.Tags[httpStatusCodeTag] == "2xx" {
+			assert.Contains(t, metric.Tags, httpHighLatencyTag)
+		} else {
+			assert.NotContains(t, metric.Tags, httpHighLatencyTag)
+		}
+	}
+}
+
 func TestHTTPAggregation_SingleReq(t *testing.T) {
 
 	conn1req1 := http.NewKey(