[NETPATH-174] Add npcollector benchmark (#27492)

comp/networkpath/npcollector/npcollectorimpl/npcollector_test.go

@@ -3,13 +3,16 @@
 // This product includes software developed at Datadog (https://www.datadoghq.com/).
 // Copyright 2024-present Datadog, Inc.
 
+// go:build test
+
 package npcollectorimpl
 
 import (
 	"bufio"
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"os"
 	"strings"
 	"testing"
 	"time"
@@ -676,3 +679,42 @@ func Test_npCollectorImpl_sendTelemetry(t *testing.T) {
 	assert.Contains(t, calls, teststatsd.MetricsArgs{Name: "datadog.network_path.check_duration", Value: 3, Tags: tags, Rate: 1})
 	assert.Contains(t, calls, teststatsd.MetricsArgs{Name: "datadog.network_path.check_interval", Value: (2 * time.Minute).Seconds(), Tags: tags, Rate: 1})
 }
+
+func Benchmark_npCollectorImpl_ScheduleConns(b *testing.B) {
+	agentConfigs := map[string]any{
+		"network_path.connections_monitoring.enabled": true,
+		"network_path.collector.workers":              50,
+	}
+
+	file, err := os.OpenFile("benchmark.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
+	assert.Nil(b, err)
+	defer file.Close()
+	w := bufio.NewWriter(file)
+	l, err := seelog.LoggerFromWriterWithMinLevelAndFormat(w, seelog.DebugLvl, "[%LEVEL] %FuncShort: %Msg\n")
+	assert.Nil(b, err)
+	utillog.SetupLogger(l, "debug")
+	defer w.Flush()
+
+	app, npCollector := newTestNpCollector(b, agentConfigs)
+
+	// TEST START
+	app.RequireStart()
+	assert.True(b, npCollector.running)
+
+	// Generate 50 random connections
+	connections := createBenchmarkConns(500, 100)
+
+	b.ResetTimer() // Reset timer after setup
+
+	for i := 0; i < b.N; i++ {
+		// add line to avoid linter error
+		_ = i
+		npCollector.ScheduleConns(connections)
+
+		waitForProcessedPathtests(npCollector, 60*time.Second, 50)
+	}
+
+	// TEST STOP
+	app.RequireStop()
+	assert.False(b, npCollector.running)
+}

comp/networkpath/npcollector/npcollectorimpl/npcollector_testutils.go

@@ -9,6 +9,8 @@ package npcollectorimpl
 
 import (
 	"fmt"
+	"math/rand"
+	"net"
 	"testing"
 	"time"
 
@@ -43,7 +45,7 @@ var testOptions = fx.Options(
 	eventplatformimpl.MockModule(),
 )
 
-func newTestNpCollector(t *testing.T, agentConfigs map[string]any) (*fxtest.App, *npCollectorImpl) {
+func newTestNpCollector(t fxtest.TB, agentConfigs map[string]any) (*fxtest.App, *npCollectorImpl) {
 	var component npcollector.Component
 	app := fxtest.New(t, fx.Options(
 		testOptions,
@@ -70,7 +72,37 @@ func createConns(numberOfConns int) []*model.Connection {
 	return conns
 }
 
-func waitForProcessedPathtests(npCollector *npCollectorImpl, timeout time.Duration, processecCount uint64) {
+func createBenchmarkConns(numberOfConns int, tcpPercent int) []*model.Connection {
+	port := rand.Intn(65535-1) + 1
+	connType := model.ConnectionType_udp
+	if rand.Intn(100) < tcpPercent {
+		connType = model.ConnectionType_tcp
+	}
+	var conns []*model.Connection
+	for i := 0; i < numberOfConns; i++ {
+		conns = append(conns, &model.Connection{
+			Laddr:     &model.Addr{Ip: fmt.Sprintf("127.0.0.%d", i), Port: int32(30000)},
+			Raddr:     &model.Addr{Ip: randomPublicIP(), Port: int32(port)},
+			Direction: model.ConnectionDirection_outgoing,
+			Type:      connType,
+		})
+	}
+	return conns
+}
+
+func randomPublicIP() string {
+	var ip string
+	for {
+		ip = fmt.Sprintf("%d.%d.%d.%d", rand.Intn(256), rand.Intn(256), rand.Intn(256), rand.Intn(256))
+		parsedIP := net.ParseIP(ip)
+		if parsedIP != nil && !parsedIP.IsLoopback() && !parsedIP.IsPrivate() {
+			break
+		}
+	}
+	return ip
+}
+
+func waitForProcessedPathtests(npCollector *npCollectorImpl, timeout time.Duration, processedCount uint64) {
 	timeoutChan := time.After(timeout)
 	tick := time.NewTicker(100 * time.Millisecond)
 	defer tick.Stop()
@@ -79,7 +111,7 @@ func waitForProcessedPathtests(npCollector *npCollectorImpl, timeout time.Durati
 		case <-timeoutChan:
 			return
 		case <-tick.C:
-			if npCollector.processedTracerouteCount.Load() >= processecCount {
+			if npCollector.processedTracerouteCount.Load() >= processedCount {
 				return
 			}
 		}