Move baseline ranking derivation to collector

Author: AlexandreYang

comp/networkpath/npcollector/impl/baseline.go

@@ -64,7 +64,7 @@ func addBaselinePath(selected []baselineCandidate, path common.Pathtest, signals
 	return addBaselineCandidate(selected, baselineCandidate{
 		path:       path,
 		pathHash:   path.GetHash(),
-		diagnostic: signals.Diagnostic,
-		bytes:      signals.Bytes,
+		diagnostic: signals.TimeoutCount > 0 || signals.RTOCount > 0 || signals.Retransmits > 0,
+		bytes:      signals.SentBytes + signals.RecvBytes,
 	})
 }

comp/networkpath/npcollector/impl/baseline_test.go

@@ -14,12 +14,12 @@ import (
 	"time"
 
 	model "github.com/DataDog/agent-payload/v5/process"
+	"github.com/DataDog/datadog-agent/comp/networkpath/npcollector/impl/common"
+	npmodel "github.com/DataDog/datadog-agent/comp/networkpath/npcollector/model"
 	"github.com/DataDog/datadog-agent/pkg/networkpath/payload"
 	"github.com/DataDog/datadog-agent/pkg/trace/teststatsd"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
-	npmodel "github.com/DataDog/datadog-agent/comp/networkpath/npcollector/model"
 )
 
 func baselineConn(host string, bytes uint64) npmodel.NetworkPathConnection {
@@ -28,7 +28,7 @@ func baselineConn(host string, bytes uint64) npmodel.NetworkPathConnection {
 		Type:      model.ConnectionType_udp,
 		Direction: model.ConnectionDirection_outgoing,
 		Family:    model.ConnectionFamily_v4,
-		Baseline:  npmodel.BaselineSignals{Bytes: bytes},
+		Baseline:  npmodel.BaselineSignals{SentBytes: bytes},
 	}
 }
 
