wip

Author: vickenty

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()

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,

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,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++ {
@@ -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)

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)

comp/dogstatsd/server/impl/intern.go

@@ -7,36 +7,40 @@ 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 strings read off the
+// wire, so that a tag or metric name seen many times is stored 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
+// Canonicalization itself is done by the standard library `unique` package: it
+// owns the single copy of each string and keeps it alive exactly as long as some
+// handle still refers to it. That removes the two problems the hand-rolled
+// interner had:
+//
+//   - a reset dropped the canonical strings, so the same tag arriving after a
+//     reset allocated a fresh copy and the agent held several copies of the same
+//     tag at once. Handles issued before a reset stay valid and stay canonical.
+//   - entries were retained forever (up to maxSize) even if no sample referenced
+//     them anymore. `unique` reclaims a string once the last handle goes away.
+//
+// What is left here is a per-worker lookaside cache. It exists for two reasons:
+// looking up by []byte without allocating a string (which unique.Make cannot do),
+// and memoizing the tag hash so a given tag is hashed once rather than once per
+// sample carrying it. maxSize now only bounds this cache, not the strings.
 type stringInterner struct {
-	strings map[string]string
+	cache   map[string]tagset.InternedTag
 	maxSize int
 	id      string
 
 	telemetry *stringInternerInstanceTelemetry
 }
 
 func newStringInterner(maxSize int, internerID int, siTelemetry *stringInternerTelemetry) *stringInterner {
-	// telemetryOnce.Do(func() { initGlobalTelemetry(telemetrycomp) })
-
 	id := fmt.Sprintf("interner_%d", internerID)
 	i := &stringInterner{
-		strings:   make(map[string]string),
+		cache:     make(map[string]tagset.InternedTag),
 		id:        id,
 		maxSize:   maxSize,
 		telemetry: siTelemetry.PrepareForID(id),
@@ -45,31 +49,52 @@ func newStringInterner(maxSize int, internerID int, siTelemetry *stringInternerT
 	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 {
+// LoadOrStore always returns a handle for the given key, interning it if this is
+// the first time the worker sees it.
+func (i *stringInterner) LoadOrStore(key []byte) tagset.InternedTag {
 	// 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 {
+	if t, found := i.cache[string(key)]; found {
+		i.telemetry.Hit()
+		return t
+	}
+
+	if len(i.cache) >= i.maxSize {
+		i.telemetry.Reset(len(i.cache))
+
+		i.cache = make(map[string]tagset.InternedTag)
+	}
+
+	t := tagset.InternBytes(key)
+	// Key the cache on the canonical string so the cache does not retain a second
+	// copy of it.
+	i.cache[t.Value()] = t
+
+	i.telemetry.Miss(len(key))
+
+	return t
+}
+
+// LoadOrStoreString is LoadOrStore for a key the caller already holds as a string.
+func (i *stringInterner) LoadOrStoreString(key string) tagset.InternedTag {
+	if t, found := i.cache[key]; found {
 		i.telemetry.Hit()
-		return s
+		return t
 	}
 
-	if len(i.strings) >= i.maxSize {
-		i.telemetry.Reset(len(i.strings))
+	if len(i.cache) >= i.maxSize {
+		i.telemetry.Reset(len(i.cache))
 
-		i.strings = make(map[string]string)
+		i.cache = make(map[string]tagset.InternedTag)
 	}
 
-	s := string(key)
-	i.strings[s] = s
+	t := tagset.Intern(key)
+	i.cache[t.Value()] = t
 
-	i.telemetry.Miss(len(s))
+	i.telemetry.Miss(len(key))
 
-	return s
+	return t
 }

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
+}