Add baseline Dynamic Tests for CNM

comp/networkpath/npcollector/impl/BUILD.bazel

@@ -6,6 +6,7 @@ load("//bazel/rules/go:dd_agent_go_test.bzl", "dd_agent_go_test")
 go_library(
     name = "impl",
     srcs = [
+        "baseline.go",
         "config.go",
         "localips.go",
         "npcollector.go",
@@ -49,6 +50,7 @@ go_library(
 dd_agent_go_test(
     name = "impl_test",
     srcs = [
+        "baseline_test.go",
         "config_test.go",
         "localips_test.go",
         "module_test.go",

comp/networkpath/npcollector/impl/baseline.go

@@ -0,0 +1,70 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026-present Datadog, Inc.
+
+package npcollectorimpl
+
+import (
+	"sort"
+
+	"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"
+)
+
+const baselineSelectionsPerSnapshot = 3
+
+type baselineCandidate struct {
+	path       common.Pathtest
+	pathHash   uint64
+	diagnostic bool
+	bytes      uint64
+}
+
+func (candidate baselineCandidate) betterThan(other baselineCandidate) bool {
+	if candidate.diagnostic != other.diagnostic {
+		return candidate.diagnostic
+	}
+	if candidate.bytes != other.bytes {
+		return candidate.bytes > other.bytes
+	}
+	return candidate.pathHash < other.pathHash
+}
+
+// addBaselineCandidate keeps selected unique, best-first, and capped at three.
+func addBaselineCandidate(selected []baselineCandidate, candidate baselineCandidate) []baselineCandidate {
+	// A discarded observation cannot become a winner unless the same path is
+	// observed later with a stronger score, at which point it is reconsidered.
+	for i := range selected {
+		if selected[i].pathHash != candidate.pathHash {
+			continue
+		}
+		if !candidate.betterThan(selected[i]) {
+			return selected
+		}
+		selected[i] = candidate
+		sort.Slice(selected, func(i, j int) bool { return selected[i].betterThan(selected[j]) })
+		return selected
+	}
+
+	if len(selected) < baselineSelectionsPerSnapshot {
+		selected = append(selected, candidate)
+	} else if candidate.betterThan(selected[len(selected)-1]) {
+		selected[len(selected)-1] = candidate
+	} else {
+		return selected
+	}
+	sort.Slice(selected, func(i, j int) bool { return selected[i].betterThan(selected[j]) })
+	return selected
+}
+
+func addBaselinePath(selected []baselineCandidate, path common.Pathtest, signals npmodel.ConnectionSignals) []baselineCandidate {
+	path.DynamicTestProfile = payload.DynamicTestProfileBaseline
+	return addBaselineCandidate(selected, baselineCandidate{
+		path:       path,
+		pathHash:   path.GetHash(),
+		diagnostic: signals.TimeoutCount > 0 || signals.RTOCount > 0 || signals.Retransmits > 0,
+		bytes:      signals.SentBytes + signals.RecvBytes,
+	})
+}

comp/networkpath/npcollector/impl/baseline_test.go

@@ -0,0 +1,222 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026-present Datadog, Inc.
+
+//go:build test
+
+package npcollectorimpl
+
+import (
+	"net/netip"
+	"slices"
+	"testing"
+	"time"
+
+	model "github.com/DataDog/agent-payload/v5/process"
+	"github.com/DataDog/datadog-agent/comp/networkpath/npcollector/impl/common"
+	"github.com/DataDog/datadog-agent/comp/networkpath/npcollector/impl/connfilter"
+	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"
+)
+
+func baselineConn(host string, bytes uint64) npmodel.NetworkPathConnection {
+	return npmodel.NetworkPathConnection{
+		Dest:      netip.MustParseAddrPort(host + ":53"),
+		Type:      model.ConnectionType_udp,
+		Direction: model.ConnectionDirection_outgoing,
+		Family:    model.ConnectionFamily_v4,
+		Signals:   npmodel.ConnectionSignals{SentBytes: bytes},
+	}
+}
+
+func scheduledBaselineHosts(t *testing.T, collector *npCollectorImpl) []string {
+	t.Helper()
+	hosts := make([]string, 0, len(collector.pathtestInputChan))
+	for len(collector.pathtestInputChan) > 0 {
+		pathtest := <-collector.pathtestInputChan
+		assert.Equal(t, payload.DynamicTestProfileBaseline, pathtest.DynamicTestProfile)
+		hosts = append(hosts, pathtest.Hostname)
+	}
+	return hosts
+}
+
+func TestBaselineSelectsCandidatesFromEverySnapshot(t *testing.T) {
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests_enabled": true,
+		"network_path.collector.monitor_ip_without_domain":           true,
+	}, &teststatsd.Client{}, nil)
+
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{
+		baselineConn("10.0.0.1", 1),
+		baselineConn("10.0.0.2", 3),
+		baselineConn("10.0.0.3", 2),
+		baselineConn("10.0.0.4", 4),
+	}))
+
+	assert.Equal(t, []string{"10.0.0.4", "10.0.0.2", "10.0.0.3"}, scheduledBaselineHosts(t, collector))
+
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{baselineConn("10.0.1.1", 10)}))
+	require.Len(t, collector.pathtestInputChan, 1)
+	assert.Equal(t, "10.0.1.1", (<-collector.pathtestInputChan).Hostname)
+}
+
+func TestBaselineReportsSnapshotTelemetry(t *testing.T) {
+	stats := &teststatsd.Client{}
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests_enabled": true,
+		"network_path.collector.monitor_ip_without_domain":           true,
+	}, stats, nil)
+	timeNowCounter := 0
+	collector.TimeNowFn = func() time.Time {
+		now := MockTimeNow().Add(time.Duration(timeNowCounter) * time.Minute)
+		timeNowCounter++
+		return now
+	}
+
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{
+		baselineConn("10.0.0.1", 2),
+		baselineConn("10.0.0.2", 1),
+	}))
+
+	assert.Equal(t, int64(2), stats.GetCountSummaries()["datadog.network_path.collector.schedule.conns_received"].Sum)
+	assert.Equal(t, 60.0, stats.GetGaugeSummaries()["datadog.network_path.collector.schedule.duration"].Last)
+}
+
+func TestBaselinePrioritizesDiagnosticConnections(t *testing.T) {
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests_enabled": true,
+		"network_path.collector.monitor_ip_without_domain":           true,
+	}, &teststatsd.Client{}, nil)
+
+	healthyLarge := baselineConn("10.0.0.1", 1_000)
+	diagnosticSmall := baselineConn("10.0.0.2", 1)
+	diagnosticSmall.Signals.TimeoutCount = 1
+	diagnosticLarge := baselineConn("10.0.0.3", 10)
+	diagnosticLarge.Signals.RTOCount = 1
+
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{
+		healthyLarge,
+		diagnosticSmall,
+		diagnosticLarge,
+		baselineConn("10.0.0.4", 500),
+	}))
+
+	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.ConnectionSignals{
+		"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.ConnectionSignals{
+		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,
+		"network_path.collector.monitor_ip_without_domain":           true,
+	}, &teststatsd.Client{}, nil)
+
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{
+		baselineConn("10.0.0.1", 1),
+		baselineConn("10.0.0.2", 90),
+		baselineConn("10.0.0.1", 100),
+		baselineConn("10.0.0.3", 80),
+		baselineConn("10.0.0.4", 70),
+	}))
+
+	assert.Equal(t, []string{"10.0.0.1", "10.0.0.2", "10.0.0.3"}, scheduledBaselineHosts(t, collector))
+}
+
+func TestBaselinePreservesWinningRCProvenance(t *testing.T) {
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests_enabled": true,
+		"network_path.collector.monitor_ip_without_domain":           true,
+	}, &teststatsd.Client{}, nil)
+	filter, errs := connfilter.NewConnFilter([]connfilter.Config{
+		{Type: connfilter.FilterTypeExclude, MatchIP: "10.0.0.1"},
+		{
+			Type:         connfilter.FilterTypeInclude,
+			MatchIP:      "10.0.0.1",
+			TestConfigID: "dynamic-a",
+			Tags:         []string{"team:payments"},
+		},
+	}, "", false)
+	require.Empty(t, errs)
+	collector.filter = filter
+
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{
+		baselineConn("10.0.0.1", 1),
+	}))
+
+	pathtest := <-collector.pathtestInputChan
+	assert.Equal(t, payload.DynamicTestProfileBaseline, pathtest.DynamicTestProfile)
+	assert.Equal(t, "dynamic-a", pathtest.TestConfigID)
+	assert.Equal(t, payload.TestConfigSourceRemote, pathtest.TestConfigSource)
+	assert.Equal(t, []string{"team:payments"}, pathtest.Tags)
+}
+
+func TestBaselineEmptySnapshotsSelectNothing(t *testing.T) {
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests_enabled": true,
+		"network_path.collector.monitor_ip_without_domain":           true,
+	}, &teststatsd.Client{}, nil)
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{{
+		Dest:      netip.MustParseAddrPort("10.0.0.1:53"),
+		Direction: model.ConnectionDirection_incoming,
+		Family:    model.ConnectionFamily_v4,
+	}}))
+
+	assert.Empty(t, collector.pathtestInputChan)
+}
+
+func TestBaselineDisabledCreatesNoCollectorMachinery(t *testing.T) {
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests_enabled": false,
+		"network_path.collector.monitor_ip_without_domain":           true,
+	}, &teststatsd.Client{}, nil)
+
+	assert.False(t, collector.collectorConfigs.networkPathCollectorEnabled())
+	assert.Nil(t, collector.pathtestInputChan)
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{baselineConn("10.0.0.1", 1)}))
+}
+
+func TestBaselineSelectionConsumesSlotsBeforeChannelAdmission(t *testing.T) {
+	stats := &teststatsd.Client{}
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests_enabled": true,
+		"network_path.collector.monitor_ip_without_domain":           true,
+		"network_path.collector.input_chan_size":                     1,
+	}, stats, nil)
+
+	collector.ScheduleNetworkPathTests(slices.Values([]npmodel.NetworkPathConnection{
+		baselineConn("10.0.0.1", 3),
+		baselineConn("10.0.0.2", 2),
+		baselineConn("10.0.0.3", 1),
+	}))
+
+	assert.Len(t, collector.pathtestInputChan, 1)
+	assert.Equal(t, int64(2), stats.GetCountSummaries()["datadog.network_path.collector.schedule.pathtest_dropped"].Sum)
+}

