diff --git a/comp/dogstatsd/server/impl/BUILD.bazel b/comp/dogstatsd/server/impl/BUILD.bazel index 3299c2eb481b..697dbb5883c7 100644 --- a/comp/dogstatsd/server/impl/BUILD.bazel +++ b/comp/dogstatsd/server/impl/BUILD.bazel @@ -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", @@ -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", diff --git a/comp/dogstatsd/server/impl/batch.go b/comp/dogstatsd/server/impl/batch.go index aeeebe0c0e19..213f4ae2a9b1 100644 --- a/comp/dogstatsd/server/impl/batch.go +++ b/comp/dogstatsd/server/impl/batch.go @@ -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() diff --git a/comp/dogstatsd/server/impl/enrich.go b/comp/dogstatsd/server/impl/enrich.go index 6a438c2dd92d..c12fbff69dc9 100644 --- a/comp/dogstatsd/server/impl/enrich.go +++ b/comp/dogstatsd/server/impl/enrich.go @@ -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" ) @@ -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 := "" @@ -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 @@ -78,7 +80,7 @@ func extractTagsMetadata(tags []string, originFromUDS string, processID uint32, metricSource = metrics.JMXCheckNameToMetricSource(jmxCheckName) continue } - tags[n] = tag + tags[n] = itag n++ } @@ -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) { @@ -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, @@ -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, @@ -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, @@ -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, diff --git a/comp/dogstatsd/server/impl/enrich_bench_test.go b/comp/dogstatsd/server/impl/enrich_bench_test.go index 67dd25ca6e95..afc9926d7bac 100644 --- a/comp/dogstatsd/server/impl/enrich_bench_test.go +++ b/comp/dogstatsd/server/impl/enrich_bench_test.go @@ -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" ) @@ -24,7 +25,7 @@ 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{ @@ -32,7 +33,7 @@ func BenchmarkExtractTagsMetadata(b *testing.B) { } 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++ { @@ -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) diff --git a/comp/dogstatsd/server/impl/enrich_test.go b/comp/dogstatsd/server/impl/enrich_test.go index aa4db5c06d7a..f787db263a0d 100644 --- a/comp/dogstatsd/server/impl/enrich_test.go +++ b/comp/dogstatsd/server/impl/enrich_test.go @@ -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" ) @@ -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) { @@ -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) { @@ -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) @@ -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) diff --git a/comp/dogstatsd/server/impl/intern.go b/comp/dogstatsd/server/impl/intern.go index 1cc1ba9ca641..b0dc668f53d6 100644 --- a/comp/dogstatsd/server/impl/intern.go +++ b/comp/dogstatsd/server/impl/intern.go @@ -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) } diff --git a/comp/dogstatsd/server/impl/intern_pipeline_test.go b/comp/dogstatsd/server/impl/intern_pipeline_test.go new file mode 100644 index 000000000000..5579df74b897 --- /dev/null +++ b/comp/dogstatsd/server/impl/intern_pipeline_test.go @@ -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) { + 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 +} diff --git a/comp/dogstatsd/server/impl/intern_telemetry.go b/comp/dogstatsd/server/impl/intern_telemetry.go index ae9ad36e69e9..166c912b2d2e 100644 --- a/comp/dogstatsd/server/impl/intern_telemetry.go +++ b/comp/dogstatsd/server/impl/intern_telemetry.go @@ -19,8 +19,9 @@ type stringInternerTelemetry struct { } type stringInternerInstanceTelemetry struct { - enabled bool - curBytes int + enabled bool + curBytes int + curEntries int resets telemetry.SimpleCounter size telemetry.SimpleGauge @@ -37,7 +38,7 @@ func newSiTelemetry(enabled bool, telemetry telemetry.Component) *stringInterner globaltlmSIRStrBytes: telemetry.NewSimpleHistogram("dogstatsd", "string_interner_str_bytes", "Number of times string with specific length were added", []float64{1, 2, 4, 8, 16, 32, 64, 128}), - resets: telemetry.NewCounter("dogstatsd", "string_interner_resets", []string{"interner_id"}, "Amount of resets of the string interner used in dogstatsd"), + resets: telemetry.NewCounter("dogstatsd", "string_interner_resets", []string{"interner_id"}, "Amount of eviction sweeps of the string interner used in dogstatsd"), size: telemetry.NewGauge("dogstatsd", "string_interner_entries", []string{"interner_id"}, "Number of entries in the string interner"), bytes: telemetry.NewGauge("dogstatsd", "string_interner_bytes", []string{"interner_id"}, "Number of bytes stored in the string interner"), hits: telemetry.NewCounter("dogstatsd", "string_interner_hits", []string{"interner_id"}, "Number of times string interner returned an existing string"), @@ -76,12 +77,29 @@ func (si *stringInternerInstanceTelemetry) Hit() { } } -// Reset increments the reset counter and updates the size and bytes gauges. -func (si *stringInternerInstanceTelemetry) Reset(length int) { - if si.enabled { - si.resets.Inc() - si.bytes.Sub(float64(si.curBytes)) - si.size.Sub(float64(length)) +// Evict counts one eviction sweep and decrements the size gauge by the number of +// tags it dropped. +// +// Unlike the reset it replaces, a sweep drops only the tags that stopped +// arriving, so the byte gauge is decremented by the evicted share rather than +// zeroed. Sizes are approximated as the running mean, since the interner does not +// keep per-entry lengths. +func (si *stringInternerInstanceTelemetry) Evict(evicted int) { + if !si.enabled { + return + } + si.resets.Inc() + si.size.Sub(float64(evicted)) + + bytes := 0 + if si.curEntries > 0 { + bytes = si.curBytes * evicted / si.curEntries + } + si.bytes.Sub(float64(bytes)) + si.curBytes -= bytes + si.curEntries -= evicted + if si.curEntries < 0 { + si.curEntries = 0 si.curBytes = 0 } } @@ -94,5 +112,6 @@ func (si *stringInternerInstanceTelemetry) Miss(length int) { si.bytes.Add(float64(length)) si.globaltlmSIRStrBytes.Observe(float64(length)) si.curBytes += length + si.curEntries++ } } diff --git a/comp/dogstatsd/server/impl/intern_test.go b/comp/dogstatsd/server/impl/intern_test.go index 52393e63f749..9c2762722071 100644 --- a/comp/dogstatsd/server/impl/intern_test.go +++ b/comp/dogstatsd/server/impl/intern_test.go @@ -48,16 +48,16 @@ func TestInternLoadOrStoreValue(t *testing.T) { // first test that the good value is returned. v := sInterner.LoadOrStore(foo) - assert.Equal("foo", v) + assert.Equal("foo", v.Value()) v = sInterner.LoadOrStore(bar) - assert.Equal("bar", v) + assert.Equal("bar", v.Value()) v = sInterner.LoadOrStore(far) - assert.Equal("far", v) + assert.Equal("far", v.Value()) v = sInterner.LoadOrStore(boo) - assert.Equal("boo", v) + assert.Equal("boo", v.Value()) } -func TestInternLoadOrStorePointer(t *testing.T) { +func TestInternLoadOrStoreHandleIdentity(t *testing.T) { telemetryComp := fxutil.Test[telemetry.Component](t, mocktelemetry.Module()) assert := assert.New(t) stringInternerTelemetry := newSiTelemetry(false, telemetryComp) @@ -67,43 +67,57 @@ func TestInternLoadOrStorePointer(t *testing.T) { bar := []byte("bar") boo := []byte("boo") - // first test that the good value is returned. - v := sInterner.LoadOrStore(foo) - assert.Equal("foo", v) + assert.Equal("foo", v.Value()) v2 := sInterner.LoadOrStore(foo) - assert.Equal(&v, &v2, "must point to the same address") + assert.Equal(v, v2, "same value must give the same handle") v2 = sInterner.LoadOrStore(bar) - assert.NotEqual(&v, &v2, "must point to a different address") + assert.NotEqual(v, v2, "different values must give different handles") v3 := sInterner.LoadOrStore(bar) - assert.Equal(&v2, &v3, "must point to the same address") + assert.Equal(v2, v3, "same value must give the same handle") v4 := sInterner.LoadOrStore(boo) - assert.NotEqual(&v, &v4, "must point to a different address") - assert.NotEqual(&v2, &v4, "must point to a different address") - assert.NotEqual(&v3, &v4, "must point to a different address") + assert.NotEqual(v, v4, "different values must give different handles") + assert.NotEqual(v2, v4, "different values must give different handles") + assert.NotEqual(v3, v4, "different values must give different handles") +} + +// Handles issued before the lookaside cache is reset stay canonical, so the +// agent never holds two copies of the same tag. The old interner reset dropped +// its strings and started allocating fresh copies. +func TestInternHandlesSurviveReset(t *testing.T) { + telemetryComp := fxutil.Test[telemetry.Component](t, mocktelemetry.Module()) + assert := assert.New(t) + stringInternerTelemetry := newSiTelemetry(false, telemetryComp) + sInterner := newStringInterner(2, 1, stringInternerTelemetry) + + before := sInterner.LoadOrStore([]byte("tag:value")) + + // force at least one reset of the lookaside cache + for i := 0; i < 8; i++ { + sInterner.LoadOrStore([]byte(fmt.Sprintf("filler:%d", i))) + } + + after := sInterner.LoadOrStore([]byte("tag:value")) + assert.Equal(before, after, "handle must stay canonical across a cache reset") + assert.Equal(before.Hash(), after.Hash()) } -func TestInternLoadOrStoreReset(t *testing.T) { +func TestInternLoadOrStoreGrowsPastSizeHint(t *testing.T) { telemetryComp := fxutil.Test[telemetry.Component](t, mocktelemetry.Module()) assert := assert.New(t) stringInternerTelemetry := newSiTelemetry(false, telemetryComp) + // the size argument is only a pre-allocation hint now, not a cap sInterner := newStringInterner(4, 1, stringInternerTelemetry) - // first test that the good value is returned. - sInterner.LoadOrStore([]byte("foo")) - assert.Equal(1, len(sInterner.strings)) - sInterner.LoadOrStore([]byte("bar")) - sInterner.LoadOrStore([]byte("bar")) - assert.Equal(2, len(sInterner.strings)) - sInterner.LoadOrStore([]byte("boo")) - assert.Equal(3, len(sInterner.strings)) - sInterner.LoadOrStore([]byte("far")) - sInterner.LoadOrStore([]byte("far")) - sInterner.LoadOrStore([]byte("far")) - assert.Equal(4, len(sInterner.strings)) - sInterner.LoadOrStore([]byte("val")) - assert.Equal(1, len(sInterner.strings)) - sInterner.LoadOrStore([]byte("val")) - assert.Equal(1, len(sInterner.strings)) + for i := 0; i < 64; i++ { + sInterner.LoadOrStore([]byte(fmt.Sprintf("tag:%d", i))) + } + assert.Equal(64, sInterner.table.Len(), "the interner must not evict tags that are still arriving") + + // and repeats keep hitting rather than re-interning + first := sInterner.LoadOrStore([]byte("tag:0")) + second := sInterner.LoadOrStore([]byte("tag:0")) + assert.Same(first, second) + assert.Equal(64, sInterner.table.Len()) } diff --git a/comp/dogstatsd/server/impl/parse.go b/comp/dogstatsd/server/impl/parse.go index bede06fc4817..614986f33604 100644 --- a/comp/dogstatsd/server/impl/parse.go +++ b/comp/dogstatsd/server/impl/parse.go @@ -16,6 +16,7 @@ import ( "github.com/DataDog/datadog-agent/comp/core/tagger/origindetection" workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def" "github.com/DataDog/datadog-agent/pkg/config/model" + "github.com/DataDog/datadog-agent/pkg/tagset" "github.com/DataDog/datadog-agent/pkg/util/containers/metrics/provider" "github.com/DataDog/datadog-agent/pkg/util/log" "github.com/DataDog/datadog-agent/pkg/util/option" @@ -73,11 +74,14 @@ type parser struct { } func newParser(cfg model.Reader, float64List *float64ListPool, workerNum int, wmeta option.Option[workloadmeta.Component], stringInternerTelemetry *stringInternerTelemetry) *parser { - stringInternerCacheSize := cfg.GetInt("dogstatsd_string_interner_size") + // The interner is no longer capped at this size: it evicts tags that stop + // arriving instead. The setting is kept as a pre-allocation hint so existing + // configs still mean something. + stringInternerSizeHint := cfg.GetInt("dogstatsd_string_interner_size") readTimestamps := cfg.GetBool("dogstatsd_no_aggregation_pipeline") return &parser{ - interner: newStringInterner(stringInternerCacheSize, workerNum, stringInternerTelemetry), + interner: newStringInterner(stringInternerSizeHint, workerNum, stringInternerTelemetry), readTimestamps: readTimestamps, float64List: float64List, dsdOriginEnabled: cfg.GetBool("dogstatsd_origin_detection_client"), @@ -107,12 +111,12 @@ func nextField(message []byte) ([]byte, []byte) { return message[:sepIndex], message[sepIndex+1:] } -func (p *parser) parseTags(rawTags []byte) []string { +func (p *parser) parseTags(rawTags []byte) []tagset.InternedTag { if len(rawTags) == 0 { return nil } tagsCount := bytes.Count(rawTags, commaSeparator) - tagsList := make([]string, tagsCount+1) + tagsList := make([]tagset.InternedTag, tagsCount+1) i := 0 for i < tagsCount { @@ -175,7 +179,7 @@ func (p *parser) parseMetricSample(message []byte) (dogstatsdMetricSample, error // sample rate, tags, container ID, timestamp, ... sampleRate := 1.0 - var tags []string + var tags []tagset.InternedTag var localData origindetection.LocalData var externalData origindetection.ExternalData var cardinality string diff --git a/comp/dogstatsd/server/impl/parse_events.go b/comp/dogstatsd/server/impl/parse_events.go index c4478dd53b54..a1d267c39d13 100644 --- a/comp/dogstatsd/server/impl/parse_events.go +++ b/comp/dogstatsd/server/impl/parse_events.go @@ -12,6 +12,7 @@ import ( "math/bits" "github.com/DataDog/datadog-agent/comp/core/tagger/origindetection" + "github.com/DataDog/datadog-agent/pkg/tagset" "github.com/DataDog/datadog-agent/pkg/util/log" ) @@ -40,7 +41,7 @@ type dogstatsdEvent struct { priority eventPriority sourceType string alertType alertType - tags []string + tags []tagset.InternedTag // localData is used for Origin Detection localData origindetection.LocalData // externalData is used for Origin Detection diff --git a/comp/dogstatsd/server/impl/parse_events_test.go b/comp/dogstatsd/server/impl/parse_events_test.go index 039f52b81e09..be38e2758405 100644 --- a/comp/dogstatsd/server/impl/parse_events_test.go +++ b/comp/dogstatsd/server/impl/parse_events_test.go @@ -10,6 +10,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/DataDog/datadog-agent/pkg/tagset" ) func parseEvent(t *testing.T, rawEvent []byte) (dogstatsdEvent, error) { @@ -27,7 +29,7 @@ func TestEventMinimal(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -41,7 +43,7 @@ func TestEventMultilinesText(t *testing.T) { assert.Equal(t, string("test\\line1\nline2\nline3"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -55,7 +57,7 @@ func TestEventPipeInTitle(t *testing.T) { assert.Equal(t, string("test\\line1\nline2\nline3"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -149,7 +151,7 @@ func TestEventMetadataTimestamp(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(21), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -163,7 +165,7 @@ func TestEventMetadataPriority(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityLow, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -177,7 +179,7 @@ func TestEventMetadataHostname(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -191,7 +193,7 @@ func TestEventMetadataAlertType(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeWarning, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -205,7 +207,7 @@ func TestEventMetadataAggregatioKey(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, string("some aggregation key"), e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -219,7 +221,7 @@ func TestEventMetadataSourceType(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string(nil), e.tags) + assert.Equal(t, []tagset.InternedTag(nil), e.tags) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, string("this is the source"), e.sourceType) @@ -233,7 +235,7 @@ func TestEventMetadataTags(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(0), e.timestamp) assert.Equal(t, priorityNormal, e.priority) - assert.Equal(t, []string{string("tag1"), string("tag2:test")}, e.tags) + assert.Equal(t, []string{string("tag1"), string("tag2:test")}, tagset.Values(e.tags)) assert.Equal(t, alertTypeInfo, e.alertType) assert.Equal(t, "", e.aggregationKey) assert.Equal(t, "", e.sourceType) @@ -247,7 +249,7 @@ func TestEventMetadataMultiple(t *testing.T) { assert.Equal(t, string("test text"), e.text) assert.Equal(t, int64(12345), e.timestamp) assert.Equal(t, priorityLow, e.priority) - assert.Equal(t, []string{string("tag1"), string("tag2:test")}, e.tags) + assert.Equal(t, []string{string("tag1"), string("tag2:test")}, tagset.Values(e.tags)) assert.Equal(t, alertTypeWarning, e.alertType) assert.Equal(t, string("aggKey"), e.aggregationKey) assert.Equal(t, string("source test"), e.sourceType) diff --git a/comp/dogstatsd/server/impl/parse_metrics.go b/comp/dogstatsd/server/impl/parse_metrics.go index bc49c7010bd9..29bb99edf7fb 100644 --- a/comp/dogstatsd/server/impl/parse_metrics.go +++ b/comp/dogstatsd/server/impl/parse_metrics.go @@ -8,8 +8,10 @@ package serverimpl import ( "bytes" "fmt" - "github.com/DataDog/datadog-agent/comp/core/tagger/origindetection" "time" + + "github.com/DataDog/datadog-agent/comp/core/tagger/origindetection" + "github.com/DataDog/datadog-agent/pkg/tagset" ) type metricType int @@ -37,7 +39,7 @@ var ( ) type dogstatsdMetricSample struct { - name string + name tagset.InternedTag // use for single value messages value float64 // use for multiple value messages @@ -46,7 +48,7 @@ type dogstatsdMetricSample struct { setValue string metricType metricType sampleRate float64 - tags []string + tags []tagset.InternedTag // localData is used for Origin Detection localData origindetection.LocalData // externalData is used for Origin Detection diff --git a/comp/dogstatsd/server/impl/parse_metrics_test.go b/comp/dogstatsd/server/impl/parse_metrics_test.go index 15004d349003..ef42933d15f7 100644 --- a/comp/dogstatsd/server/impl/parse_metrics_test.go +++ b/comp/dogstatsd/server/impl/parse_metrics_test.go @@ -49,7 +49,7 @@ func TestParseGauge(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Equal(t, 666.0, sample.value) assert.InEpsilon(t, 666.0, sample.value, epsilon) require.Nil(t, sample.values) @@ -64,7 +64,7 @@ func TestParseGaugeMultiple(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Len(t, sample.values, 2) assert.InEpsilon(t, 666.0, sample.values[0], epsilon) assert.InEpsilon(t, 777.0, sample.values[1], epsilon) @@ -79,7 +79,7 @@ func TestParseCounter(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 21.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, countType, sample.metricType) @@ -93,7 +93,7 @@ func TestParseCounterMultiple(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Len(t, sample.values, 2) assert.InEpsilon(t, 666.0, sample.values[0], epsilon) assert.InEpsilon(t, 777.0, sample.values[1], epsilon) @@ -108,13 +108,13 @@ func TestParseCounterWithTags(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "custom_counter", sample.name) + assert.Equal(t, "custom_counter", sample.name.Value()) assert.InEpsilon(t, 1.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, countType, sample.metricType) assert.Equal(t, 2, len(sample.tags)) - assert.Equal(t, "protocol:http", sample.tags[0]) - assert.Equal(t, "bench", sample.tags[1]) + assert.Equal(t, "protocol:http", sample.tags[0].Value()) + assert.Equal(t, "bench", sample.tags[1].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) assert.Zero(t, sample.ts) } @@ -124,7 +124,7 @@ func TestParseHistogram(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 21.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, histogramType, sample.metricType) @@ -138,7 +138,7 @@ func TestParseHistogramrMultiple(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Len(t, sample.values, 2) assert.InEpsilon(t, 21.0, sample.values[0], epsilon) assert.InEpsilon(t, 22.0, sample.values[1], epsilon) @@ -153,7 +153,7 @@ func TestParseTimer(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 21.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, timingType, sample.metricType) @@ -167,7 +167,7 @@ func TestParseTimerMultiple(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Len(t, sample.values, 2) assert.InEpsilon(t, 21.0, sample.values[0], epsilon) assert.InEpsilon(t, 22.0, sample.values[1], epsilon) @@ -182,7 +182,7 @@ func TestParseSet(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Equal(t, "abc", sample.setValue) assert.Equal(t, setType, sample.metricType) assert.Len(t, sample.tags, 0) @@ -197,7 +197,7 @@ func TestParseSetMultiple(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Equal(t, "abc:def", sample.setValue) assert.Equal(t, setType, sample.metricType) assert.Len(t, sample.tags, 0) @@ -210,7 +210,7 @@ func TestSampleDistribution(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 3.5, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, distributionType, sample.metricType) @@ -223,7 +223,7 @@ func TestParseDistributionMultiple(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Len(t, sample.values, 2) assert.InEpsilon(t, 3.5, sample.values[0], epsilon) assert.InEpsilon(t, 4.5, sample.values[1], epsilon) @@ -237,7 +237,7 @@ func TestParseSetUnicode(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.Equal(t, "♬†øU†øU¥ºuT0♪", sample.setValue) assert.Equal(t, setType, sample.metricType) assert.Len(t, sample.tags, 0) @@ -250,13 +250,13 @@ func TestParseGaugeWithTags(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 666.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 2, len(sample.tags)) - assert.Equal(t, "sometag1:somevalue1", sample.tags[0]) - assert.Equal(t, "sometag2:somevalue2", sample.tags[1]) + assert.Equal(t, "sometag1:somevalue1", sample.tags[0].Value()) + assert.Equal(t, "sometag2:somevalue2", sample.tags[1].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) assert.Zero(t, sample.ts) } @@ -265,7 +265,7 @@ func TestParseGaugeWithNoTags(t *testing.T) { sample, err := parseMetricSample(t, make(map[string]any), []byte("daemon:666|g")) assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 666.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) @@ -279,7 +279,7 @@ func TestParseGaugeWithSampleRate(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 666.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) @@ -293,7 +293,7 @@ func TestParseGaugeWithPoundOnly(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "daemon", sample.name) + assert.Equal(t, "daemon", sample.name.Value()) assert.InEpsilon(t, 666.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) @@ -307,12 +307,12 @@ func TestParseGaugeWithUnicode(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "♬†øU†øU¥ºuT0♪", sample.name) + assert.Equal(t, "♬†øU†øU¥ºuT0♪", sample.name.Value()) assert.InEpsilon(t, 666.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "intitulé:T0µ", sample.tags[0]) + assert.Equal(t, "intitulé:T0µ", sample.tags[0].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) assert.Zero(t, sample.ts) } @@ -363,12 +363,12 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "onetag", sample.tags[0]) + assert.Equal(t, "onetag", sample.tags[0].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) assert.Zero(t, sample.ts) @@ -382,12 +382,12 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "onetag", sample.tags[0]) + assert.Equal(t, "onetag", sample.tags[0].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) assert.Equal(t, sample.ts, time.Unix(1657100430, 0)) @@ -397,7 +397,7 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) @@ -411,7 +411,7 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) @@ -425,12 +425,12 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "thereisatag", sample.tags[0]) + assert.Equal(t, "thereisatag", sample.tags[0].Value()) assert.InEpsilon(t, 0.21, sample.sampleRate, epsilon) assert.Equal(t, sample.ts, time.Unix(1657100440, 0)) @@ -440,12 +440,12 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "thereisatag", sample.tags[0]) + assert.Equal(t, "thereisatag", sample.tags[0].Value()) assert.InEpsilon(t, 0.21, sample.sampleRate, epsilon) assert.Equal(t, sample.ts, time.Unix(1657100540, 0)) @@ -453,12 +453,12 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "thereisatag", sample.tags[0]) + assert.Equal(t, "thereisatag", sample.tags[0].Value()) assert.InEpsilon(t, 0.21, sample.sampleRate, epsilon) assert.Equal(t, sample.ts, time.Unix(1657100540, 0)) @@ -466,12 +466,12 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "atag", sample.tags[0]) + assert.Equal(t, "atag", sample.tags[0].Value()) assert.InEpsilon(t, 0.25, sample.sampleRate, epsilon) assert.Equal(t, sample.ts, time.Unix(1657100540, 0)) @@ -479,12 +479,12 @@ func TestParseGaugeWithTimestamp(t *testing.T) { assert.NoError(t, err) - assert.Equal(t, "metric", sample.name) + assert.Equal(t, "metric", sample.name.Value()) assert.InEpsilon(t, 1234.0, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, gaugeType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "atag", sample.tags[0]) + assert.Equal(t, "atag", sample.tags[0].Value()) assert.InEpsilon(t, 0.25, sample.sampleRate, epsilon) assert.Equal(t, sample.ts, time.Unix(1657100540, 0)) } @@ -514,12 +514,12 @@ func TestParseManyPipes(t *testing.T) { require.NoError(t, err) - assert.Equal(t, "example.metric", sample.name) + assert.Equal(t, "example.metric", sample.name.Value()) assert.InEpsilon(t, 2.39283, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, distributionType, sample.metricType) require.Equal(t, 1, len(sample.tags)) - assert.Equal(t, "environment:dev", sample.tags[0]) + assert.Equal(t, "environment:dev", sample.tags[0].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) }) @@ -531,13 +531,13 @@ func TestParseManyPipes(t *testing.T) { require.NoError(t, err) - assert.Equal(t, "example.metric", sample.name) + assert.Equal(t, "example.metric", sample.name.Value()) assert.InEpsilon(t, 2.39283, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, distributionType, sample.metricType) require.Equal(t, 1, len(sample.tags)) assert.Equal(t, sample.ts, time.Unix(1657100540, 0)) - assert.Equal(t, "environment:dev", sample.tags[0]) + assert.Equal(t, "environment:dev", sample.tags[0].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) }) @@ -550,13 +550,13 @@ func TestParseManyPipes(t *testing.T) { require.NoError(t, err) - assert.Equal(t, "example.metric", sample.name) + assert.Equal(t, "example.metric", sample.name.Value()) assert.InEpsilon(t, 2.39283, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, distributionType, sample.metricType) require.Equal(t, 1, len(sample.tags)) assert.Equal(t, sample.ts, time.Unix(1657100540, 0)) - assert.Equal(t, "environment:dev", sample.tags[0]) + assert.Equal(t, "environment:dev", sample.tags[0].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) }) @@ -568,13 +568,13 @@ func TestParseManyPipes(t *testing.T) { require.NoError(t, err) - assert.Equal(t, "example.metric", sample.name) + assert.Equal(t, "example.metric", sample.name.Value()) assert.InEpsilon(t, 2.39283, sample.value, epsilon) require.Nil(t, sample.values) assert.Equal(t, distributionType, sample.metricType) require.Equal(t, 1, len(sample.tags)) assert.Equal(t, sample.ts, time.Unix(1657100540, 0)) - assert.Equal(t, "environment:dev", sample.tags[0]) + assert.Equal(t, "environment:dev", sample.tags[0].Value()) assert.InEpsilon(t, 1.0, sample.sampleRate, epsilon) }) } diff --git a/comp/dogstatsd/server/impl/parse_service_checks.go b/comp/dogstatsd/server/impl/parse_service_checks.go index b2efc84823d7..6ae534104863 100644 --- a/comp/dogstatsd/server/impl/parse_service_checks.go +++ b/comp/dogstatsd/server/impl/parse_service_checks.go @@ -12,6 +12,7 @@ import ( "strconv" "github.com/DataDog/datadog-agent/comp/core/tagger/origindetection" + "github.com/DataDog/datadog-agent/pkg/tagset" "github.com/DataDog/datadog-agent/pkg/util/log" ) @@ -30,7 +31,7 @@ type dogstatsdServiceCheck struct { timestamp int64 hostname string message string - tags []string + tags []tagset.InternedTag // localData is used for Origin Detection localData origindetection.LocalData // externalData is used for Origin Detection diff --git a/comp/dogstatsd/server/impl/parse_service_checks_test.go b/comp/dogstatsd/server/impl/parse_service_checks_test.go index bea1aa6d31bc..c7acbb73188a 100644 --- a/comp/dogstatsd/server/impl/parse_service_checks_test.go +++ b/comp/dogstatsd/server/impl/parse_service_checks_test.go @@ -10,6 +10,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/DataDog/datadog-agent/pkg/tagset" ) func parseServiceCheck(t *testing.T, rawServiceCheck []byte) (dogstatsdServiceCheck, error) { @@ -27,7 +29,7 @@ func TestServiceCheckMinimal(t *testing.T) { assert.Equal(t, int64(0), sc.timestamp) assert.Equal(t, serviceCheckStatusOk, sc.status) assert.Equal(t, "", sc.message) - assert.Equal(t, []string(nil), sc.tags) + assert.Equal(t, []tagset.InternedTag(nil), sc.tags) } func TestServiceCheckError(t *testing.T) { @@ -68,7 +70,7 @@ func TestServiceCheckMetadataTimestamp(t *testing.T) { assert.Equal(t, int64(21), sc.timestamp) assert.Equal(t, serviceCheckStatusOk, sc.status) assert.Equal(t, "", sc.message) - assert.Equal(t, []string(nil), sc.tags) + assert.Equal(t, []tagset.InternedTag(nil), sc.tags) } func TestServiceCheckMetadataHostname(t *testing.T) { @@ -80,7 +82,7 @@ func TestServiceCheckMetadataHostname(t *testing.T) { assert.Equal(t, int64(0), sc.timestamp) assert.Equal(t, serviceCheckStatusOk, sc.status) assert.Equal(t, "", sc.message) - assert.Equal(t, []string(nil), sc.tags) + assert.Equal(t, []tagset.InternedTag(nil), sc.tags) } func TestServiceCheckMetadataTags(t *testing.T) { @@ -91,7 +93,7 @@ func TestServiceCheckMetadataTags(t *testing.T) { assert.Equal(t, int64(0), sc.timestamp) assert.Equal(t, serviceCheckStatusOk, sc.status) assert.Equal(t, "", sc.message) - assert.Equal(t, []string{"tag1", "tag2:test", "tag3"}, sc.tags) + assert.Equal(t, []string{"tag1", "tag2:test", "tag3"}, tagset.Values(sc.tags)) } func TestServiceCheckMetadataMessage(t *testing.T) { @@ -102,7 +104,7 @@ func TestServiceCheckMetadataMessage(t *testing.T) { assert.Equal(t, int64(0), sc.timestamp) assert.Equal(t, serviceCheckStatusOk, sc.status) assert.Equal(t, "this is fine", sc.message) - assert.Equal(t, []string(nil), sc.tags) + assert.Equal(t, []tagset.InternedTag(nil), sc.tags) } func TestServiceCheckMetadataMultiple(t *testing.T) { @@ -114,7 +116,7 @@ func TestServiceCheckMetadataMultiple(t *testing.T) { assert.Equal(t, int64(21), sc.timestamp) assert.Equal(t, serviceCheckStatusOk, sc.status) assert.Equal(t, "this is fine", sc.message) - assert.Equal(t, []string{"tag1:test", "tag2"}, sc.tags) + assert.Equal(t, []string{"tag1:test", "tag2"}, tagset.Values(sc.tags)) // multiple time the same tag sc, err = parseServiceCheck(t, []byte("_sc|agent.up|0|d:21|h:localhost|h:localhost2|d:22")) @@ -124,5 +126,5 @@ func TestServiceCheckMetadataMultiple(t *testing.T) { assert.Equal(t, int64(22), sc.timestamp) assert.Equal(t, serviceCheckStatusOk, sc.status) assert.Equal(t, "", sc.message) - assert.Equal(t, []string(nil), sc.tags) + assert.Equal(t, []tagset.InternedTag(nil), sc.tags) } diff --git a/comp/dogstatsd/server/impl/parse_test.go b/comp/dogstatsd/server/impl/parse_test.go index c68df275eedf..085510943bb2 100644 --- a/comp/dogstatsd/server/impl/parse_test.go +++ b/comp/dogstatsd/server/impl/parse_test.go @@ -11,6 +11,8 @@ import ( "testing" "github.com/stretchr/testify/assert" + + "github.com/DataDog/datadog-agent/pkg/tagset" ) func TestIdentifyEvent(t *testing.T) { @@ -44,7 +46,7 @@ func TestParseTags(t *testing.T) { rawTags := []byte("tag:test,mytag,good:boy") tags := p.parseTags(rawTags) expectedTags := []string{"tag:test", "mytag", "good:boy"} - assert.ElementsMatch(t, expectedTags, tags) + assert.ElementsMatch(t, expectedTags, tagset.Values(tags)) } func TestParseTagsEmpty(t *testing.T) { diff --git a/comp/dogstatsd/server/impl/server.go b/comp/dogstatsd/server/impl/server.go index 2f9db9ae276e..b25b7a968704 100644 --- a/comp/dogstatsd/server/impl/server.go +++ b/comp/dogstatsd/server/impl/server.go @@ -41,6 +41,7 @@ import ( "github.com/DataDog/datadog-agent/pkg/metrics/event" "github.com/DataDog/datadog-agent/pkg/metrics/servicecheck" "github.com/DataDog/datadog-agent/pkg/status/health" + "github.com/DataDog/datadog-agent/pkg/tagset" "github.com/DataDog/datadog-agent/pkg/util/option" "github.com/DataDog/datadog-agent/pkg/util/sort" statutil "github.com/DataDog/datadog-agent/pkg/util/stat" @@ -143,8 +144,11 @@ type dsdServer struct { histToDist bool histToDistPrefix string extraTags []string - Debug serverdebug.Component - filterList filterlist.Component + // extraITags is extraTags interned once at startup, so that appending them to + // every sample does not re-intern them. + extraITags []tagset.InternedTag + Debug serverdebug.Component + filterList filterlist.Component tCapture replay.Component pidMap pidmap.Component @@ -306,6 +310,7 @@ func newServerCompat(cfg model.ReaderWriter, log log.Component, hostname hostnam histToDist: histToDist, histToDistPrefix: histToDistPrefix, extraTags: extraTags, + extraITags: tagset.InternAll(extraTags), eolTerminationUDP: eolTerminationUDP, eolTerminationUDS: eolTerminationUDS, eolTerminationNamedPipe: eolTerminationNamedPipe, @@ -837,11 +842,13 @@ func (s *dsdServer) parseMetricMessage(metricSamples []metrics.MetricSample, par } if s.mapper != nil { - mapResult := s.mapper.Map(sample.name) + mapResult := s.mapper.Map(sample.name.Value()) if mapResult != nil { s.log.Tracef("Dogstatsd mapper: metric mapped from %q to %q with tags %v", sample.name, mapResult.Name, mapResult.Tags) - sample.name = mapResult.Name - sample.tags = append(sample.tags, mapResult.Tags...) + sample.name = parser.interner.LoadOrStoreString(mapResult.Name) + for _, tag := range mapResult.Tags { + sample.tags = append(sample.tags, parser.interner.LoadOrStoreString(tag)) + } } } @@ -852,12 +859,12 @@ func (s *dsdServer) parseMetricMessage(metricSamples []metrics.MetricSample, par } for idx := range metricSamples { - // All metricSamples already share the same Tags slice. We can + // All metricSamples already share the same ITags slice. We can // extends the first one and reuse it for the rest. if idx == 0 { - metricSamples[idx].Tags = append(metricSamples[idx].Tags, s.extraTags...) + metricSamples[idx].ITags = append(metricSamples[idx].ITags, s.extraITags...) } else { - metricSamples[idx].Tags = metricSamples[0].Tags + metricSamples[idx].ITags = metricSamples[0].ITags } // If we're receiving runtime metrics, we need to convert the default source to the runtime source diff --git a/comp/dogstatsd/server/impl/server_util_test.go b/comp/dogstatsd/server/impl/server_util_test.go index a1540a6d018b..a2e9fa940a03 100644 --- a/comp/dogstatsd/server/impl/server_util_test.go +++ b/comp/dogstatsd/server/impl/server_util_test.go @@ -267,7 +267,7 @@ func (m tMetricSample) testMetric(t *testing.T, actual metrics.MetricSample) { assert.Equal(t, m.Name, actual.Name, s, "name") assert.Equal(t, m.Value, actual.Value, s, "value") assert.Equal(t, m.Mtype, actual.Mtype, s, "type") - assert.ElementsMatch(t, m.Tags, actual.Tags, s, "tags") + assert.ElementsMatch(t, m.Tags, actual.GetRawTags(), s, "tags") assert.Equal(t, m.SampleRate, actual.SampleRate, s, "sample rate") assert.Equal(t, m.RawValue, actual.RawValue, s, "raw value") assert.Equal(t, m.Timestamp, actual.Timestamp, s, "timestamp") diff --git a/comp/dogstatsd/serverDebug/impl/debug.go b/comp/dogstatsd/serverDebug/impl/debug.go index 4b6c783ec59f..93cb6f442bf7 100644 --- a/comp/dogstatsd/serverDebug/impl/debug.go +++ b/comp/dogstatsd/serverDebug/impl/debug.go @@ -166,6 +166,7 @@ func (d *serverDebugImpl) StoreMetricStats(sample metrics.MetricSample) { // key defer d.tagsAccumulator.Reset() + d.tagsAccumulator.AppendInterned(sample.ITags...) d.tagsAccumulator.Append(sample.Tags...) key := d.keyGen.Generate(sample.Name, "", d.tagsAccumulator) diff --git a/pkg/aggregator/context_resolver_bench_test.go b/pkg/aggregator/context_resolver_bench_test.go index 7b2ed9b15d20..77b8064226d0 100644 --- a/pkg/aggregator/context_resolver_bench_test.go +++ b/pkg/aggregator/context_resolver_bench_test.go @@ -13,20 +13,34 @@ import ( filterlistimpl "github.com/DataDog/datadog-agent/comp/filterlist/impl" "github.com/DataDog/datadog-agent/pkg/aggregator/internal/tags" "github.com/DataDog/datadog-agent/pkg/metrics" + "github.com/DataDog/datadog-agent/pkg/tagset" ) func benchmarkContextResolver(numContexts int, b *testing.B) { + benchmarkContextResolverTags(numContexts, false, b) +} + +// benchmarkContextResolverTags tracks contexts for samples carrying either plain +// string tags or interned handles, so the two representations can be compared +// through the exact same aggregator path. +func benchmarkContextResolverTags(numContexts int, interned bool, b *testing.B) { var samples []metrics.MetricSample matcher := filterlistimpl.NewNoopTagMatcher() for i := 0; i < numContexts; i++ { - samples = append(samples, metrics.MetricSample{ + sample := metrics.MetricSample{ Name: "my.metric.name", Value: 1, Mtype: metrics.GaugeType, - Tags: []string{"foo", "bar", strconv.Itoa(i)}, SampleRate: 1, - }) + } + tags := []string{"foo", "bar", strconv.Itoa(i)} + if interned { + sample.ITags = tagset.InternAll(tags) + } else { + sample.Tags = tags + } + samples = append(samples, sample) } cache := tags.NewStore(true, "test") cr := newContextResolver(nooptagger.NewComponent(), cache, "0") @@ -51,3 +65,18 @@ func BenchmarkContextResolver1000(b *testing.B) { func BenchmarkContextResolver1000000(b *testing.B) { benchmarkContextResolver(1000000, b) } + +// Same benchmarks, with samples that carry interned tags, as the dogstatsd +// pipeline produces them. + +func BenchmarkContextResolverInterned1(b *testing.B) { + benchmarkContextResolverTags(1, true, b) +} + +func BenchmarkContextResolverInterned1000(b *testing.B) { + benchmarkContextResolverTags(1000, true, b) +} + +func BenchmarkContextResolverInterned1000000(b *testing.B) { + benchmarkContextResolverTags(1000000, true, b) +} diff --git a/pkg/aggregator/time_sampler.go b/pkg/aggregator/time_sampler.go index 82cfda1d4213..1864696a626d 100644 --- a/pkg/aggregator/time_sampler.go +++ b/pkg/aggregator/time_sampler.go @@ -119,7 +119,7 @@ func (s *TimeSampler) sample(metricSample *metrics.MetricSample, timestamp float } // Add sample to bucket if err := bucketMetrics.AddSample(contextKey, metricSample, timestamp, s.interval, nil, pkgconfigsetup.Datadog()); err != nil { - log.Debugf("TimeSampler #%d Ignoring sample '%s' on host '%s' and tags '%s': %s", s.id, metricSample.Name, metricSample.Host, metricSample.Tags, err) + log.Debugf("TimeSampler #%d Ignoring sample '%s' on host '%s' and tags '%s': %s", s.id, metricSample.Name, metricSample.Host, metricSample.GetRawTags(), err) return } } diff --git a/pkg/metriclookback/ringbuffer/buffer.go b/pkg/metriclookback/ringbuffer/buffer.go index 560a124c11f9..e557654747fc 100644 --- a/pkg/metriclookback/ringbuffer/buffer.go +++ b/pkg/metriclookback/ringbuffer/buffer.go @@ -668,7 +668,7 @@ func newContextStore(shardCount int) *contextStore { } func (s *contextStore) retain(source Source, sample metrics.MetricSample) (uint64, int) { - tags := canonicalTags(sample.Tags) + tags := canonicalTags(sample.GetRawTags()) key := buildContextKey(source, sample, tags) ctx := metricContext{ source: source, diff --git a/pkg/metrics/metric_sample.go b/pkg/metrics/metric_sample.go index 6062c8b46cc4..8033d1e7d84a 100644 --- a/pkg/metrics/metric_sample.go +++ b/pkg/metrics/metric_sample.go @@ -6,6 +6,8 @@ package metrics import ( + "slices" + tagger "github.com/DataDog/datadog-agent/comp/core/tagger/def" taggertypes "github.com/DataDog/datadog-agent/pkg/tagger/types" "github.com/DataDog/datadog-agent/pkg/tagset" @@ -96,11 +98,20 @@ const UnitMilliseconds = "millisecond" // MetricSample represents a raw metric sample type MetricSample struct { - Name string - Value float64 - RawValue string - Mtype MetricType - Tags []string + Name string + Value float64 + RawValue string + Mtype MetricType + Tags []string + // ITags carries the client tags as interned handles, each with its hash + // already computed. The dogstatsd pipeline populates this instead of Tags so + // that tags parsed off the wire travel to the aggregator without being + // re-hashed or re-copied per sample. + // + // When ITags is non-empty it is authoritative and Tags is only materialized + // on demand, by GetRawTags. Producers that do not intern (checks, python, + // otlp, ...) keep using Tags and leave this nil. + ITags []tagset.InternedTag Host string SampleRate float64 Timestamp float64 // Seconds since epoch (accepts fractional seconds) @@ -126,7 +137,17 @@ func (m *MetricSample) GetHost() string { // GetTags returns the metric sample tags func (m *MetricSample) GetTags(taggerBuffer, metricBuffer tagset.TagsAccumulator, tagger tagger.Component) { - metricBuffer.Append(m.Tags...) + if len(m.ITags) > 0 { + if hashing, ok := metricBuffer.(*tagset.HashingTagsAccumulator); ok { + hashing.AppendInterned(m.ITags...) + } else { + for _, t := range m.ITags { + metricBuffer.Append(t.Value()) + } + } + } else { + metricBuffer.Append(m.Tags...) + } tagger.EnrichTags(taggerBuffer, m.OriginInfo) } @@ -139,8 +160,13 @@ func (m *MetricSample) GetMetricType() MetricType { func (m *MetricSample) Copy() *MetricSample { dst := &MetricSample{} *dst = *m - dst.Tags = make([]string, len(m.Tags)) - copy(dst.Tags, m.Tags) + // Keep Tags nil when the source is interned, otherwise GetRawTags would hand + // out an empty slice instead of resolving ITags. + if m.Tags != nil { + dst.Tags = make([]string, len(m.Tags)) + copy(dst.Tags, m.Tags) + } + dst.ITags = slices.Clone(m.ITags) return dst } @@ -161,7 +187,14 @@ func (m *MetricSample) GetValue() float64 { // GetRawTags returns the metric sample tags, satisfying observer.MetricView. // The caller must not retain the slice — it may be returned to a pool. +// +// For interned samples this resolves the handles into Tags and caches the result +// on the sample, so an active observer costs one materialization per sample and +// nothing when no observer is attached. func (m *MetricSample) GetRawTags() []string { + if len(m.Tags) == 0 && len(m.ITags) > 0 { + m.Tags = tagset.Values(m.ITags) + } return m.Tags } diff --git a/pkg/tagset/BUILD.bazel b/pkg/tagset/BUILD.bazel index e5148b257a7c..8585757efb7f 100644 --- a/pkg/tagset/BUILD.bazel +++ b/pkg/tagset/BUILD.bazel @@ -11,6 +11,7 @@ go_library( "hashed_tags_pvt.go", "hashing_tags_accumulator.go", "hashless_tags_accumulator.go", + "intern.go", "types.go", ], importpath = "github.com/DataDog/datadog-agent/pkg/tagset", @@ -30,6 +31,7 @@ dd_agent_go_test( "hashed_tags_test.go", "hashing_tags_accumulator_test.go", "hashless_tags_accumulator_test.go", + "intern_test.go", ], embed = [":tagset"], deps = [ diff --git a/pkg/tagset/hashing_tags_accumulator.go b/pkg/tagset/hashing_tags_accumulator.go index d00038876f1c..467212e1d319 100644 --- a/pkg/tagset/hashing_tags_accumulator.go +++ b/pkg/tagset/hashing_tags_accumulator.go @@ -60,6 +60,20 @@ func (h *HashingTagsAccumulator) Append(tags ...string) { } } +// AppendInterned appends already-interned tags to the builder, reusing the hash +// memoized on each tag. Unlike Append this does no hashing at all, which is the +// point of carrying interned tags down from the parser. +// +// What lands in the accumulator is the interned copy of the string, so anything +// that later copies these tags out — an aggregator context, for instance — shares +// that copy rather than making its own. +func (h *HashingTagsAccumulator) AppendInterned(tags ...InternedTag) { + for _, t := range tags { + h.data = append(h.data, t.Value()) + h.hash = append(h.hash, t.Hash()) + } +} + // AppendHashed appends tags and corresponding hashes to the builder func (h *HashingTagsAccumulator) AppendHashed(src HashedTags) { h.data = append(h.data, src.data...) diff --git a/pkg/tagset/intern.go b/pkg/tagset/intern.go new file mode 100644 index 000000000000..c45d451b2e64 --- /dev/null +++ b/pkg/tagset/intern.go @@ -0,0 +1,261 @@ +// 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 tagset + +import ( + "time" + + "github.com/twmb/murmur3" +) + +// Tag is one interned tag: a single copy of the tag string, plus the murmur3 +// hash tagset uses as tag identity, computed once when the tag is first seen. +// +// Interning gives us somewhere to memoize the hash, so a tag is hashed once for +// as long as it keeps arriving rather than once per metric sample carrying it. +// +// Tags are referred to by pointer (see InternedTag), so a tag costs 8 bytes in a +// sample or a context rather than the 16 bytes of a string header, and copying a +// tag around never copies its bytes. +type Tag struct { + value string + hash uint64 + + // lastSeen is the Table epoch in which this tag last arrived. It is written + // only by the goroutine that owns the Table, and is what makes the table + // self-sizing: see Table. + lastSeen uint32 +} + +// InternedTag is a reference to an interned tag. The nil value is a valid empty +// tag: Value returns "". +type InternedTag = *Tag + +// Intern returns a standalone interned tag for s, not owned by any Table. +// +// Use this for the handful of tags that come from configuration rather than off +// the wire (static tags, infra mode tags, tests): they are interned once at +// startup and live forever, so they neither need nor benefit from a table. +// +// Nothing depends on two equal tags being the same *Tag — tag identity is +// (hash, value), never pointer equality — so a standalone tag mixes freely with +// table-owned ones. +func Intern(s string) InternedTag { + return &Tag{value: s, hash: murmur3.StringSum64(s)} +} + +// InternAll returns standalone interned tags for a slice of strings. +func InternAll(tags []string) []InternedTag { + if tags == nil { + return nil + } + out := make([]InternedTag, len(tags)) + for i, t := range tags { + out[i] = Intern(t) + } + return out +} + +// Value returns the interned tag string. It never allocates: every holder of +// this tag shares the one copy. +func (t *Tag) Value() string { + if t == nil { + return "" + } + return t.value +} + +// Hash returns the memoized murmur3 hash of the tag. It matches +// murmur3.StringSum64(t.Value()). +func (t *Tag) Hash() uint64 { + if t == nil { + return murmur3.StringSum64("") + } + return t.hash +} + +// touch records that the tag arrived in the given epoch. +// +// The store is guarded by a comparison because a hot tag arrives many times per +// epoch and writing every time dirties one cache line per tag per sample. The +// tag has just been read, so the compare is free; skipping the write is not. +func (t *Tag) touch(epoch uint32) { + if t.lastSeen != epoch { + t.lastSeen = epoch + } +} + +// String implements fmt.Stringer so interned tags render as their value in logs. +func (t *Tag) String() string { + return t.Value() +} + +// Values resolves interned tags into a plain string slice. This is the boundary +// where tags become ordinary strings again, on the way to the serializer. +func Values(tags []InternedTag) []string { + if tags == nil { + return nil + } + out := make([]string, len(tags)) + for i, t := range tags { + out[i] = t.Value() + } + return out +} + +// AppendValues appends the resolved tag strings to dst. +func AppendValues(dst []string, tags []InternedTag) []string { + for _, t := range tags { + dst = append(dst, t.Value()) + } + return dst +} + +const ( + // tableEpochInterval is how long an epoch lasts. Eviction granularity is one + // epoch, so this trades promptness against how often we walk the table. + tableEpochInterval = 10 * time.Second + + // tableEpochsRetained is how many epochs a tag may go unseen before it is + // evicted. With the interval above, a tag that stops arriving is dropped + // after roughly 30 seconds. + tableEpochsRetained = 3 + + // tableCheckInterval is how many lookups to serve between clock reads. The + // clock is not read per lookup: at dogstatsd rates that would show up in a + // profile, and epoch boundaries do not need to be precise. + tableCheckInterval = 8192 +) + +// Table interns tags read off the wire, so that a tag seen many times is stored +// once and hashed once. +// +// The table is self-sizing rather than capped at a configured entry count: each +// tag records the epoch it was last seen in, and a tag that has not arrived for +// tableEpochsRetained epochs is evicted. Nothing has to be released by hand, and +// no tuning knob decides how many distinct tags a workload is allowed. +// +// Dropping a tag that some in-flight sample or aggregator context still refers +// to is harmless: they hold the *Tag directly, so it stays alive and valid for +// as long as they need it. The only consequence is that if the tag comes back +// later we intern a second copy, which is the same thing the old size-capped +// interner did on reset — except that here it can only happen to a tag that went +// quiet, never to a hot one. +// +// A Table is not safe for concurrent use. Each dogstatsd worker owns one. +type Table struct { + tags map[string]*Tag + + epoch uint32 + lastEpochStart time.Time + opsUntilCheck int + + // onEvict, if set, is called with the number of tags dropped by a sweep. + onEvict func(evicted int) +} + +// NewTable returns an empty Table. sizeHint pre-allocates room for that many +// tags; it is only a hint, the table grows and shrinks with the workload. +func NewTable(sizeHint int) *Table { + return &Table{ + tags: make(map[string]*Tag, sizeHint), + epoch: 1, + lastEpochStart: time.Now(), + opsUntilCheck: tableCheckInterval, + } +} + +// SetEvictionCallback registers a callback invoked after each sweep that evicted +// at least one tag. +func (t *Table) SetEvictionCallback(onEvict func(evicted int)) { + t.onEvict = onEvict +} + +// LoadOrStore returns the interned tag for key, interning it if this is the +// first time the table has seen it. found reports whether it was already known. +func (t *Table) LoadOrStore(key []byte) (InternedTag, bool) { + // The map lookup with string(key) does not allocate a string: the compiler + // recognizes the pattern and looks up the bytes directly. + // See https://github.com/golang/go/commit/f5f5a8b6209f84961687d993b93ea0d397f5d5bf + if tag, ok := t.tags[string(key)]; ok { + tag.touch(t.epoch) + t.tick() + return tag, true + } + + return t.store(string(key)), false +} + +// LoadOrStoreString is LoadOrStore for a key the caller already holds as a string. +func (t *Table) LoadOrStoreString(key string) (InternedTag, bool) { + if tag, ok := t.tags[key]; ok { + tag.touch(t.epoch) + t.tick() + return tag, true + } + + return t.store(key), false +} + +func (t *Table) store(key string) InternedTag { + tag := &Tag{ + value: key, + hash: murmur3.StringSum64(key), + lastSeen: t.epoch, + } + t.tags[key] = tag + t.tick() + return tag +} + +// Len returns the number of tags currently interned. +func (t *Table) Len() int { + return len(t.tags) +} + +// tick advances the epoch and sweeps once the epoch interval has elapsed. The +// clock is only read every tableCheckInterval operations. +func (t *Table) tick() { + t.opsUntilCheck-- + if t.opsUntilCheck > 0 { + return + } + t.opsUntilCheck = tableCheckInterval + + now := time.Now() + if now.Sub(t.lastEpochStart) < tableEpochInterval { + return + } + t.lastEpochStart = now + t.advanceEpoch() +} + +// advanceEpoch starts a new epoch and sweeps the tags that have now gone unseen +// for long enough. +func (t *Table) advanceEpoch() { + t.epoch++ + t.sweep() +} + +// sweep drops tags that have not been seen for tableEpochsRetained epochs. +func (t *Table) sweep() { + if t.epoch <= tableEpochsRetained { + return + } + cutoff := t.epoch - tableEpochsRetained + + evicted := 0 + for key, tag := range t.tags { + if tag.lastSeen < cutoff { + delete(t.tags, key) + evicted++ + } + } + + if evicted > 0 && t.onEvict != nil { + t.onEvict(evicted) + } +} diff --git a/pkg/tagset/intern_test.go b/pkg/tagset/intern_test.go new file mode 100644 index 000000000000..9c6e47aaf6b1 --- /dev/null +++ b/pkg/tagset/intern_test.go @@ -0,0 +1,155 @@ +// 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 tagset + +import ( + "fmt" + "testing" + "unsafe" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/twmb/murmur3" +) + +// unsafeStringData identifies the backing array of a string, to show that two +// interned references share one copy rather than each holding their own. +func unsafeStringData(s string) *byte { + return unsafe.StringData(s) +} + +func TestInternedTagValueAndHash(t *testing.T) { + tag := Intern("env:prod") + assert.Equal(t, "env:prod", tag.Value()) + assert.Equal(t, murmur3.StringSum64("env:prod"), tag.Hash()) + + // the nil tag is usable, so a zero-valued sample does not panic + var empty InternedTag + assert.Equal(t, "", empty.Value()) + assert.Equal(t, murmur3.StringSum64(""), empty.Hash()) +} + +func TestTableLoadOrStoreDeduplicates(t *testing.T) { + table := NewTable(4) + + first, found := table.LoadOrStore([]byte("env:prod")) + assert.False(t, found) + second, found := table.LoadOrStore([]byte("env:prod")) + assert.True(t, found) + + assert.Same(t, first, second, "the same tag must resolve to the same interned copy") + assert.Equal(t, 1, table.Len()) + + // and the strings share storage rather than being separate copies + assert.Equal(t, unsafeStringData(first.Value()), unsafeStringData(second.Value())) +} + +func TestTableLoadOrStoreStringMatchesBytes(t *testing.T) { + table := NewTable(4) + + fromBytes, _ := table.LoadOrStore([]byte("env:prod")) + fromString, found := table.LoadOrStoreString("env:prod") + + assert.True(t, found) + assert.Same(t, fromBytes, fromString) +} + +func TestTableGrowsWithoutACap(t *testing.T) { + table := NewTable(4) + + for i := 0; i < 10_000; i++ { + table.LoadOrStore([]byte(fmt.Sprintf("tag:%d", i))) + } + + assert.Equal(t, 10_000, table.Len(), + "the table is sized by liveness, so tags still arriving are never dropped") +} + +func TestTableEvictsTagsThatStopArriving(t *testing.T) { + table := NewTable(4) + + var evicted int + table.SetEvictionCallback(func(n int) { evicted += n }) + + stale, _ := table.LoadOrStore([]byte("stale:tag")) + fresh, _ := table.LoadOrStore([]byte("fresh:tag")) + require.Equal(t, 2, table.Len()) + + // advance past the retention window, keeping only one of the two alive + for i := 0; i <= tableEpochsRetained+1; i++ { + table.advanceEpoch() + table.LoadOrStoreString("fresh:tag") + } + + assert.Equal(t, 1, table.Len(), "the tag that stopped arriving must be evicted") + assert.Equal(t, 1, evicted) + + _, found := table.LoadOrStoreString("fresh:tag") + assert.True(t, found, "the tag that kept arriving must be retained") + + // An evicted tag stays valid for whoever still holds it: samples and contexts + // keep the *Tag directly, so eviction can never invalidate them. + assert.Equal(t, "stale:tag", stale.Value()) + assert.Equal(t, murmur3.StringSum64("stale:tag"), stale.Hash()) + assert.Equal(t, "fresh:tag", fresh.Value()) +} + +func TestTableReinternsAfterEviction(t *testing.T) { + table := NewTable(4) + + before, _ := table.LoadOrStore([]byte("quiet:tag")) + for i := 0; i <= tableEpochsRetained+1; i++ { + table.advanceEpoch() + } + require.Equal(t, 0, table.Len()) + + after, found := table.LoadOrStore([]byte("quiet:tag")) + assert.False(t, found, "an evicted tag is interned afresh when it comes back") + assert.NotSame(t, before, after) + + // The transient duplicate is still a correct tag: same value, same hash, so it + // dedupes against the old copy anywhere tag identity is (hash, value). + assert.Equal(t, before.Value(), after.Value()) + assert.Equal(t, before.Hash(), after.Hash()) +} + +func TestAppendInternedMatchesAppend(t *testing.T) { + table := NewTable(4) + tags := []InternedTag{} + for _, s := range []string{"env:prod", "service:api", "az:us-east-1a"} { + tag, _ := table.LoadOrStoreString(s) + tags = append(tags, tag) + } + + interned := NewHashingTagsAccumulator() + interned.AppendInterned(tags...) + + plain := NewHashingTagsAccumulatorWithTags([]string{"env:prod", "service:api", "az:us-east-1a"}) + + assert.Equal(t, plain.Get(), interned.Get()) + assert.Equal(t, plain.Hashes(), interned.Hashes(), + "memoized hashes must match what Append computes, or context keys would diverge") + assert.Equal(t, plain.Hash(), interned.Hash()) +} + +func TestValues(t *testing.T) { + assert.Nil(t, Values(nil)) + assert.Equal(t, []string{"a", "b"}, Values(InternAll([]string{"a", "b"}))) +} + +func BenchmarkTableLoadOrStoreHit(b *testing.B) { + table := NewTable(64) + keys := make([][]byte, 32) + for i := range keys { + keys[i] = []byte(fmt.Sprintf("tag%d:value%d", i, i)) + table.LoadOrStore(keys[i]) + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + table.LoadOrStore(keys[i%len(keys)]) + } +} diff --git a/pkg/util/infratags/BUILD.bazel b/pkg/util/infratags/BUILD.bazel index f8971bcf12a8..ab1ded7792a7 100644 --- a/pkg/util/infratags/BUILD.bazel +++ b/pkg/util/infratags/BUILD.bazel @@ -8,6 +8,7 @@ go_library( visibility = ["//visibility:public"], deps = [ "//pkg/config/model", + "//pkg/tagset", ], ) diff --git a/pkg/util/infratags/infratags.go b/pkg/util/infratags/infratags.go index 1528c0de0623..8cd01e6a844c 100644 --- a/pkg/util/infratags/infratags.go +++ b/pkg/util/infratags/infratags.go @@ -16,6 +16,7 @@ import ( "strings" pkgconfigmodel "github.com/DataDog/datadog-agent/pkg/config/model" + "github.com/DataDog/datadog-agent/pkg/tagset" ) // InfraModeCloudCostTag is the tag appended to eligible integration metrics in cloud_cost_only mode. @@ -36,7 +37,10 @@ func tagsForMode(infraMode string) (tags []string, ok bool) { // A nil *Tagger disables tagging. type Tagger struct { infraModeTags []string - taggedChecks map[string]struct{} // nil = all non-custom checks eligible + // infraModeITags is infraModeTags interned once, for callers that carry + // interned tags (the DogStatsD pipeline). + infraModeITags []tagset.InternedTag + taggedChecks map[string]struct{} // nil = all non-custom checks eligible } // NewTagger resolves the infra mode tagging configuration from cfg. @@ -49,13 +53,13 @@ func NewTagger(cfg pkgconfigmodel.Reader) *Tagger { } checks := cfg.GetStringSlice("integration." + infraMode + ".tagged") if len(checks) == 0 { - return &Tagger{infraModeTags: tags} + return &Tagger{infraModeTags: tags, infraModeITags: tagset.InternAll(tags)} } taggedChecks := make(map[string]struct{}, len(checks)) for _, c := range checks { taggedChecks[c] = struct{}{} } - return &Tagger{infraModeTags: tags, taggedChecks: taggedChecks} + return &Tagger{infraModeTags: tags, infraModeITags: tagset.InternAll(tags), taggedChecks: taggedChecks} } // IsCheckEligible reports whether the given check should receive infra mode tags. @@ -84,3 +88,13 @@ func (t *Tagger) AppendTags(tags []string) []string { return append(tags, t.infraModeTags...) } + +// AppendInternedTags appends the infra mode tags to tags as interned handles. +// The infra mode tags are interned once, when the Tagger is built. +func (t *Tagger) AppendInternedTags(tags []tagset.InternedTag) []tagset.InternedTag { + if t == nil || len(t.infraModeITags) == 0 { + return tags + } + + return append(tags, t.infraModeITags...) +}