Skip to content

Commit b5626d3

Browse files
authored
Merge branch 'main' into config-revamp/peer-service
2 parents 610583c + 3ded665 commit b5626d3

5 files changed

Lines changed: 146 additions & 36 deletions

File tree

.gitlab/bp-runner.fail-on-breach.yml

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
# Generated: 2026-03-30
33
#
44
# Generation Strategy: tight
5-
# Formula: CI_bound / (1 ± T) (T = 5.0%)
5+
# Formula: CI_bound / (1 ± T) (T = 15.0%)
66
#
77
# SLO Checking:
88
# - BREACH: 90% CI boundary crosses threshold
@@ -20,26 +20,26 @@ experiments:
2020
# Go oldstable
2121
- name: normal_operation_cgo-cpu-bound/go-oldstable-profile-trace-asm
2222
thresholds:
23-
- agg_http_req_duration_p99 < 1.194 ms
23+
- agg_http_req_duration_p99 < 1.5 ms
2424
- name: high_load_cgo-cpu-bound/go-oldstable-profile-trace-asm
2525
thresholds:
26-
- throughput > 65.587 op/s
26+
- throughput > 60 op/s
2727
- name: normal_operation_cgo-cpu-bound/go-oldstable-only-trace
2828
thresholds:
29-
- agg_http_req_duration_p99 < 125.0 ms
29+
- agg_http_req_duration_p99 < 1.5 ms
3030
- name: high_load_cgo-cpu-bound/go-oldstable-only-trace
3131
thresholds:
32-
- throughput > 65.753 op/s
32+
- throughput > 60 op/s
3333
# Go stable
3434
- name: normal_operation_cgo-cpu-bound/go-stable-profile-trace-asm
3535
thresholds:
36-
- agg_http_req_duration_p99 < 1.194 ms
36+
- agg_http_req_duration_p99 < 1.5 ms
3737
- name: high_load_cgo-cpu-bound/go-stable-profile-trace-asm
3838
thresholds:
39-
- throughput > 65.775 op/s
39+
- throughput > 60 op/s
4040
- name: normal_operation_cgo-cpu-bound/go-stable-only-trace
4141
thresholds:
42-
- agg_http_req_duration_p99 < 125.0 ms
42+
- agg_http_req_duration_p99 < 1.5 ms
4343
- name: high_load_cgo-cpu-bound/go-stable-only-trace
4444
thresholds:
45-
- throughput > 65.743 op/s
45+
- throughput > 60 op/s