comp/networkpath/npcollector/impl/common/pathtest.go

@@ -26,16 +26,17 @@ type PathtestMetadata struct {
 
 // Pathtest details of information necessary to run a traceroute
 type Pathtest struct {
-	Hostname          string
-	Port              uint16
-	Protocol          payload.Protocol
-	SourceContainerID string
-	Namespace         string
-	Origin            payload.PathOrigin
-	TestConfigID      string
-	TestConfigSource  payload.TestConfigSource
-	Tags              []string
-	Metadata          PathtestMetadata
+	Hostname           string
+	Port               uint16
+	Protocol           payload.Protocol
+	SourceContainerID  string
+	Namespace          string
+	Origin             payload.PathOrigin
+	TestConfigID       string
+	TestConfigSource   payload.TestConfigSource
+	DynamicTestProfile payload.DynamicTestProfile
+	Tags               []string
+	Metadata           PathtestMetadata
 }
 
 // GetHash returns the hash of the Pathtest

comp/networkpath/npcollector/impl/config.go

@@ -18,6 +18,7 @@ import (
 
 type collectorConfigs struct {
 	connectionsMonitoringEnabled    bool
+	baselineTestsEnabled            bool
 	netflowMonitoringEnabled        bool
 	workers                         int
 	timeout                         time.Duration
@@ -54,6 +55,7 @@ func newConfig(agentConfig config.Component, logger log.Component) *collectorCon
 	}
 	return &collectorConfigs{
 		connectionsMonitoringEnabled: agentConfig.GetBool("network_path.connections_monitoring.enabled"),
+		baselineTestsEnabled:         agentConfig.GetBool("network_path.connections_monitoring.baseline_tests_enabled"),
 		netflowMonitoringEnabled:     agentConfig.GetBool("network_path.netflow_monitoring.enabled"),
 		workers:                      agentConfig.GetInt("network_path.collector.workers"),
 		timeout:                      agentConfig.GetDuration("network_path.collector.timeout") * time.Millisecond,
@@ -91,5 +93,5 @@ func newConfig(agentConfig config.Component, logger log.Component) *collectorCon
 // networkPathCollectorEnabled checks if Network Path Collector should be enabled
 // Network Path Collector is expected to be enabled if a feature depend on it.
 func (c *collectorConfigs) networkPathCollectorEnabled() bool {
-	return c.connectionsMonitoringEnabled || c.netflowMonitoringEnabled
+	return c.connectionsMonitoringEnabled || c.baselineTestsEnabled || c.netflowMonitoringEnabled
 }

comp/networkpath/npcollector/impl/config_test.go

@@ -30,6 +30,10 @@ func TestNetworkPathCollectorEnabled(t *testing.T) {
 	config.connectionsMonitoringEnabled = false
 	assert.False(t, config.networkPathCollectorEnabled())
 
+	config.baselineTestsEnabled = true
+	assert.True(t, config.networkPathCollectorEnabled())
+
+	config.baselineTestsEnabled = false
 	config.netflowMonitoringEnabled = true
 	assert.True(t, config.networkPathCollectorEnabled())
 }
@@ -47,6 +51,7 @@ func TestNewConfig(t *testing.T) {
 			},
 			expectedConfig: &collectorConfigs{
 				connectionsMonitoringEnabled: false,
+				baselineTestsEnabled:         false,
 				netflowMonitoringEnabled:     false,
 				workers:                      4,
 				timeout:                      1000 * time.Millisecond,
@@ -120,6 +125,7 @@ func TestNewConfig(t *testing.T) {
 			},
 			expectedConfig: &collectorConfigs{
 				connectionsMonitoringEnabled: false,
+				baselineTestsEnabled:         false,
 				netflowMonitoringEnabled:     false,
 				workers:                      8,
 				timeout:                      5000 * time.Millisecond,