@@ -93,9 +93,9 @@ func TestBaselinePrioritizesDiagnosticConnections(t *testing.T) {
 
 	healthyLarge := baselineConn("10.0.0.1", 1_000)
 	diagnosticSmall := baselineConn("10.0.0.2", 1)
-	diagnosticSmall.Baseline.Diagnostic = true
+	diagnosticSmall.Baseline.TimeoutCount = 1
 	diagnosticLarge := baselineConn("10.0.0.3", 10)
-	diagnosticLarge.Baseline.Diagnostic = true
+	diagnosticLarge.Baseline.RTOCount = 1
 
 	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{
 		healthyLarge,
@@ -107,6 +107,31 @@ func TestBaselinePrioritizesDiagnosticConnections(t *testing.T) {
 	assert.Equal(t, []string{"10.0.0.3", "10.0.0.2", "10.0.0.1"}, scheduledBaselineHosts(t, collector))
 }
 
+func TestBaselineDiagnosticSignals(t *testing.T) {
+	for name, signals := range map[string]npmodel.BaselineSignals{
+		"timeout":    {TimeoutCount: 1},
+		"rto":        {RTOCount: 1},
+		"retransmit": {Retransmits: 1},
+	} {
+		t.Run(name, func(t *testing.T) {
+			selected := addBaselinePath(nil, common.Pathtest{Hostname: name}, signals)
+
+			require.Len(t, selected, 1)
+			assert.True(t, selected[0].diagnostic)
+		})
+	}
+}
+
+func TestBaselineCombinesSentAndReceivedBytes(t *testing.T) {
+	selected := addBaselinePath(nil, common.Pathtest{Hostname: "host"}, npmodel.BaselineSignals{
+		SentBytes: 5,
+		RecvBytes: 6,
+	})
+
+	require.Len(t, selected, 1)
+	assert.Equal(t, uint64(11), selected[0].bytes)
+}
+
 func TestBaselineKeepsStrongestObservationPerPath(t *testing.T) {
 	_, collector := newTestNpCollector(t, map[string]any{
 		"network_path.connections_monitoring.baseline_tests_enabled": true,

comp/networkpath/npcollector/model/BUILD.bazel

@@ -1,5 +1,4 @@
 load("@rules_go//go:def.bzl", "go_library")
-load("//bazel/rules/go:dd_agent_go_test.bzl", "dd_agent_go_test")
 
 go_library(
     name = "model",
@@ -8,10 +7,3 @@ go_library(
     visibility = ["//visibility:public"],
     deps = ["@com_github_datadog_agent_payload_v5//process"],
 )
-
-dd_agent_go_test(
-    name = "model_test",
-    srcs = ["connection_test.go"],
-    embed = [":model"],
-    deps = ["@com_github_stretchr_testify//assert"],
-)

comp/networkpath/npcollector/model/connection.go

@@ -18,18 +18,11 @@ const TCPTimeoutErrno uint16 = 110
 
 // BaselineSignals contains the normalized CNM values used by baseline selection.
 type BaselineSignals struct {
-	Diagnostic bool
-	Bytes      uint64
-}
-
-// NewBaselineSignals normalizes the CNM deltas used by baseline selection.
-// Both direct and process-agent producers use this constructor so their ranking
-// inputs remain equivalent.
-func NewBaselineSignals(timeoutCount, rtoCount, retransmits, sentBytes, recvBytes uint64) BaselineSignals {
-	return BaselineSignals{
-		Diagnostic: timeoutCount > 0 || rtoCount > 0 || retransmits > 0,
-		Bytes:      sentBytes + recvBytes,
-	}
+	TimeoutCount uint64
+	RTOCount     uint64
+	Retransmits  uint64
+	SentBytes    uint64
+	RecvBytes    uint64
 }
 
 // NetworkPathConnection is the minimum information needed about a connection to schedule a network path test

comp/networkpath/npcollector/model/connection_test.go

@@ -292,13 +292,13 @@ func (d *directSender) networkPathConnections(conns *network.Connections) iter.S
 				Domain:            getDNSNameForIP(conns, conn.Dest),
 				IntraHost:         conn.IntraHost,
 				SystemProbeConn:   conn.Pid == d.sysProbePID,
-				Baseline: npmodel.NewBaselineSignals(
-					uint64(conn.TCPFailures[npmodel.TCPTimeoutErrno]),
-					uint64(conn.Last.TCPRTOCount),
-					uint64(conn.Last.Retransmits),
-					conn.Last.SentBytes,
-					conn.Last.RecvBytes,
-				),
+				Baseline: npmodel.BaselineSignals{
+					TimeoutCount: uint64(conn.TCPFailures[npmodel.TCPTimeoutErrno]),
+					RTOCount:     uint64(conn.Last.TCPRTOCount),
+					Retransmits:  uint64(conn.Last.Retransmits),
+					SentBytes:    conn.Last.SentBytes,
+					RecvBytes:    conn.Last.RecvBytes,
+				},
 			}
 			if !yield(npc) {
 				return

pkg/network/sender/sender_linux.go

@@ -170,8 +170,13 @@ func TestNetworkPathConnectionsBaselineSignals(t *testing.T) {
 	}))
 
 	require.Len(t, got, 1)
-	assert.True(t, got[0].Baseline.Diagnostic)
-	assert.Equal(t, uint64(11), got[0].Baseline.Bytes)
+	assert.Equal(t, npmodel.BaselineSignals{
+		TimeoutCount: 1,
+		RTOCount:     2,
+		Retransmits:  3,
+		SentBytes:    5,
+		RecvBytes:    6,
+	}, got[0].Baseline)
 }
 
 func TestNetworkConnectionBatchingWithDNS(t *testing.T) {

pkg/network/sender/sender_linux_test.go

@@ -255,13 +255,13 @@ func networkPathConnections(conns *model.Connections) iter.Seq[npmodel.NetworkPa
 				Family:            conn.Family,
 				IntraHost:         conn.IntraHost,
 				SystemProbeConn:   conn.SystemProbeConn,
-				Baseline: npmodel.NewBaselineSignals(
-					uint64(conn.TcpFailuresByErrCode[uint32(npmodel.TCPTimeoutErrno)]),
-					uint64(conn.LastTcpRtoCount),
-					uint64(conn.LastRetransmits),
-					conn.LastBytesSent,
-					conn.LastBytesReceived,
-				),
+				Baseline: npmodel.BaselineSignals{
+					TimeoutCount: uint64(conn.TcpFailuresByErrCode[uint32(npmodel.TCPTimeoutErrno)]),
+					RTOCount:     uint64(conn.LastTcpRtoCount),
+					Retransmits:  uint64(conn.LastRetransmits),
+					SentBytes:    conn.LastBytesSent,
+					RecvBytes:    conn.LastBytesReceived,
+				},
 			}
 			if !yield(npc) {
 				return

pkg/process/checks/net.go

@@ -182,8 +182,13 @@ func TestNetworkPathConnectionsBaselineSignals(t *testing.T) {
 	got := slices.Collect(networkPathConnections(&model.Connections{Conns: []*model.Connection{conn}}))
 
 	require.Len(t, got, 1)
-	assert.True(t, got[0].Baseline.Diagnostic)
-	assert.Equal(t, uint64(11), got[0].Baseline.Bytes)
+	assert.Equal(t, npmodel.BaselineSignals{
+		TimeoutCount: 1,
+		RTOCount:     2,
+		Retransmits:  3,
+		SentBytes:    5,
+		RecvBytes:    6,
+	}, got[0].Baseline)
 }
 
 func TestNetworkConnectionBatchingWithDNS(t *testing.T) {