Skip to content
Draft
Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
55 commits
Select commit Hold shift + click to select a range
778ceec
empty
AlexandreYang Aug 15, 2026
cbab65c
Add baseline CNM dynamic tests
AlexandreYang Aug 15, 2026
c5e261b
Address baseline dynamic test review
AlexandreYang Aug 15, 2026
47147ad
Harden baseline dynamic test wiring
AlexandreYang Aug 15, 2026
8022154
[review-fix 3] Refresh generated Bazel metadata
AlexandreYang Aug 15, 2026
412f580
[review-fix 4] Satisfy Network Path lint
AlexandreYang Aug 15, 2026
757b8ce
Focus baseline Dynamic Tests E2E coverage
AlexandreYang Aug 15, 2026
235cc7a
[review-fix 1] Preserve traceroute defaults and hide baseline flag
AlexandreYang Aug 15, 2026
9aac167
Split host traffic Dynamic Tests suites
AlexandreYang Aug 15, 2026
25ed7aa
[review-fix 1] Clarify effective Dynamic Tests state
AlexandreYang Aug 15, 2026
f093369
[review-fix 1] Preserve baseline selections across windows
AlexandreYang Aug 15, 2026
46cde90
[review-fix 2] Retry one-shots under backpressure
AlexandreYang Aug 15, 2026
3cb688c
[review-fix 3] Clarify platform and E2E assumptions
AlexandreYang Aug 15, 2026
99403ac
[review-fix 4] Publish config and remove external E2E pull
AlexandreYang Aug 15, 2026
ec8e923
[review-fix 5] Keep baseline rollout setting internal
AlexandreYang Aug 15, 2026
239d15e
Use recurring scheduling for baseline path tests
AlexandreYang Aug 15, 2026
9a43c6d
Simplify baseline dynamic test configuration
AlexandreYang Aug 15, 2026
acff989
Simplify baseline selector design
AlexandreYang Aug 16, 2026
b1d70ae
Remove unnecessary baseline saturation handling
AlexandreYang Aug 16, 2026
8d3c2b3
Simplify baseline Dynamic Tests activation
AlexandreYang Aug 16, 2026
c1cd703
Require explicit traceroute for baseline tests
AlexandreYang Aug 16, 2026
1dc2cfb
Simplify baseline Dynamic Tests activation
AlexandreYang Aug 16, 2026
d749ade
Extract baseline selector package
AlexandreYang Aug 16, 2026
86e2828
Reuse path test hash in baseline selector
AlexandreYang Aug 16, 2026
0e3b8fd
Rename collector baseline tests
AlexandreYang Aug 16, 2026
fb3f6e1
Simplify baseline Dynamic Tests telemetry
AlexandreYang Aug 16, 2026
282cde5
Simplify baseline ranking signals
AlexandreYang Aug 16, 2026
610aa4d
Simplify baseline path selection
AlexandreYang Aug 16, 2026
2d1d8c7
Simplify baseline candidate selection
AlexandreYang Aug 16, 2026
a98bf96
Organize baseline collector logic
AlexandreYang Aug 16, 2026
34fde89
Group baseline connection signals
AlexandreYang Aug 16, 2026
c9d8cc3
Revert unrelated event forwarding refactor
AlexandreYang Aug 16, 2026
e9898c4
Simplify CNM baseline release note
AlexandreYang Aug 16, 2026
136f119
Simplify CNM baseline scheduling setup
AlexandreYang Aug 16, 2026
d3c3e68
Unify network path test scheduling
AlexandreYang Aug 16, 2026
ca027f0
Restore network path subnet lookup
AlexandreYang Aug 16, 2026
c5ec567
Separate baseline path scheduling
AlexandreYang Aug 16, 2026
ab2d7ab
Revert "Separate baseline path scheduling"
AlexandreYang Aug 16, 2026
79f5be4
Clarify network path scheduling modes
AlexandreYang Aug 16, 2026
aecda3c
Clarify standard path scheduling helper
AlexandreYang Aug 16, 2026
5461d8c
Group network path scheduling helpers
AlexandreYang Aug 16, 2026
8325dd8
Explain baseline candidate selection
AlexandreYang Aug 16, 2026
c650133
Move baseline ranking derivation to collector
AlexandreYang Aug 16, 2026
8525a1e
Clarify baseline scheduling mode
AlexandreYang Aug 16, 2026
b225e69
Generalize network path connection signals
AlexandreYang Aug 16, 2026
adfc8be
[review-fix 1] Preserve baseline RC provenance
AlexandreYang Aug 16, 2026
9ac5b16
[review-fix 2] Keep RC filters standard-only
AlexandreYang Aug 16, 2026
8afc6af
Revert "[review-fix 2] Keep RC filters standard-only"
AlexandreYang Aug 16, 2026
db7b106
[review-fix 1] Isolate baseline filters from Dynamic RC
AlexandreYang Aug 16, 2026
97ee4b7
Revert "[review-fix 1] Isolate baseline filters from Dynamic RC"
AlexandreYang Aug 16, 2026
6c21596
Clarify baseline and Dynamic RC compatibility
AlexandreYang Aug 16, 2026
00e11c5
Place baseline RC contract at scheduling boundary
AlexandreYang Aug 16, 2026
dc92f11
Document RC provenance at baseline selection
AlexandreYang Aug 16, 2026
1304876
[review-fix 1] Restore resolver symlinks verbatim
AlexandreYang Aug 16, 2026
3f6a287
[review-fix 2] Fall back from stale resolver links
AlexandreYang Aug 16, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions comp/metadata/inventoryagent/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@ The payload is a JSON dict with the following fields
`network_path.remote_config.enabled` config option).
- `feature_network_path_connections_monitoring_enabled` - **bool**: True if connections monitoring is enabled for Network Path (see:
`network_path.connections_monitoring.enabled` config option).
- `feature_network_path_baseline_tests_enabled` - **bool**: True if included baseline Network Path Dynamic Tests are configured (see:
`network_path.connections_monitoring.baseline_tests.enabled` config option).
- `feature_traceroute_enabled` - **bool**: True if the Traceroute module is enabled in the System Probe (see: `traceroute.enabled` config option in `system-probe.yaml`).
- `feature_oom_kill_enabled` - **bool**: True if the OOM Kill check is enabled for System Probe (see: `system_probe_config.enable_oom_kill` config option in `system-probe.yaml`).
- `feature_synthetics_collector_enabled` - **bool**: True if Synthetics Test module is enabled.
Expand Down Expand Up @@ -168,6 +170,7 @@ Here an example of an inventory payload:
"feature_networks_enabled": false,
"feature_network_path_remote_config_enabled": false,
"feature_network_path_connections_monitoring_enabled": false,
"feature_network_path_baseline_tests_enabled": false,
"feature_traceroute_enabled": false,
"feature_synthetics_collector_enabled": false,
"feature_process_enabled": false,
Expand Down
1 change: 1 addition & 0 deletions comp/metadata/inventoryagent/impl/inventoryagent.go
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,7 @@ func (ia *inventoryagent) fetchCoreAgentMetadata() {
ia.data["feature_container_images_enabled"] = ia.conf.GetBool("container_image.enabled")
ia.data["feature_network_path_remote_config_enabled"] = ia.conf.GetBool("network_path.remote_config.enabled")
ia.data["feature_network_path_connections_monitoring_enabled"] = ia.conf.GetBool("network_path.connections_monitoring.enabled")
ia.data["feature_network_path_baseline_tests_enabled"] = ia.conf.GetBool("network_path.connections_monitoring.baseline_tests.enabled")

ia.data["feature_csm_vm_containers_enabled"] = ia.conf.GetBool("sbom.enabled") && ia.conf.GetBool("container_image.enabled") && ia.conf.GetBool("sbom.container_image.enabled")
ia.data["feature_csm_vm_hosts_enabled"] = ia.conf.GetBool("sbom.enabled") && ia.conf.GetBool("sbom.host.enabled")
Expand Down
36 changes: 19 additions & 17 deletions comp/metadata/inventoryagent/impl/inventoryagent_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -193,23 +193,24 @@ func TestInitData(t *testing.T) {
"eks_fargate": true,
"synthetics.collector.enabled": true,

"fips.enabled": true,
"logs_enabled": true,
"compliance_config.enabled": true,
"compliance_config.host_benchmarks.enabled": true,
"apm_config.enabled": true,
"ec2_prefer_imdsv2": true,
"process_config.container_collection.enabled": true,
"remote_configuration.enabled": true,
"remote_updates": true,
"process_config.process_collection.enabled": true,
"container_image.enabled": true,
"network_path.remote_config.enabled": true,
"network_path.connections_monitoring.enabled": true,
"sbom.enabled": true,
"sbom.container_image.enabled": true,
"sbom.host.enabled": true,
"infrastructure_mode": "basic",
"fips.enabled": true,
"logs_enabled": true,
"compliance_config.enabled": true,
"compliance_config.host_benchmarks.enabled": true,
"apm_config.enabled": true,
"ec2_prefer_imdsv2": true,
"process_config.container_collection.enabled": true,
"remote_configuration.enabled": true,
"remote_updates": true,
"process_config.process_collection.enabled": true,
"container_image.enabled": true,
"network_path.remote_config.enabled": true,
"network_path.connections_monitoring.enabled": true,
"network_path.connections_monitoring.baseline_tests.enabled": true,
"sbom.enabled": true,
"sbom.container_image.enabled": true,
"sbom.host.enabled": true,
"infrastructure_mode": "basic",
}
ia := getTestInventoryPayload(t, overrides, sysprobeOverrides)
ia.refreshMetadata()
Expand Down Expand Up @@ -244,6 +245,7 @@ func TestInitData(t *testing.T) {
"feature_container_images_enabled": true,
"feature_network_path_remote_config_enabled": true,
"feature_network_path_connections_monitoring_enabled": true,
"feature_network_path_baseline_tests_enabled": true,

"feature_dynamic_instrumentation_enabled": true,
"feature_cws_enabled": true,
Expand Down
8 changes: 8 additions & 0 deletions comp/networkpath/npcollector/impl/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ load("//bazel/rules/go:dd_agent_go_test.bzl", "dd_agent_go_test")
go_library(
name = "impl",
srcs = [
"baseline_selector.go",
"config.go",
"localips.go",
"npcollector.go",
Expand All @@ -19,6 +20,7 @@ go_library(
deps = [
"//comp/core/config",
"//comp/core/log/def",
"//comp/core/sysprobeconfig/def",
"//comp/def",
"//comp/forwarder/eventplatform/def",
"//comp/networkpath/npcollector/def",
Expand All @@ -34,11 +36,13 @@ go_library(
"//pkg/config/structure",
"//pkg/logs/message",
"//pkg/network/tracer/networkfilter",
"//pkg/networkpath/config",
"//pkg/networkpath/payload",
"//pkg/networkpath/traceroute/config",
"//pkg/remoteconfig/state",
"//pkg/util/cloudproviders/network",
"//pkg/util/log",
"@com_github_cespare_xxhash_v2//:xxhash",
"@com_github_datadog_agent_payload_v5//process",
"@com_github_datadog_datadog_go_v5//statsd",
"@com_github_patrickmn_go_cache//:go-cache",
Expand All @@ -49,6 +53,8 @@ go_library(
dd_agent_go_test(
name = "impl_test",
srcs = [
"baseline_selector_test.go",
"baseline_window_test.go",
"config_test.go",
"localips_test.go",
"module_test.go",
Expand All @@ -65,6 +71,8 @@ dd_agent_go_test(
"//comp/core/hostname/hostnameimpl",
"//comp/core/log/def",
"//comp/core/log/mock",
"//comp/core/sysprobeconfig/def",
"//comp/core/sysprobeconfig/mock",
"//comp/core/telemetry/mock",
"//comp/forwarder/defaultforwarder/mock",
"//comp/forwarder/eventplatform/def",
Expand Down
195 changes: 195 additions & 0 deletions comp/networkpath/npcollector/impl/baseline_selector.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,195 @@
// 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 (
"encoding/binary"
"math"
"sort"

"github.com/cespare/xxhash/v2"

"github.com/DataDog/datadog-agent/comp/networkpath/npcollector/impl/common"
npmodel "github.com/DataDog/datadog-agent/comp/networkpath/npcollector/model"
)

const (
baselineDiagnosticCandidates = 128
baselineHealthyCandidates = 128
baselineSelectionsPerWindow = 3
)

type baselineCandidate struct {
pathtest common.Pathtest
hash uint64
timeout bool
count uint64
rttVar uint64
}

type baselinePool struct {
capacity int
items map[uint64]*baselineCandidate
}

func newBaselinePool(capacity int) baselinePool {
return baselinePool{capacity: capacity, items: make(map[uint64]*baselineCandidate, capacity)}
}

// baselinePathtestHash is deliberately local to the bounded selector. The
// selector hashes every eligible connection, so use the Agent's existing
// allocation-efficient xxhash dependency instead of the store's general hash.
func baselinePathtestHash(digest *xxhash.Digest, p common.Pathtest) uint64 {
digest.Reset()
writeBaselineHashString(digest, string(p.Origin))
writeBaselineHashString(digest, p.Namespace)
writeBaselineHashString(digest, p.Hostname)
var port [2]byte
binary.LittleEndian.PutUint16(port[:], p.Port)
_, _ = digest.Write(port[:])
writeBaselineHashString(digest, string(p.Protocol))
writeBaselineHashString(digest, p.SourceContainerID)
return digest.Sum64()
}

func writeBaselineHashString(digest *xxhash.Digest, value string) {
var length [8]byte
binary.LittleEndian.PutUint64(length[:], uint64(len(value)))
_, _ = digest.Write(length[:])
_, _ = digest.WriteString(value)
}

func saturatingAdd(a, b uint64) (uint64, bool) {
if math.MaxUint64-a < b {
return math.MaxUint64, true
}
return a + b, false
Comment thread
AlexandreYang marked this conversation as resolved.
Outdated
}

func (p *baselinePool) remove(hash uint64) { delete(p.items, hash) }

func (p *baselinePool) weakest(diagnostic bool) *baselineCandidate {
var weakest *baselineCandidate
for _, candidate := range p.items {
if weakest == nil || baselineBetter(weakest, candidate, diagnostic) {
weakest = candidate
}
}
return weakest
}

func (p *baselinePool) add(hash uint64, pathtest common.Pathtest, timeout bool, weight, rttVar uint64, diagnostic bool) (replaced, discarded, saturated bool) {
if candidate, found := p.items[hash]; found {
candidate.timeout = candidate.timeout || timeout
candidate.count, saturated = saturatingAdd(candidate.count, weight)
candidate.rttVar = max(candidate.rttVar, rttVar)
return false, false, saturated
}

if len(p.items) < p.capacity {
p.items[hash] = &baselineCandidate{pathtest: pathtest, hash: hash, timeout: timeout, count: weight, rttVar: rttVar}
return false, false, false
}

weakest := p.weakest(diagnostic)
// Timeout/RTO is the primary diagnostic class. Do not let a non-timeout
// candidate evict one when the diagnostic pool contains only timeouts.
if diagnostic && weakest.timeout && !timeout {
return false, true, false
}
delete(p.items, weakest.hash)
estimate, overflow := saturatingAdd(weakest.count, weight)
// Reuse the evicted entry. High-cardinality snapshots should not allocate a
// candidate object for every connection that passes through a bounded pool.
*weakest = baselineCandidate{
pathtest: pathtest,
hash: hash,
timeout: timeout,
count: estimate,
rttVar: rttVar,
}
p.items[hash] = weakest
return true, false, overflow
}

func baselineBetter(a, b *baselineCandidate, diagnostic bool) bool {
if diagnostic && a.timeout != b.timeout {
return a.timeout
}
if a.count != b.count {
return a.count > b.count
}
if diagnostic && a.rttVar != b.rttVar {
return a.rttVar > b.rttVar
}
return a.hash < b.hash
}

func (p *baselinePool) sorted(diagnostic bool) []*baselineCandidate {
result := make([]*baselineCandidate, 0, len(p.items))
for _, candidate := range p.items {
result = append(result, candidate)
}
sort.Slice(result, func(i, j int) bool { return baselineBetter(result[i], result[j], diagnostic) })
return result
}

type baselineSelector struct {
diagnostic baselinePool
healthy baselinePool
hashDigest xxhash.Digest
}

func newBaselineSelector() *baselineSelector {
return &baselineSelector{
diagnostic: newBaselinePool(baselineDiagnosticCandidates),
healthy: newBaselinePool(baselineHealthyCandidates),
}
}

type baselineAdmission struct {
replaced bool
discarded bool
saturated bool
}

func (s *baselineSelector) add(pathtest common.Pathtest, conn npmodel.NetworkPathConnection) baselineAdmission {
hash := baselinePathtestHash(&s.hashDigest, pathtest)
diagnostic := conn.TCPTimeout || conn.TCPRTO || conn.Retransmits > 0
if diagnostic {
s.healthy.remove(hash)
replaced, discarded, saturated := s.diagnostic.add(hash, pathtest, conn.TCPTimeout || conn.TCPRTO, conn.Retransmits, conn.RTTVar, true)
return baselineAdmission{replaced: replaced, discarded: discarded, saturated: saturated || conn.NumericSaturated}
}
if _, found := s.diagnostic.items[hash]; found {
_, _, saturated := s.diagnostic.add(hash, pathtest, false, 0, conn.RTTVar, true)
return baselineAdmission{saturated: saturated || conn.NumericSaturated}
}
replaced, discarded, saturated := s.healthy.add(hash, pathtest, false, conn.Bytes, 0, false)
return baselineAdmission{replaced: replaced, discarded: discarded, saturated: saturated || conn.NumericSaturated}
}

func (s *baselineSelector) selectPathtests() []common.Pathtest {
selected := make([]common.Pathtest, 0, baselineSelectionsPerWindow)
for _, candidate := range s.diagnostic.sorted(true) {
selected = append(selected, candidate.pathtest)
if len(selected) == baselineSelectionsPerWindow {
return selected
}
}
for _, candidate := range s.healthy.sorted(false) {
selected = append(selected, candidate.pathtest)
if len(selected) == baselineSelectionsPerWindow {
break
}
}
return selected
}

func (s *baselineSelector) reset() {
s.diagnostic = newBaselinePool(baselineDiagnosticCandidates)
s.healthy = newBaselinePool(baselineHealthyCandidates)
}
Loading
Loading