Simplify CNM baseline scheduling setup

Author: AlexandreYang

comp/networkpath/npcollector/impl/baseline.go

@@ -7,6 +7,7 @@ package npcollectorimpl
 
 import (
 	"iter"
+	"net/netip"
 	"sort"
 
 	"github.com/DataDog/datadog-agent/comp/networkpath/npcollector/impl/common"
@@ -60,17 +61,15 @@ func addBaselineCandidate(selected []baselineCandidate, candidate baselineCandid
 	return selected
 }
 
-func (s *npCollectorImpl) scheduleBaselineNetworkPathTests(conns iter.Seq[npmodel.NetworkPathConnection]) {
-	vpcSubnets, err := s.getVPCSubnets()
-	if err != nil {
-		s.logger.Errorf("Failed to get VPC subnets to skip: %s", err)
-		return
-	}
-
+func (s *npCollectorImpl) scheduleBaselineNetworkPathTests(conns iter.Seq[npmodel.NetworkPathConnection], vpcSubnets []netip.Prefix) {
+	startTime := s.TimeNowFn()
+	connCount := 0
 	selected := make([]baselineCandidate, 0, baselineSelectionsPerSnapshot)
 	for conn := range conns {
+		connCount++
 		evaluation := s.evaluateNetworkPathForConn(conn, payload.PathOriginNetworkTraffic, vpcSubnets)
 		if !evaluation.shouldSchedule {
+			s.logger.Tracef("Skipped connection: addr=%s, protocol=%s", conn.Dest, conn.Type)
 			continue
 		}
 		path := s.makePathtest(conn, payload.PathOriginNetworkTraffic)
@@ -88,4 +87,6 @@ func (s *npCollectorImpl) scheduleBaselineNetworkPathTests(conns iter.Seq[npmode
 			s.logger.Errorf("Error scheduling baseline pathtest: %s", err)
 		}
 	}
+	_ = s.statsdClient.Count(common.NetworkPathCollectorMetricPrefix+"schedule.conns_received", int64(connCount), []string{}, 1)
+	_ = s.statsdClient.Gauge(common.NetworkPathCollectorMetricPrefix+"schedule.duration", s.TimeNowFn().Sub(startTime).Seconds(), nil, 1)
 }

comp/networkpath/npcollector/impl/baseline_test.go

@@ -11,6 +11,7 @@ import (
 	"net/netip"
 	"slices"
 	"testing"
+	"time"
 
 	model "github.com/DataDog/agent-payload/v5/process"
 	"github.com/DataDog/datadog-agent/pkg/networkpath/payload"
@@ -62,6 +63,28 @@ func TestBaselineSelectsCandidatesFromEverySnapshot(t *testing.T) {
 	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,

comp/networkpath/npcollector/impl/npcollector.go

@@ -286,31 +286,31 @@ func (s *npCollectorImpl) getVPCSubnets() ([]netip.Prefix, error) {
 }
 
 func (s *npCollectorImpl) ScheduleNetworkPathTests(conns iter.Seq[npmodel.NetworkPathConnection]) {
+	if !s.collectorConfigs.connectionsMonitoringEnabled && !s.collectorConfigs.baselineTestsEnabled {
+		return
+	}
+
+	vpcSubnets, err := s.getVPCSubnets()
+	if err != nil {
+		s.logger.Errorf("Failed to get VPC subnets to skip: %s", err)
+		return
+	}
+
 	if s.collectorConfigs.connectionsMonitoringEnabled {
-		s.scheduleNetworkPathTests(payload.PathOriginNetworkTraffic, conns)
-	} else if s.collectorConfigs.baselineTestsEnabled {
-		s.scheduleBaselineNetworkPathTests(conns)
+		s.scheduleNetworkPathTests(payload.PathOriginNetworkTraffic, conns, vpcSubnets)
+	} else {
+		s.scheduleBaselineNetworkPathTests(conns, vpcSubnets)
 	}
 }
 
 func (s *npCollectorImpl) ScheduleNetflowPathTests(conns iter.Seq[npmodel.NetworkPathConnection]) {
 	if !s.collectorConfigs.netflowMonitoringEnabled {
 		return
 	}
-	s.scheduleNetworkPathTests(payload.PathOriginNetflow, conns)
+	s.scheduleNetworkPathTests(payload.PathOriginNetflow, conns, nil)
 }
 
-func (s *npCollectorImpl) scheduleNetworkPathTests(origin payload.PathOrigin, conns iter.Seq[npmodel.NetworkPathConnection]) {
-	var vpcSubnets []netip.Prefix
-	if origin == payload.PathOriginNetworkTraffic {
-		var err error
-		vpcSubnets, err = s.getVPCSubnets()
-		if err != nil {
-			s.logger.Errorf("Failed to get VPC subnets to skip: %s", err)
-			return
-		}
-	}
-
+func (s *npCollectorImpl) scheduleNetworkPathTests(origin payload.PathOrigin, conns iter.Seq[npmodel.NetworkPathConnection], vpcSubnets []netip.Prefix) {
 	startTime := s.TimeNowFn()
 	connCount := 0
 	for conn := range conns {