internal/telemetry/client.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,6 @@ func newClient(tracerConfig internal.TracerConfig, config ClientConfig) (*client
4949
tracerConfig: tracerConfig,
5050
writer: writer,
5151
clientConfig: config,
52-
flushMapper: mapper.NewDefaultMapper(config.HeartbeatInterval, config.ExtendedHeartbeatInterval),
5352
payloadQueue: internal.NewRingQueue[transport.Payload](config.PayloadQueueSize),
5453

5554
dependencies: dependencies{
@@ -69,6 +68,8 @@ func newClient(tracerConfig internal.TracerConfig, config ClientConfig) (*client
6968
backend: newLoggerBackend(config.MaxDistinctLogs),
7069
}
7170

71+
client.flushMapper = mapper.NewDefaultMapper(config.HeartbeatInterval, config.ExtendedHeartbeatInterval, client.configuration.All)
72+
7273
client.dataSources = append(client.dataSources,
7374
&client.integrations,
7475
&client.products,

internal/telemetry/client_test.go

Lines changed: 67 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,7 +179,73 @@ func TestClientFlush(t *testing.T) {
179179
assert.Equal(t, transport.RequestTypeAppClientConfigurationChange, batch[0].RequestType)
180180
assert.Equal(t, transport.RequestTypeAppExtendedHeartBeat, batch[1].RequestType)
181181

182-
assert.Len(t, batch[1].Payload.(transport.AppExtendedHeartbeat).Configuration, 0)
182+
extHB := batch[1].Payload.(transport.AppExtendedHeartbeat)
183+
require.Len(t, extHB.Configuration, 1)
184+
assert.Equal(t, "key", extHB.Configuration[0].Name)
185+
assert.Equal(t, "value", extHB.Configuration[0].Value)
186+
},
187+
},
188+
{
189+
name: "extended-heartbeat-config-multiple",
190+
clientConfig: ClientConfig{
191+
ExtendedHeartbeatInterval: time.Nanosecond,
192+
},
193+
when: func(c *client) {
194+
c.RegisterAppConfigs(
195+
Configuration{Name: "key1", Value: "value1", Origin: OriginDefault},
196+
Configuration{Name: "key2", Value: "value2", Origin: OriginEnvVar},
197+
)
198+
199+
time.Sleep(time.Microsecond)
200+
},
201+
expect: func(t *testing.T, payloads []transport.Payload) {
202+
payload := payloads[0]
203+
require.IsType(t, transport.MessageBatch{}, payload)
204+
batch := payload.(transport.MessageBatch)
205+
require.Len(t, batch, 2)
206+
assert.Equal(t, transport.RequestTypeAppClientConfigurationChange, batch[0].RequestType)
207+
assert.Equal(t, transport.RequestTypeAppExtendedHeartBeat, batch[1].RequestType)
208+
209+
extHB := batch[1].Payload.(transport.AppExtendedHeartbeat)
210+
require.Len(t, extHB.Configuration, 2)
211+
configMap := make(map[string]transport.ConfKeyValue)
212+
for _, c := range extHB.Configuration {
213+
configMap[c.Name] = c
214+
}
215+
assert.Equal(t, "value1", configMap["key1"].Value)
216+
assert.Equal(t, "value2", configMap["key2"].Value)
217+
},
218+
},
219+
{
220+
name: "extended-heartbeat-config-dedup",
221+
clientConfig: ClientConfig{
222+
ExtendedHeartbeatInterval: time.Nanosecond,
223+
},
224+
when: func(c *client) {
225+
c.RegisterAppConfigs(
226+
Configuration{Name: "key1", Value: "original", Origin: OriginDefault},
227+
)
228+
c.RegisterAppConfigs(
229+
Configuration{Name: "key1", Value: "updated", Origin: OriginDefault},
230+
)
231+
232+
time.Sleep(time.Microsecond)
233+
},
234+
expect: func(t *testing.T, payloads []transport.Payload) {
235+
payload := payloads[0]
236+
require.IsType(t, transport.MessageBatch{}, payload)
237+
batch := payload.(transport.MessageBatch)
238+
239+
var extHB transport.AppExtendedHeartbeat
240+
for _, msg := range batch {
241+
if msg.RequestType == transport.RequestTypeAppExtendedHeartBeat {
242+
extHB = msg.Payload.(transport.AppExtendedHeartbeat)
243+
}
244+
}
245+
246+
require.Len(t, extHB.Configuration, 1)
247+
assert.Equal(t, "key1", extHB.Configuration[0].Name)
248+
assert.Equal(t, "updated", extHB.Configuration[0].Value)
183249
},
184250
},
185251
{

internal/telemetry/configuration.go

Lines changed: 61 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,33 @@ import (
1717
"github.com/DataDog/dd-trace-go/v2/internal/telemetry/internal/transport"
1818
)
1919

20+
// configuration is a data source that tracks SDK configuration key-value pairs
21+
// (e.g. DD_ENV, DD_SERVICE) reported by products via RegisterAppConfig/RegisterAppConfigs.
22+
//
23+
// Flow:
24+
// - Products call Add() to register configs. Each config is stored in config and
25+
// its key is marked in pending.
26+
// - On each flush tick (~60s), the client calls Payload() which returns an
27+
// AppClientConfigurationChange containing only new/updated configs since the
28+
// last flush, then clears pending. The config map itself is never cleared,
29+
// so it accumulates the full state from app-started through any subsequent
30+
// config changes.
31+
// - When heartbeatEnricher emits an extended heartbeat (~24h), it calls All()
32+
// which returns the full accumulated config state from config. Since Add()
33+
// overwrites previous values for the same key, All() always reflects the
34+
// correct current state — merging startup configs with any subsequent updates.
35+
// This also ensures startup configs consumed by appStartedReducer (and
36+
// otherwise invisible to the mapper pipeline) are included.
37+
// - Both Payload() and All() normalize configs via normalize() which applies
38+
// default origin, sanitizes values, and assigns fallback seqIDs.
2039
type configuration struct {
21-
mu sync.Mutex
40+
mu sync.Mutex
41+
// config holds all registered configs for the lifetime of the SDK. Entries are
42+
// never removed; updated configs overwrite previous values for the same key.
2243
config map[configKey]transport.ConfKeyValue
44+
// pending tracks which keys in config have been added or updated since the last
45+
// Payload() call. Cleared after each flush so only deltas are reported.
46+
pending map[configKey]struct{}
2347
// fallbackSeqID is used only for legacy configs that don't already have a seqID.
2448
// New code should report configs with seqIDs via config/configProvider.
2549
fallbackSeqID uint64
@@ -43,6 +67,7 @@ func (c *configuration) Add(kv Configuration) {
4367

4468
if c.config == nil {
4569
c.config = make(map[configKey]transport.ConfKeyValue)
70+
c.pending = make(map[configKey]struct{})
4671
}
4772

4873
ID := idOrEmpty(kv.ID)
@@ -55,39 +80,56 @@ func (c *configuration) Add(kv Configuration) {
5580
ID: ID,
5681
SeqID: kv.SeqID,
5782
}
83+
c.pending[key] = struct{}{}
84+
}
85+
86+
// normalize applies default origin, sanitizes the value, and assigns a fallback
87+
// seqID if needed. The normalized conf is written back to c.config[key].
88+
func (c *configuration) normalize(key configKey) transport.ConfKeyValue {
89+
conf := c.config[key]
90+
if conf.Origin == "" {
91+
conf.Origin = transport.OriginDefault
92+
}
93+
conf.Value = SanitizeConfigValue(conf.Value)
94+
if conf.SeqID == 0 {
95+
c.fallbackSeqID++
96+
conf.SeqID = c.fallbackSeqID
97+
}
98+
c.config[key] = conf
99+
return conf
58100
}
59101

60102
func (c *configuration) Payload() transport.Payload {
61103
c.mu.Lock()
62104
defer c.mu.Unlock()
63-
if len(c.config) == 0 {
105+
if len(c.pending) == 0 {
64106
return nil
65107
}
66108

67-
configs := make([]transport.ConfKeyValue, len(c.config))
68-
idx := 0
69-
for key, conf := range c.config {
70-
if conf.Origin == "" {
71-
conf.Origin = transport.OriginDefault
72-
}
73-
conf.Value = SanitizeConfigValue(conf.Value)
74-
75-
// Fallback seqID for legacy code that doesn't report via config/configProvider
76-
if conf.SeqID == 0 {
77-
c.fallbackSeqID++
78-
conf.SeqID = c.fallbackSeqID
79-
}
80-
81-
configs[idx] = conf
82-
idx++
83-
delete(c.config, key)
109+
configs := make([]transport.ConfKeyValue, 0, len(c.pending))
110+
for key := range c.pending {
111+
configs = append(configs, c.normalize(key))
84112
}
113+
clear(c.pending)
85114

86115
return transport.AppClientConfigurationChange{
87116
Configuration: configs,
88117
}
89118
}
90119

120+
// All returns a sanitized snapshot of all accumulated configs. Used by
121+
// heartbeatEnricher to populate the configuration field in extended heartbeats.
122+
func (c *configuration) All() []transport.ConfKeyValue {
123+
c.mu.Lock()
124+
defer c.mu.Unlock()
125+
126+
configs := make([]transport.ConfKeyValue, 0, len(c.config))
127+
for key := range c.config {
128+
configs = append(configs, c.normalize(key))
129+
}
130+
return configs
131+
}
132+
91133
// SanitizeConfigValue sanitizes the value of a configuration key to ensure it can be marshalled.
92134
func SanitizeConfigValue(value any) any {
93135
if value == nil {

internal/telemetry/internal/mapper/default.go

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,12 +14,12 @@ import (
1414
)
1515

1616
// NewDefaultMapper returns a Mapper that transforms payloads into a MessageBatch and adds a heartbeat message.
17-
// The heartbeat message is added every heartbeatInterval.
18-
func NewDefaultMapper(heartbeatInterval, extendedHeartBeatInterval time.Duration) Mapper {
17+
func NewDefaultMapper(heartbeatInterval, extendedHeartBeatInterval time.Duration, getConfigs func() []transport.ConfKeyValue) Mapper {
1918
mapper := &defaultMapper{
2019
heartbeatEnricher: heartbeatEnricher{
2120
heartbeatRL: rate.NewLimiter(rate.Every(heartbeatInterval), 1),
2221
extendedHeartbeatRL: rate.NewLimiter(rate.Every(extendedHeartBeatInterval), 1),
22+
getConfigs: getConfigs,
2323
},
2424
}
2525

@@ -62,19 +62,17 @@ type heartbeatEnricher struct {
6262
heartbeatRL *rate.Limiter
6363
extendedHeartbeatRL *rate.Limiter
6464

65+
getConfigs func() []transport.ConfKeyValue
66+
6567
extendedHeartbeat transport.AppExtendedHeartbeat
6668
heartbeat transport.AppHeartbeat
6769
}
6870

6971
func (t *heartbeatEnricher) Transform(payloads []transport.Payload) ([]transport.Payload, Mapper) {
70-
// Built the extended heartbeat using other payloads
7172
// Composition described here:
7273
// https://github.com/DataDog/instrumentation-telemetry-api-docs/blob/main/GeneratedDocumentation/ApiDocs/v2/producing-telemetry.md#app-extended-heartbeat
7374
for _, payload := range payloads {
7475
switch payload := payload.(type) {
75-
case transport.AppStarted:
76-
// Should be sent only once anyway
77-
t.extendedHeartbeat.Configuration = payload.Configuration
7876
case transport.AppDependenciesLoaded:
7977
if t.extendedHeartbeat.Dependencies == nil {
8078
t.extendedHeartbeat.Dependencies = payload.Dependencies
@@ -86,6 +84,9 @@ func (t *heartbeatEnricher) Transform(payloads []transport.Payload) ([]transport
8684
}
8785

8886
if t.extendedHeartbeatRL.Allow() {
87+
if t.getConfigs != nil {
88+
t.extendedHeartbeat.Configuration = t.getConfigs()
89+
}
8990
return append(payloads, t.extendedHeartbeat), t
9091
}
9192

0 commit comments

Comments
 (0)