Skip to content
Draft

wip #54928

Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 2 additions & 0 deletions comp/dogstatsd/server/impl/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ dd_agent_go_test(
"enrich_serverless_test.go",
"enrich_test.go",
"filterlist_init_test.go",
"intern_pipeline_test.go",
"intern_test.go",
"parse_events_fuzz_test.go",
"parse_events_test.go",
Expand Down Expand Up @@ -132,6 +133,7 @@ dd_agent_go_test(
"//pkg/metrics/event",
"//pkg/metrics/servicecheck",
"//pkg/tagger/types",
"//pkg/tagset",
"//pkg/util/fxutil",
"//pkg/util/hostname",
"//pkg/util/infratags",
Expand Down
1 change: 1 addition & 0 deletions comp/dogstatsd/server/impl/batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@ func (s *shardKeyGenerator) Generate(sample metrics.MetricSample, shards int) ui
// TODO(remy): re-using this tagsBuffer later in the pipeline (by sharing
// it in the sample?) would reduce CPU usage, avoiding to recompute
// the tags hashes while generating the context key.
s.tagsBuffer.AppendInterned(sample.ITags...)
s.tagsBuffer.Append(sample.Tags...)
h := s.keyGenerator.Generate(sample.Name, sample.Host, s.tagsBuffer)
s.tagsBuffer.Reset()
Expand Down
22 changes: 13 additions & 9 deletions comp/dogstatsd/server/impl/enrich.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (
metricsevent "github.com/DataDog/datadog-agent/pkg/metrics/event"
"github.com/DataDog/datadog-agent/pkg/metrics/servicecheck"
taggertypes "github.com/DataDog/datadog-agent/pkg/tagger/types"
"github.com/DataDog/datadog-agent/pkg/tagset"
"github.com/DataDog/datadog-agent/pkg/util/infratags"
utilstrings "github.com/DataDog/datadog-agent/pkg/util/strings"
)
Expand Down Expand Up @@ -47,7 +48,7 @@ type enrichConfig struct {
// (origins, cardinality), and the JMX check name extracted from dd.internal.jmx_check_name (empty
// string if absent). The JMX check name is returned so callers can pass it directly to
// AppendJMXDogstatsdInfraTags without re-scanning the tag slice.
func extractTagsMetadata(tags []string, originFromUDS string, processID uint32, localData origindetection.LocalData, externalData origindetection.ExternalData, cardinality string, conf enrichConfig) ([]string, string, taggertypes.OriginInfo, metrics.MetricSource, string) {
func extractTagsMetadata(tags []tagset.InternedTag, originFromUDS string, processID uint32, localData origindetection.LocalData, externalData origindetection.ExternalData, cardinality string, conf enrichConfig) ([]tagset.InternedTag, string, taggertypes.OriginInfo, metrics.MetricSource, string) {
host := conf.defaultHostname
metricSource := GetDefaultMetricSource()
jmxCheckName := ""
Expand All @@ -63,7 +64,8 @@ func extractTagsMetadata(tags []string, originFromUDS string, processID uint32,
origin.LocalData.ProcessID = processID

n := 0
for _, tag := range tags {
for _, itag := range tags {
tag := itag.Value()
if strings.HasPrefix(tag, hostTagPrefix) {
host = tag[len(hostTagPrefix):]
continue
Expand All @@ -78,7 +80,7 @@ func extractTagsMetadata(tags []string, originFromUDS string, processID uint32,
metricSource = metrics.JMXCheckNameToMetricSource(jmxCheckName)
continue
}
tags[n] = tag
tags[n] = itag
n++
}

Expand Down Expand Up @@ -148,10 +150,10 @@ func tsToFloatForSamples(ts time.Time) float64 {
}

func enrichMetricSample(dest []metrics.MetricSample, ddSample dogstatsdMetricSample, origin string, processID uint32, listenerID string, conf enrichConfig, filterList *utilstrings.Matcher) []metrics.MetricSample {
metricName := ddSample.name
metricName := ddSample.name.Value()
tags, hostnameFromTags, extractedOrigin, metricSource, jmxCheckName := extractTagsMetadata(ddSample.tags, origin, processID, ddSample.localData, ddSample.externalData, ddSample.cardinality, conf)
if conf.infraTagger.IsCheckEligible(jmxCheckName) {
tags = conf.infraTagger.AppendTags(tags)
tags = conf.infraTagger.AppendInternedTags(tags)
}

if !isExcluded(metricName, conf.metricPrefix, conf.metricPrefixBlacklist) {
Expand Down Expand Up @@ -180,7 +182,7 @@ func enrichMetricSample(dest []metrics.MetricSample, ddSample dogstatsdMetricSam
metrics.MetricSample{
Host: hostnameFromTags,
Name: metricName,
Tags: tags,
ITags: tags,
Mtype: mtype,
Value: ddSample.values[idx],
SampleRate: ddSample.sampleRate,
Expand All @@ -199,7 +201,7 @@ func enrichMetricSample(dest []metrics.MetricSample, ddSample dogstatsdMetricSam
return append(dest, metrics.MetricSample{
Host: hostnameFromTags,
Name: metricName,
Tags: tags,
ITags: tags,
Mtype: mtype,
Value: ddSample.value,
SampleRate: ddSample.sampleRate,
Expand Down Expand Up @@ -237,7 +239,8 @@ func enrichEventAlertType(dogstatsdAlertType alertType) metricsevent.AlertType {
}

func enrichEvent(event dogstatsdEvent, origin string, processID uint32, conf enrichConfig) *metricsevent.Event {
tags, hostnameFromTags, extractedOrigin, _, _ := extractTagsMetadata(event.tags, origin, processID, event.localData, event.externalData, event.cardinality, conf)
itags, hostnameFromTags, extractedOrigin, _, _ := extractTagsMetadata(event.tags, origin, processID, event.localData, event.externalData, event.cardinality, conf)
tags := tagset.Values(itags)

enrichedEvent := &metricsevent.Event{
Title: event.title,
Expand Down Expand Up @@ -274,7 +277,8 @@ func enrichServiceCheckStatus(status serviceCheckStatus) servicecheck.ServiceChe
}

func enrichServiceCheck(serviceCheck dogstatsdServiceCheck, origin string, processID uint32, conf enrichConfig) *servicecheck.ServiceCheck {
tags, hostnameFromTags, extractedOrigin, _, _ := extractTagsMetadata(serviceCheck.tags, origin, processID, serviceCheck.localData, serviceCheck.externalData, serviceCheck.cardinality, conf)
itags, hostnameFromTags, extractedOrigin, _, _ := extractTagsMetadata(serviceCheck.tags, origin, processID, serviceCheck.localData, serviceCheck.externalData, serviceCheck.cardinality, conf)
tags := tagset.Values(itags)

enrichedServiceCheck := &servicecheck.ServiceCheck{
CheckName: serviceCheck.name,
Expand Down
7 changes: 4 additions & 3 deletions comp/dogstatsd/server/impl/enrich_bench_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (

"github.com/DataDog/datadog-agent/comp/core/tagger/origindetection"
"github.com/DataDog/datadog-agent/pkg/metrics"
"github.com/DataDog/datadog-agent/pkg/tagset"
utilstrings "github.com/DataDog/datadog-agent/pkg/util/strings"
)

Expand All @@ -24,15 +25,15 @@ func buildTags(tagCount int) []string {
}

// used to store the result and avoid optimizations
var tags []string
var tags []tagset.InternedTag

func BenchmarkExtractTagsMetadata(b *testing.B) {
conf := enrichConfig{
defaultHostname: "hostname",
}
for i := 20; i <= 200; i += 20 {
b.Run(fmt.Sprintf("%d-tags", i), func(sb *testing.B) {
baseTags := append([]string{hostTagPrefix + "foo", entityIDTagPrefix + "bar"}, buildTags(i/10)...)
baseTags := tagset.InternAll(append([]string{hostTagPrefix + "foo", entityIDTagPrefix + "bar"}, buildTags(i/10)...))
sb.ResetTimer()

for n := 0; n < sb.N; n++ {
Expand All @@ -46,7 +47,7 @@ func BenchmarkMetricsExclusion(b *testing.B) {
conf := enrichConfig{}

sample := dogstatsdMetricSample{
name: "datadog.agent.testing.metric.does_not_match",
name: tagset.Intern("datadog.agent.testing.metric.does_not_match"),
}

out := make([]metrics.MetricSample, 0, 10)
Expand Down
23 changes: 18 additions & 5 deletions comp/dogstatsd/server/impl/enrich_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
"github.com/DataDog/datadog-agent/pkg/metrics/event"
"github.com/DataDog/datadog-agent/pkg/metrics/servicecheck"
taggertypes "github.com/DataDog/datadog-agent/pkg/tagger/types"
"github.com/DataDog/datadog-agent/pkg/tagset"
"github.com/DataDog/datadog-agent/pkg/util/infratags"
utilstrings "github.com/DataDog/datadog-agent/pkg/util/strings"
)
Expand Down Expand Up @@ -50,7 +51,7 @@ func parseAndEnrichSingleMetricMessage(t *testing.T, message []byte, conf enrich
if len(samples) != 1 {
return metrics.MetricSample{}, errors.New("wrong number of metrics parsed")
}
return samples[0], nil
return resolveSampleTags(samples)[0], nil
}

func parseAndEnrichMultipleMetricMessage(t *testing.T, message []byte, conf enrichConfig) ([]metrics.MetricSample, error) {
Expand All @@ -63,7 +64,18 @@ func parseAndEnrichMultipleMetricMessage(t *testing.T, message []byte, conf enri
}

samples := []metrics.MetricSample{}
return enrichMetricSample(samples, parsed, "", 0, "", conf, nil), nil
return resolveSampleTags(enrichMetricSample(samples, parsed, "", 0, "", conf, nil)), nil
}

// resolveSampleTags materializes the interned tags the dogstatsd pipeline
// produces into MetricSample.Tags, so that tests can keep asserting on plain
// strings. In production this resolution happens further down, at the tag
// accumulator.
func resolveSampleTags(samples []metrics.MetricSample) []metrics.MetricSample {
for i := range samples {
samples[i].Tags = tagset.Values(samples[i].ITags)
}
return samples
}

func parseAndEnrichServiceCheckMessage(t *testing.T, message []byte, conf enrichConfig) (*servicecheck.ServiceCheck, error) {
Expand Down Expand Up @@ -1500,8 +1512,8 @@ func TestEnrichTags(t *testing.T) {
tt.wantedOrigin.ProductOrigin = origindetection.ProductOriginDogStatsD

t.Run(tt.name, func(t *testing.T) {
tags, host, origin, metricSource, _ := extractTagsMetadata(tt.args.tags, tt.args.originFromUDS, 0, tt.args.localData, tt.args.externalData, tt.args.cardinality, tt.args.conf)
assert.Equal(t, tt.wantedTags, tags)
tags, host, origin, metricSource, _ := extractTagsMetadata(tagset.InternAll(tt.args.tags), tt.args.originFromUDS, 0, tt.args.localData, tt.args.externalData, tt.args.cardinality, tt.args.conf)
assert.Equal(t, tt.wantedTags, tagset.Values(tags))
assert.Equal(t, tt.wantedHost, host)
assert.Equal(t, tt.wantedOrigin, origin)
assert.Equal(t, tt.wantedMetricSource, metricSource)
Expand Down Expand Up @@ -1548,7 +1560,8 @@ func TestEnrichTagsWithJMXCheckName(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
tags, _, _, metricSource, _ := extractTagsMetadata(tt.tags, "", 0, origindetection.LocalData{}, origindetection.ExternalData{}, "", enrichConfig{})
itags, _, _, metricSource, _ := extractTagsMetadata(tagset.InternAll(tt.tags), "", 0, origindetection.LocalData{}, origindetection.ExternalData{}, "", enrichConfig{})
tags := tagset.Values(itags)
assert.Equal(t, tt.wantedTags, tags)
assert.Equal(t, tt.wantedMetricSource, metricSource)
assert.NotContains(t, tags, tt.jmxCheckName)
Expand Down
81 changes: 36 additions & 45 deletions comp/dogstatsd/server/impl/intern.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,69 +7,60 @@ package serverimpl

import (
"fmt"

"github.com/DataDog/datadog-agent/pkg/tagset"
)

// stringInterner is a string cache providing a longer life for strings,
// helping to avoid GC runs because they're re-used many times instead of
// created every time.
// stringInterner hands out tagset.InternedTag values for tag and metric names
// read off the wire, so that a string seen many times is stored once and hashed
// once.
//
// The current interning strategy is fairly simple, but can require manual
// adjustments of the `maxSize` to improve performance, which is not ideal.

// However the current strategy works well enough, and there is an
// accepted go proposal to offer an "interning" mechanism from the
// go runtime directly.

// Once this is available, the interner design should be re-visited to
// take advantage of the new "Unique" api that is proposed below.
// ref: https://github.com/golang/go/issues/62483
// The interning itself lives in tagset.Table, which sizes itself by liveness:
// tags that keep arriving are kept, tags that stop arriving are evicted. This
// replaces the old fixed `dogstatsd_string_interner_size` cap, which had to be
// tuned per workload and, when exceeded, threw the whole table away — so a
// workload with more distinct tags than the cap re-allocated the same strings
// over and over, and the agent held several copies of a tag at once.
//
// One interner per dogstatsd worker; not safe for concurrent use.
type stringInterner struct {
strings map[string]string
maxSize int
id string
table *tagset.Table
id string

telemetry *stringInternerInstanceTelemetry
}

func newStringInterner(maxSize int, internerID int, siTelemetry *stringInternerTelemetry) *stringInterner {
// telemetryOnce.Do(func() { initGlobalTelemetry(telemetrycomp) })

func newStringInterner(sizeHint int, internerID int, siTelemetry *stringInternerTelemetry) *stringInterner {
id := fmt.Sprintf("interner_%d", internerID)
i := &stringInterner{
strings: make(map[string]string),
table: tagset.NewTable(sizeHint),
id: id,
maxSize: maxSize,
telemetry: siTelemetry.PrepareForID(id),
}
i.table.SetEvictionCallback(i.telemetry.Evict)

return i
}

// LoadOrStore always returns the string from the cache, adding it into the
// cache if needed.
// If we need to store a new entry and the cache is at its maximum capacity,
// it is reset.
func (i *stringInterner) LoadOrStore(key []byte) string {
// here is the string interner trick: the map lookup using
// string(key) doesn't actually allocate a string, but is
// returning the string value -> no new heap allocation
// for this string.
// See https://github.com/golang/go/commit/f5f5a8b6209f84961687d993b93ea0d397f5d5bf
if s, found := i.strings[string(key)]; found {
i.telemetry.Hit()
return s
}
// LoadOrStore returns the interned tag for key, interning it if this is the first
// time this worker has seen it.
func (i *stringInterner) LoadOrStore(key []byte) tagset.InternedTag {
tag, found := i.table.LoadOrStore(key)
i.record(found, len(key))
return tag
}

if len(i.strings) >= i.maxSize {
i.telemetry.Reset(len(i.strings))
// LoadOrStoreString is LoadOrStore for a key the caller already holds as a string.
func (i *stringInterner) LoadOrStoreString(key string) tagset.InternedTag {
tag, found := i.table.LoadOrStoreString(key)
i.record(found, len(key))
return tag
}

i.strings = make(map[string]string)
func (i *stringInterner) record(found bool, length int) {
if found {
i.telemetry.Hit()
return
}

s := string(key)
i.strings[s] = s

i.telemetry.Miss(len(s))

return s
i.telemetry.Miss(length)
}
81 changes: 81 additions & 0 deletions comp/dogstatsd/server/impl/intern_pipeline_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
// 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 2016-present Datadog, Inc.

package serverimpl

import (
"testing"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"

"github.com/DataDog/datadog-agent/pkg/metrics"
"github.com/DataDog/datadog-agent/pkg/tagset"
)

// TestInternedTagsReachTheSample checks that tags parsed off the wire arrive at
// the aggregator as handles rather than being copied into plain strings, and that
// the handles are canonical: the same tag seen on two different messages resolves
// to the same handle, so the aggregator holds one copy of the string.
func TestInternedTagsReachTheSample(t *testing.T) {

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Include the new pipeline test in the Bazel test target

The impl_test target has an explicit source list that does not include intern_pipeline_test.go, so the repository's supported Bazel test path silently skips both newly added end-to-end interning assertions. Add this file to comp/dogstatsd/server/impl/BUILD.bazel so the coverage runs under Bazel.

AGENTS.md reference: AGENTS.md:L93-L96

Useful? React with 👍 / 👎.

conf := enrichConfig{defaultHostname: "default-hostname"}

first, err := parseAndEnrichMultipleMetricMessageNoResolve(t,
[]byte("daemon:666|g|#env:prod,service:api"), conf)
require.NoError(t, err)
require.Len(t, first, 1)

second, err := parseAndEnrichMultipleMetricMessageNoResolve(t,
[]byte("other:1|c|#env:prod,service:api"), conf)
require.NoError(t, err)
require.Len(t, second, 1)

// The pipeline populates ITags, not Tags.
require.Len(t, first[0].ITags, 2)
assert.Nil(t, first[0].Tags, "the interned pipeline must not materialize Tags")

assert.Equal(t, []string{"env:prod", "service:api"}, tagset.Values(first[0].ITags))

// Handles are canonical across parsers and messages.
assert.Equal(t, first[0].ITags, second[0].ITags,
"identical tags on different messages must share handles")

// The memoized hash matches what the accumulator would compute for the string.
for _, itag := range first[0].ITags {
assert.Equal(t, tagset.Intern(itag.Value()).Hash(), itag.Hash())
}
}

// TestInternedTagsFeedAccumulator checks the aggregator side: GetTags must push
// the handles into the hashing accumulator with their precomputed hashes, and the
// resulting tag set must be identical to the one a plain-string sample produces.
func TestInternedTagsFeedAccumulator(t *testing.T) {
conf := enrichConfig{defaultHostname: "default-hostname"}

samples, err := parseAndEnrichMultipleMetricMessageNoResolve(t,
[]byte("daemon:666|g|#env:prod,service:api"), conf)
require.NoError(t, err)
require.Len(t, samples, 1)

interned := tagset.NewHashingTagsAccumulator()
interned.AppendInterned(samples[0].ITags...)

plain := tagset.NewHashingTagsAccumulatorWithTags([]string{"env:prod", "service:api"})

assert.Equal(t, plain.Get(), interned.Get())
assert.Equal(t, plain.Hash(), interned.Hash(),
"interned tags must hash identically to the same tags as strings")
}

func parseAndEnrichMultipleMetricMessageNoResolve(t *testing.T, message []byte, conf enrichConfig) ([]metrics.MetricSample, error) {
deps := newServerDeps(t)
stringInternerTelemetry := newSiTelemetry(false, deps.Telemetry)
parser := newParser(deps.Config, newFloat64ListPool(deps.Config, deps.Telemetry), 1, deps.WMeta, stringInternerTelemetry)
parsed, err := parser.parseMetricSample(message)
if err != nil {
return nil, err
}
return enrichMetricSample(nil, parsed, "", 0, "", conf, nil), nil
}
Loading
Loading