Skip to content

Commit 6e49f81

Browse files
kakkoyunkakkoyun
authored
feat(tracer): periodically poll agent /info endpoint for dynamic capability updates (#4451)
### What does this PR do? The tracer currently queries the Datadog Agent's `/info` endpoint **once at startup**. Any agent-side capability changes (peer-tag policies, span events support, `client_drop_p0s`, etc.) require a full tracer restart to take effect. This PR adds periodic polling of `/info` every 5 seconds so the tracer picks up agent config changes dynamically, without a restart. **Thread-safety design:** `agentFeatures` is wrapped in `atomicAgentFeatures` backed by `atomic.Pointer[agentFeatures]`, giving lock-free snapshot reads on the hot path. A CAS-loop `update()` method prevents lost-update races if two writers ever race. All reads were updated from `c.agent.X` to `c.agent.load().X`. **Static vs dynamic field split:** Fields baked into components at init (transport URL, statsd address, AppSec config, concentrator aggregation key, obfuscator feature flags) are preserved across every poll. Only fields safe to update at runtime (DropP0s, Stats, peerTags, spanEventsAvailable, obfuscationVersion) are refreshed. ### Motivation - Jira: https://datadoghq.atlassian.net/browse/APMLP-860 - Operators changing agent configuration (e.g. enabling peer tags, adjusting stats collection) currently require bouncing all tracers. With periodic polling, changes propagate within 5 seconds. ### Reviewer's Checklist - [x] Changed code has unit tests for its functionality at or near 100% coverage. - `TestRefreshAgentFeaturesPreservesStaticFields` — static fields frozen, dynamic fields updated - `TestPollAgentInfoUpdatesFeaturesDynamically` — convergence verified via `assert.Eventually` - `TestPollAgentInfoRetainsLastKnownGoodOnError` — network failure preserves last known good - `TestPollAgentInfoRetainsLastKnownGoodOn404` — 404 (agent downgrade) preserves last known good - `TestPollAgentInfoGoroutineStopsOnTracerStop` — no goroutine leak on `Stop()` - [ ] [System-Tests](https://github.com/DataDog/system-tests/) covering this feature have been added and enabled with the va.b.c-dev version tag. - [ ] There is a benchmark for any new code, or changes to existing code. - [x] If this interacts with the agent in a new way, a system test has been added. *(polls existing `/info` endpoint — no new agent interaction)* - [x] New code is free of linting errors. You can check this by running `make lint` locally. - [x] New code doesn't break existing tests. You can check this by running `make test` locally. - [ ] Add an appropriate team label so this PR gets put in the right place for the release notes. - [ ] All generated files are up to date. You can check this by running `make generate` locally. - [ ] Non-trivial go.mod changes, e.g. adding new modules, are reviewed by @DataDog/dd-trace-go-guild. Make sure all nested modules are up to date by running `make fix-modules` locally. *(no go.mod changes)* Co-authored-by: kemal.akkoyun <[email protected]>
1 parent e9cc077 commit 6e49f81

10 files changed

Lines changed: 521 additions & 79 deletions

File tree

‎ddtrace/tracer/log.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,7 @@ func logStartup(t *tracer) {
149149
Architecture: runtime.GOARCH,
150150
GlobalService: globalconfig.ServiceName(),
151151
LambdaMode: fmt.Sprintf("%t", t.config.internalConfig.LogToStdout()),
152-
AgentFeatures: t.config.agent,
152+
AgentFeatures: t.config.agent.load(),
153153
Integrations: t.config.integrations,
154154
AppSec: appsec.Enabled(),
155155
PartialFlushEnabled: partialFlushEnabled,

‎ddtrace/tracer/option.go‎

Lines changed: 102 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"runtime/debug"
2424
"strconv"
2525
"strings"
26+
"sync/atomic"
2627
"time"
2728

2829
"golang.org/x/mod/semver"
@@ -146,8 +147,13 @@ type config struct {
146147
appsecStartOptions []appsecconfig.StartOption
147148

148149
// agent holds the capabilities of the agent and determines some
149-
// of the behaviour of the tracer.
150-
agent agentFeatures
150+
// of the behaviour of the tracer. It is updated atomically so that
151+
// periodic polling can refresh it without locking the hot path.
152+
agent atomicAgentFeatures
153+
154+
// agentInfoPollInterval overrides the default polling interval for /info.
155+
// A zero value uses defaultAgentInfoPollInterval.
156+
agentInfoPollInterval time.Duration
151157

152158
// integrations reports if the user has instrumented a Datadog integration and
153159
// if they have a version of the library available to integrate.
@@ -465,8 +471,9 @@ func newConfig(opts ...StartOption) (*config, error) {
465471

466472
// if using stdout or traces are disabled or we are in ci visibility agentless mode, agent is disabled
467473
agentDisabled := c.internalConfig.LogToStdout() || !c.enabled.get() || c.ciVisibilityAgentless
468-
c.agent = loadAgentFeatures(agentDisabled, c.agentURL, c.httpClient)
469-
if c.agent.v1ProtocolAvailable {
474+
af := loadAgentFeatures(agentDisabled, c.agentURL, c.httpClient)
475+
c.agent.store(af)
476+
if af.v1ProtocolAvailable {
470477
c.traceProtocol = traceProtocolV1
471478
if t, ok := c.transport.(*httpTransport); ok {
472479
t.traceURL = fmt.Sprintf("%s%s", c.agentURL.String(), tracesAPIPathV1)
@@ -483,7 +490,7 @@ func newConfig(opts ...StartOption) (*config, error) {
483490
}
484491
if c.statsdClient == nil {
485492
// configure statsd client
486-
addr := resolveDogstatsdAddr(c)
493+
addr := resolveDogstatsdAddr(c.dogstatsdAddr, af)
487494
globalconfig.SetDogstatsdAddr(addr)
488495
c.dogstatsdAddr = addr
489496
}
@@ -507,7 +514,7 @@ func newConfig(opts ...StartOption) (*config, error) {
507514
Site: env.Get("DD_SITE"),
508515
}
509516
c.llmobs.AgentFeatures = llmobsconfig.AgentFeatures{
510-
EVPProxyV2: c.agent.evpProxyV2,
517+
EVPProxyV2: af.evpProxyV2,
511518
}
512519

513520
return c, nil
@@ -540,14 +547,13 @@ func apmTracingDisabled(c *config) {
540547
// address and the agent-reported port. If the agent reports a port, it will be used
541548
// instead of the user-defined address' port. UDS paths are honored regardless of the
542549
// agent-reported port.
543-
func resolveDogstatsdAddr(c *config) string {
544-
addr := c.dogstatsdAddr
550+
func resolveDogstatsdAddr(addr string, af agentFeatures) string {
545551
if addr == "" {
546552
// no config defined address; use host and port from env vars
547553
// or default to localhost:8125 if not set
548554
addr = defaultDogstatsdAddr()
549555
}
550-
agentport := c.agent.StatsdPort
556+
agentport := af.StatsdPort
551557
if agentport == 0 {
552558
// the agent didn't report a port; use the already resolved address as
553559
// features are loaded from the trace-agent, which might be not running
@@ -665,23 +671,72 @@ func (a *agentFeatures) HasFlag(feat string) bool {
665671
return ok
666672
}
667673

668-
// loadAgentFeatures queries the trace-agent for its capabilities and updates
669-
// the tracer's behaviour.
670-
func loadAgentFeatures(agentDisabled bool, agentURL *url.URL, httpClient *http.Client) (features agentFeatures) {
671-
if agentDisabled {
672-
// there is no agent; all features off
673-
return
674+
// atomicAgentFeatures wraps agentFeatures in an atomic pointer for lock-free
675+
// reads on the hot path. All fields update atomically as a single consistent
676+
// snapshot from one /info response.
677+
type atomicAgentFeatures struct {
678+
val atomic.Pointer[agentFeatures]
679+
}
680+
681+
func (a *atomicAgentFeatures) load() agentFeatures {
682+
if p := a.val.Load(); p != nil {
683+
return *p
684+
}
685+
return agentFeatures{}
686+
}
687+
688+
func (a *atomicAgentFeatures) store(f agentFeatures) {
689+
a.val.Store(&f)
690+
}
691+
692+
// update atomically reads the current snapshot, calls fn with it, and stores
693+
// the result. fn may be called more than once if a concurrent store races the
694+
// CAS; it must therefore be a pure transform with no observable side-effects.
695+
func (a *atomicAgentFeatures) update(fn func(agentFeatures) agentFeatures) {
696+
for {
697+
old := a.val.Load()
698+
var cur agentFeatures
699+
if old != nil {
700+
cur = *old
701+
}
702+
next := fn(cur)
703+
if a.val.CompareAndSwap(old, &next) {
704+
return
705+
}
674706
}
675-
resp, err := httpClient.Get(fmt.Sprintf("%s/info", agentURL))
707+
}
708+
709+
// errAgentFeaturesNotSupported is returned by fetchAgentFeatures when the
710+
// agent does not expose the /info endpoint (pre-7.28.0 agents). Callers
711+
// should treat this as "no capabilities known" at startup, or "keep the
712+
// previous snapshot" during a poll.
713+
var errAgentFeaturesNotSupported = errors.New("agent does not support /info")
714+
715+
// fetchAgentFeatures queries the trace-agent's /info endpoint and parses the
716+
// response into an agentFeatures value. It returns an error if the request
717+
// fails or the response cannot be decoded; the caller should retain the
718+
// previous snapshot in that case. A 404 response returns
719+
// errAgentFeaturesNotSupported so callers can distinguish it from a real
720+
// network failure. The request is bound to ctx so callers can cancel it on
721+
// shutdown.
722+
func fetchAgentFeatures(ctx context.Context, agentURL *url.URL, httpClient *http.Client) (agentFeatures, error) {
723+
req, err := http.NewRequestWithContext(ctx, http.MethodGet, agentURL.JoinPath("info").String(), nil)
676724
if err != nil {
677-
log.Error("Loading features: %s", err.Error())
678-
return
725+
return agentFeatures{}, fmt.Errorf("creating /info request: %w", err)
679726
}
680-
if resp.StatusCode == http.StatusNotFound {
681-
// agent is older than 7.28.0, features not discoverable
682-
return
727+
resp, err := httpClient.Do(req)
728+
if err != nil {
729+
return agentFeatures{}, fmt.Errorf("loading features: %w", err)
683730
}
684731
defer resp.Body.Close()
732+
if resp.StatusCode == http.StatusNotFound {
733+
// agent is older than 7.28.0; /info not available
734+
io.Copy(io.Discard, resp.Body) //nolint:errcheck // best-effort drain for connection reuse
735+
return agentFeatures{}, errAgentFeaturesNotSupported
736+
}
737+
if resp.StatusCode != http.StatusOK {
738+
return agentFeatures{}, fmt.Errorf("unexpected /info status: %d", resp.StatusCode)
739+
}
685740
type infoResponse struct {
686741
Endpoints []string `json:"endpoints"`
687742
ClientDropP0s bool `json:"client_drop_p0s"`
@@ -695,13 +750,11 @@ func loadAgentFeatures(agentDisabled bool, agentURL *url.URL, httpClient *http.C
695750
DefaultEnv string `json:"default_env"`
696751
} `json:"config"`
697752
}
698-
699753
var info infoResponse
700-
if err := json.NewDecoder(resp.Body).Decode(&info); err != nil {
701-
log.Error("Decoding features: %s", err.Error())
702-
return
754+
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&info); err != nil {
755+
return agentFeatures{}, fmt.Errorf("decoding features: %w", err)
703756
}
704-
757+
var features agentFeatures
705758
features.DropP0s = info.ClientDropP0s
706759
features.StatsdPort = info.Config.StatsdPort
707760
features.defaultEnv = info.Config.DefaultEnv
@@ -730,9 +783,30 @@ func loadAgentFeatures(agentDisabled bool, agentURL *url.URL, httpClient *http.C
730783
features.featureFlags[flag] = struct{}{}
731784
}
732785
features.reachable = true
786+
return features, nil
787+
}
788+
789+
// loadAgentFeatures queries the trace-agent for its capabilities at startup and
790+
// stores the result. It handles the agentDisabled case and logs errors.
791+
func loadAgentFeatures(agentDisabled bool, agentURL *url.URL, httpClient *http.Client) agentFeatures {
792+
if agentDisabled {
793+
// there is no agent; all features off
794+
return agentFeatures{}
795+
}
796+
features, err := fetchAgentFeatures(context.Background(), agentURL, httpClient)
797+
if err != nil && !errors.Is(err, errAgentFeaturesNotSupported) {
798+
log.Error("%s", err.Error())
799+
}
733800
return features
734801
}
735802

803+
// agentEnabled reports whether the tracer should communicate with the agent.
804+
// The agent is considered disabled in serverless (LogToStdout), when the
805+
// tracer itself is disabled, or in CI visibility agentless mode.
806+
func (c *config) agentEnabled() bool {
807+
return !c.internalConfig.LogToStdout() && c.enabled.get() && !c.ciVisibilityAgentless
808+
}
809+
736810
// MarkIntegrationImported labels the given integration as imported
737811
func MarkIntegrationImported(integration string) bool {
738812
s, ok := contribIntegrations[integration]
@@ -770,7 +844,8 @@ func (c *config) loadContribIntegrations(deps []*debug.Module) {
770844
// - Trace Agent exposes 'stats' endpoint
771845
// - Stats Computation is enabled on the tracer (or has 'discovery' FF)
772846
func (c *config) canComputeStats() bool {
773-
return c.agent.Stats && c.agent.DropP0s && (c.internalConfig.HasFeature("discovery") || c.internalConfig.StatsComputationEnabled())
847+
a := c.agent.load()
848+
return a.Stats && a.DropP0s && (c.internalConfig.HasFeature("discovery") || c.internalConfig.StatsComputationEnabled())
774849
}
775850

776851
// canDropP0s determines whether P0 spans can be dropped.

‎ddtrace/tracer/option_test.go‎

Lines changed: 24 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -202,13 +202,13 @@ func TestLoadAgentFeatures(t *testing.T) {
202202
t.Run("disabled", func(t *testing.T) {
203203
cfg, err := newTestConfig(WithLambdaMode(true), WithAgentTimeout(2))
204204
assert.NoError(t, err)
205-
assert.Zero(t, cfg.agent)
205+
assert.Zero(t, cfg.agent.load())
206206
})
207207

208208
t.Run("unreachable", func(t *testing.T) {
209209
cfg, err := newTestConfig(WithAgentAddr("127.0.0.1:0"))
210210
assert.NoError(t, err)
211-
assert.Zero(t, cfg.agent)
211+
assert.Zero(t, cfg.agent.load())
212212
})
213213

214214
t.Run("StatusNotFound", func(t *testing.T) {
@@ -218,7 +218,7 @@ func TestLoadAgentFeatures(t *testing.T) {
218218
defer srv.Close()
219219
cfg, err := newTestConfig(WithAgentAddr(strings.TrimPrefix(srv.URL, "http://")), WithAgentTimeout(2))
220220
require.NoError(t, err)
221-
assert.Zero(t, cfg.agent)
221+
assert.Zero(t, cfg.agent.load())
222222
})
223223

224224
t.Run("error", func(t *testing.T) {
@@ -228,7 +228,7 @@ func TestLoadAgentFeatures(t *testing.T) {
228228
defer srv.Close()
229229
cfg, err := newTestConfig(WithAgentAddr(strings.TrimPrefix(srv.URL, "http://")), WithAgentTimeout(2))
230230
require.NoError(t, err)
231-
assert.Zero(t, cfg.agent)
231+
assert.Zero(t, cfg.agent.load())
232232
})
233233
})
234234

@@ -239,19 +239,20 @@ func TestLoadAgentFeatures(t *testing.T) {
239239
defer srv.Close()
240240
cfg, err := newTestConfig(WithAgentAddr(strings.TrimPrefix(srv.URL, "http://")), WithAgentTimeout(2))
241241
assert.NoError(t, err)
242-
assert.True(t, cfg.agent.DropP0s)
243-
assert.Equal(t, cfg.agent.StatsdPort, 8999)
244-
assert.EqualValues(t, cfg.agent.featureFlags, map[string]struct{}{
242+
a := cfg.agent.load()
243+
assert.True(t, a.DropP0s)
244+
assert.Equal(t, a.StatsdPort, 8999)
245+
assert.EqualValues(t, a.featureFlags, map[string]struct{}{
245246
"a": {},
246247
"b": {},
247248
})
248-
assert.True(t, cfg.agent.Stats)
249-
assert.True(t, cfg.agent.HasFlag("a"))
250-
assert.True(t, cfg.agent.HasFlag("b"))
251-
assert.EqualValues(t, cfg.agent.peerTags, []string{"peer.hostname"})
252-
assert.Equal(t, 2, cfg.agent.obfuscationVersion)
253-
assert.False(t, cfg.agent.hasTelemetryProxy)
254-
assert.True(t, cfg.agent.reachable)
249+
assert.True(t, a.Stats)
250+
assert.True(t, a.HasFlag("a"))
251+
assert.True(t, a.HasFlag("b"))
252+
assert.EqualValues(t, a.peerTags, []string{"peer.hostname"})
253+
assert.Equal(t, 2, a.obfuscationVersion)
254+
assert.False(t, a.hasTelemetryProxy)
255+
assert.True(t, a.reachable)
255256
})
256257

257258
t.Run("telemetry_proxy", func(t *testing.T) {
@@ -261,9 +262,10 @@ func TestLoadAgentFeatures(t *testing.T) {
261262
defer srv.Close()
262263
cfg, err := newTestConfig(WithAgentAddr(strings.TrimPrefix(srv.URL, "http://")), WithAgentTimeout(2))
263264
assert.NoError(t, err)
264-
assert.True(t, cfg.agent.Stats)
265-
assert.True(t, cfg.agent.hasTelemetryProxy)
266-
assert.True(t, cfg.agent.reachable)
265+
a := cfg.agent.load()
266+
assert.True(t, a.Stats)
267+
assert.True(t, a.hasTelemetryProxy)
268+
assert.True(t, a.reachable)
267269
})
268270

269271
t.Run("default_env", func(t *testing.T) {
@@ -273,7 +275,7 @@ func TestLoadAgentFeatures(t *testing.T) {
273275
defer srv.Close()
274276
cfg, err := newConfig(WithAgentAddr(strings.TrimPrefix(srv.URL, "http://")), WithAgentTimeout(2))
275277
assert.NoError(t, err)
276-
assert.Equal(t, "prod", cfg.agent.defaultEnv)
278+
assert.Equal(t, "prod", cfg.agent.load().defaultEnv)
277279
})
278280

279281
t.Run("discovery", func(t *testing.T) {
@@ -284,9 +286,10 @@ func TestLoadAgentFeatures(t *testing.T) {
284286
defer srv.Close()
285287
cfg, err := newTestConfig(WithAgentAddr(strings.TrimPrefix(srv.URL, "http://")), WithAgentTimeout(2))
286288
assert.NoError(t, err)
287-
assert.True(t, cfg.agent.DropP0s)
288-
assert.True(t, cfg.agent.Stats)
289-
assert.Equal(t, 8999, cfg.agent.StatsdPort)
289+
a := cfg.agent.load()
290+
assert.True(t, a.DropP0s)
291+
assert.True(t, a.Stats)
292+
assert.Equal(t, 8999, a.StatsdPort)
290293
})
291294
}
292295

0 commit comments

Comments
 (0)