Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[netpath] Add rate limit to dynamic paths #33841

Merged
merged 5 commits into from
Feb 10, 2025
+302 −90
Merged

[netpath] Add rate limit to dynamic paths #33841

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9387c95
[netpath] Add rate limit circuit breaker to dynamic paths
pimlu Feb 7, 2025
f9cdc43
clamp max time
pimlu Feb 7, 2025
8745ab5
use builtin rate limit package
pimlu Feb 10, 2025
acb0a73
revert tcp processor (whoops)
pimlu Feb 10, 2025
090b05d
better metrics
pimlu Feb 10, 2025
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
23 changes: 12 additions & 11 deletions comp/networkpath/npcollector/npcollectorimpl/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"time"

"github.com/DataDog/datadog-agent/comp/core/config"
"github.com/DataDog/datadog-agent/comp/networkpath/npcollector/npcollectorimpl/pathteststore"
)

type collectorConfigs struct {
Expand All @@ -18,31 +19,31 @@ type collectorConfigs struct {
maxTTL int
pathtestInputChanSize int
pathtestProcessingChanSize int
pathtestContextsLimit int
pathtestTTL time.Duration
pathtestInterval time.Duration
storeConfig pathteststore.Config
flushInterval time.Duration
reverseDNSEnabled bool
reverseDNSTimeout time.Duration
networkDevicesNamespace string
}

func newConfig(agentConfig config.Component) *collectorConfigs {

return &collectorConfigs{
connectionsMonitoringEnabled: agentConfig.GetBool("network_path.connections_monitoring.enabled"),
workers: agentConfig.GetInt("network_path.collector.workers"),
timeout: agentConfig.GetDuration("network_path.collector.timeout") * time.Millisecond,
maxTTL: agentConfig.GetInt("network_path.collector.max_ttl"),
pathtestInputChanSize: agentConfig.GetInt("network_path.collector.input_chan_size"),
pathtestProcessingChanSize: agentConfig.GetInt("network_path.collector.processing_chan_size"),
pathtestContextsLimit: agentConfig.GetInt("network_path.collector.pathtest_contexts_limit"),
pathtestTTL: agentConfig.GetDuration("network_path.collector.pathtest_ttl"),
pathtestInterval: agentConfig.GetDuration("network_path.collector.pathtest_interval"),
flushInterval: agentConfig.GetDuration("network_path.collector.flush_interval"),
reverseDNSEnabled: agentConfig.GetBool("network_path.collector.reverse_dns_enrichment.enabled"),
reverseDNSTimeout: agentConfig.GetDuration("network_path.collector.reverse_dns_enrichment.timeout") * time.Millisecond,
networkDevicesNamespace: agentConfig.GetString("network_devices.namespace"),
storeConfig: pathteststore.Config{
ContextsLimit: agentConfig.GetInt("network_path.collector.pathtest_contexts_limit"),
TTL: agentConfig.GetDuration("network_path.collector.pathtest_ttl"),
Interval: agentConfig.GetDuration("network_path.collector.pathtest_interval"),
MaxPerMinute: agentConfig.GetInt("network_path.collector.pathtest_max_per_minute"),
},
flushInterval: agentConfig.GetDuration("network_path.collector.flush_interval"),
reverseDNSEnabled: agentConfig.GetBool("network_path.collector.reverse_dns_enrichment.enabled"),
reverseDNSTimeout: agentConfig.GetDuration("network_path.collector.reverse_dns_enrichment.timeout") * time.Millisecond,
networkDevicesNamespace: agentConfig.GetString("network_devices.namespace"),
}
}

Expand Down
25 changes: 17 additions & 8 deletions comp/networkpath/npcollector/npcollectorimpl/npcollector.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,9 +91,10 @@ func newNpCollectorImpl(epForwarder eventplatform.Forwarder, collectorConfigs *c
collectorConfigs.maxTTL,
collectorConfigs.pathtestInputChanSize,
collectorConfigs.pathtestProcessingChanSize,
collectorConfigs.pathtestContextsLimit,
collectorConfigs.pathtestTTL,
collectorConfigs.pathtestInterval,
collectorConfigs.storeConfig.ContextsLimit,
collectorConfigs.storeConfig.TTL,
collectorConfigs.storeConfig.Interval,
collectorConfigs.storeConfig.MaxPerMinute,
collectorConfigs.flushInterval,
collectorConfigs.reverseDNSEnabled,
collectorConfigs.reverseDNSTimeout,
Expand All @@ -105,7 +106,8 @@ func newNpCollectorImpl(epForwarder eventplatform.Forwarder, collectorConfigs *c
rdnsquerier: rdnsquerier,
logger: logger,

pathtestStore: pathteststore.NewPathtestStore(collectorConfigs.pathtestTTL, collectorConfigs.pathtestInterval, collectorConfigs.pathtestContextsLimit, logger),
// pathtestStore is set in start() after statsd.Client is configured
pathtestStore: nil,
pathtestInputChan: make(chan *common.Pathtest, collectorConfigs.pathtestInputChanSize),
pathtestProcessingChan: make(chan *pathteststore.PathtestContext, collectorConfigs.pathtestProcessingChanSize),
flushInterval: collectorConfigs.flushInterval,
Expand Down Expand Up @@ -194,6 +196,15 @@ func (s *npCollectorImpl) scheduleOne(pathtest *common.Pathtest) error {
return fmt.Errorf("collector input channel is full (channel capacity is %d)", cap(s.pathtestInputChan))
}
}

func (s *npCollectorImpl) initStatsdClient(statsdClient ddgostatsd.ClientInterface) {
// Assigning statsd.Client in start() stage since we can't do it in newNpCollectorImpl
// due to statsd.Client not being configured yet.
s.statsdClient = statsdClient

s.pathtestStore = pathteststore.NewPathtestStore(s.collectorConfigs.storeConfig, s.logger, statsdClient, s.TimeNowFn)
}

func (s *npCollectorImpl) start() error {
if s.running {
return errors.New("server already started")
Expand All @@ -202,10 +213,8 @@ func (s *npCollectorImpl) start() error {

s.logger.Info("Start NpCollector")

// Assigning statsd.Client in start() stage since we can't do it in newNpCollectorImpl
// due to statsd.Client not being configured yet.
s.metricSender = metricsender.NewMetricSenderStatsd(statsd.Client)
s.statsdClient = statsd.Client
s.initStatsdClient(statsd.Client)
s.metricSender = metricsender.NewMetricSenderStatsd(s.statsdClient)

go s.listenPathtests()
go s.flushLoop()
Expand Down
36 changes: 23 additions & 13 deletions comp/networkpath/npcollector/npcollectorimpl/npcollector_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ func Test_NpCollector_runningAndProcessing(t *testing.T) {

app.RequireStart()

npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.metricSender = metricsender.NewMetricSenderStatsd(stats)

assert.True(t, npCollector.running)
Expand Down Expand Up @@ -255,7 +255,7 @@ func Test_NpCollector_ScheduleConns_ScheduleDurationMetric(t *testing.T) {
_, npCollector := newTestNpCollector(t, agentConfigs)

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.metricSender = metricsender.NewMetricSenderStatsd(stats)

conns := []*model.Connection{
Expand Down Expand Up @@ -304,7 +304,7 @@ func Test_newNpCollectorImpl_defaultConfigs(t *testing.T) {
assert.Equal(t, 4, npCollector.workers)
assert.Equal(t, 1000, cap(npCollector.pathtestInputChan))
assert.Equal(t, 1000, cap(npCollector.pathtestProcessingChan))
assert.Equal(t, 5000, npCollector.collectorConfigs.pathtestContextsLimit)
assert.Equal(t, 5000, npCollector.collectorConfigs.storeConfig.ContextsLimit)
assert.Equal(t, "default", npCollector.networkDevicesNamespace)
}

Expand All @@ -324,7 +324,7 @@ func Test_newNpCollectorImpl_overrideConfigs(t *testing.T) {
assert.Equal(t, 2, npCollector.workers)
assert.Equal(t, 300, cap(npCollector.pathtestInputChan))
assert.Equal(t, 400, cap(npCollector.pathtestProcessingChan))
assert.Equal(t, 500, npCollector.collectorConfigs.pathtestContextsLimit)
assert.Equal(t, 500, npCollector.collectorConfigs.storeConfig.ContextsLimit)
assert.Equal(t, "ns1", npCollector.networkDevicesNamespace)
}

Expand Down Expand Up @@ -514,7 +514,7 @@ func Test_npCollectorImpl_ScheduleConns(t *testing.T) {
utillog.SetupLogger(l, "debug")

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)

npCollector.ScheduleConns(tt.conns, tt.dns)

Expand Down Expand Up @@ -623,7 +623,8 @@ func Test_npCollectorImpl_flushWrapper(t *testing.T) {
_, npCollector := newTestNpCollector(t, agentConfigs)

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)

npCollector.TimeNowFn = func() time.Time {
return tt.flushEndTime
}
Expand All @@ -648,18 +649,27 @@ func Test_npCollectorImpl_flushWrapper(t *testing.T) {
}

func Test_npCollectorImpl_flush(t *testing.T) {
mockNow := time.Now()
mockTimeNow := func() time.Time {
return mockNow
}

// GIVEN
agentConfigs := map[string]any{
"network_path.connections_monitoring.enabled": true,
"network_path.collector.workers": 6,
}
_, npCollector := newTestNpCollector(t, agentConfigs)
npCollector.TimeNowFn = mockTimeNow

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.pathtestStore.Add(&common.Pathtest{Hostname: "host1", Port: 53})
npCollector.pathtestStore.Add(&common.Pathtest{Hostname: "host2", Port: 53})

// simulate some time passing so that the PathTestStore rate limit has some budget to work with
mockNow = mockNow.Add(10 * time.Second)

// WHEN
npCollector.flush()

Expand All @@ -684,7 +694,7 @@ func Test_npCollectorImpl_flushLoop(t *testing.T) {
defer npCollector.stop()

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.pathtestStore.Add(&common.Pathtest{Hostname: "host1", Port: 53})
npCollector.pathtestStore.Add(&common.Pathtest{Hostname: "host2", Port: 53})

Expand Down Expand Up @@ -713,7 +723,7 @@ func Test_npCollectorImpl_sendTelemetry(t *testing.T) {
_, npCollector := newTestNpCollector(t, agentConfigs)

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.metricSender = metricsender.NewMetricSenderStatsd(stats)
path := payload.NetworkPath{
Origin: payload.PathOriginNetworkTraffic,
Expand Down Expand Up @@ -796,7 +806,7 @@ func Test_npCollectorImpl_enrichPathWithRDNS(t *testing.T) {
_, npCollector := newTestNpCollector(t, agentConfigs)

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.metricSender = metricsender.NewMetricSenderStatsd(stats)

// WHEN
Expand Down Expand Up @@ -848,7 +858,7 @@ func Test_npCollectorImpl_enrichPathWithRDNS(t *testing.T) {
}
_, npCollector = newTestNpCollector(t, agentConfigs)

npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.metricSender = metricsender.NewMetricSenderStatsd(stats)

// WHEN
Expand Down Expand Up @@ -881,7 +891,7 @@ func Test_npCollectorImpl_enrichPathWithRDNSKnownHostName(t *testing.T) {
_, npCollector := newTestNpCollector(t, agentConfigs)

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.metricSender = metricsender.NewMetricSenderStatsd(stats)

// WHEN
Expand All @@ -905,7 +915,7 @@ func Test_npCollectorImpl_getReverseDNSResult(t *testing.T) {
_, npCollector := newTestNpCollector(t, agentConfigs)

stats := &teststatsd.Client{}
npCollector.statsdClient = stats
npCollector.initStatsdClient(stats)
npCollector.metricSender = metricsender.NewMetricSenderStatsd(stats)

tts := []struct {
Expand Down
Loading
Loading