Add client metadata to key (#306)

* Add client metadata to key

This enables persisting client metadata from the
incoming metrics, through aggregation, and into
the produced metrics.

Author: axw

processor/lsmintervalprocessor/config/config.go

@@ -18,6 +18,8 @@
 package config // import "github.com/elastic/opentelemetry-collector-components/processor/lsmintervalprocessor/config"
 
 import (
+	"fmt"
+	"strings"
 	"time"
 
 	"go.opentelemetry.io/collector/component"
@@ -29,21 +31,34 @@ type Config struct {
 	// Directory is the data directory used by the database to store files.
 	// If the directory is empty in-memory storage is used.
 	Directory string `mapstructure:"directory"`
+
 	// PassThrough is a configuration that determines whether summary
 	// metrics should be passed through as they are or aggregated. This
 	// is because they lead to lossy aggregations.
 	PassThrough PassThrough `mapstructure:"pass_through"`
+
 	// Intervals is a list of interval configuration that the processor
 	// will aggregate over. The interval duration must be in increasing
 	// order and must be a factor of the smallest interval duration.
 	// TODO (lahsivjar): Make specifying interval easier. We can just
 	// optimize the timer to run on differnt times and remove any
 	// restriction on different interval configuration.
-	Intervals      []IntervalConfig `mapstructure:"intervals"`
-	ResourceLimit  LimitConfig      `mapstructure:"resource_limit"`
-	ScopeLimit     LimitConfig      `mapstructure:"scope_limit"`
-	MetricLimit    LimitConfig      `mapstructure:"metric_limit"`
-	DatapointLimit LimitConfig      `mapstructure:"datapoint_limit"`
+	Intervals []IntervalConfig `mapstructure:"intervals"`
+
+	// MetadataKeys is a list of client.Metadata keys that will be
+	// propagated through the metrics aggregated by the processor.
+	//
+	// Only the listed metadata keys will be propagated to the
+	// resulting metrics.
+	//
+	// Entries are case-insensitive. Duplicated entries will
+	// trigger a validation error.
+	MetadataKeys []string `mapstructure:"metadata_keys"`
+
+	ResourceLimit  LimitConfig `mapstructure:"resource_limit"`
+	ScopeLimit     LimitConfig `mapstructure:"scope_limit"`
+	MetricLimit    LimitConfig `mapstructure:"metric_limit"`
+	DatapointLimit LimitConfig `mapstructure:"datapoint_limit"`
 }
 
 // PassThrough determines whether metrics should be passed through as they
@@ -93,7 +108,15 @@ type Attribute struct {
 	Value any    `mapstructure:"value"`
 }
 
-func (config *Config) Validate() error {
+func (cfg *Config) Validate() error {
 	// TODO (lahsivjar): Add validation for interval duration
+	uniq := map[string]bool{}
+	for _, k := range cfg.MetadataKeys {
+		l := strings.ToLower(k)
+		if _, has := uniq[l]; has {
+			return fmt.Errorf("duplicate entry in metadata_keys: %q (case-insensitive)", l)
+		}
+		uniq[l] = true
+	}
 	return nil
 }

processor/lsmintervalprocessor/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.116.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.116.0
 	github.com/stretchr/testify v1.10.0
+	go.opentelemetry.io/collector/client v1.22.0
 	go.opentelemetry.io/collector/component v0.116.0
 	go.opentelemetry.io/collector/component/componenttest v0.116.0
 	go.opentelemetry.io/collector/confmap v1.22.0
@@ -19,6 +20,7 @@ require (
 	go.opentelemetry.io/collector/pdata v1.22.0
 	go.opentelemetry.io/collector/processor v0.116.0
 	go.opentelemetry.io/collector/processor/processortest v0.116.0
+	go.opentelemetry.io/otel v1.32.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
 )
@@ -81,7 +83,6 @@ require (
 	go.opentelemetry.io/collector/pipeline v0.116.0 // indirect
 	go.opentelemetry.io/collector/processor/xprocessor v0.116.0 // indirect
 	go.opentelemetry.io/collector/semconv v0.116.0 // indirect
-	go.opentelemetry.io/otel v1.32.0 // indirect
 	go.opentelemetry.io/otel/metric v1.32.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.32.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.32.0 // indirect

processor/lsmintervalprocessor/go.sum

@@ -140,6 +140,8 @@ github.com/ua-parser/uap-go v0.0.0-20240611065828-3a4781585db6/go.mod h1:BUbeWZi
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+go.opentelemetry.io/collector/client v1.22.0 h1:AAUzHuqYQqxoNqacw1WXgGF/MxtBTwNZuhBvJIorgA0=
+go.opentelemetry.io/collector/client v1.22.0/go.mod h1:wcCSdTwbDVNTycoqs7BiDNVj3e1Ta7EnWH2sAofKnEk=
 go.opentelemetry.io/collector/component v0.116.0 h1:SQE1YeVfYCN7bw1n4hknUwJE5U/1qJL552sDhAdSlaA=
 go.opentelemetry.io/collector/component v0.116.0/go.mod h1:MYgXFZWDTq0uPgF1mkLSFibtpNqksRVAOrmihckOQEs=
 go.opentelemetry.io/collector/component/componentstatus v0.116.0 h1:wpgY0H2K9IPBzaNAvavKziK86VZ7TuNFQbS9OC4Z6Cs=

processor/lsmintervalprocessor/internal/merger/key.go

@@ -20,54 +20,95 @@ package merger // import "github.com/elastic/opentelemetry-collector-components/
 import (
 	"encoding/binary"
 	"errors"
+	"fmt"
+	"slices"
 	"time"
 )
 
-// TODO (lahsivjar): Think about multitenancy, should be part of the key
 type Key struct {
 	Interval       time.Duration
 	ProcessingTime time.Time
-}
 
-// NewKey creates a new instance of the merger key.
-func NewKey(ivl time.Duration, pTime time.Time) Key {
-	return Key{
-		Interval:       ivl,
-		ProcessingTime: pTime,
-	}
+	// Metadata holds an ordered list of arbitrary keys and associated
+	// string values to associate with the interval and processing time.
+	Metadata []KeyValues
 }
 
-// SizeBinary returns the size of the Key when binary encoded.
-// The interval, represented by time.Duration, is encoded to
-// 2 bytes by converting it into seconds. This allows a max of
-// ~18 hours duration.
-func (k *Key) SizeBinary() int {
-	// 2 bytes for interval, 8 bytes for processing time
-	return 10
+type KeyValues struct {
+	Key    string
+	Values []string
 }
 
-// Marshal marshals the key into binary representation.
-func (k *Key) Marshal() ([]byte, error) {
-	ivlSeconds := uint16(k.Interval.Seconds())
-
-	var (
-		offset int
-		d      [10]byte
-	)
-	binary.BigEndian.PutUint16(d[offset:], ivlSeconds)
-	offset += 2
+// AppendBinary marshals the key into its binary representation,
+// appending it to b.
+func (k *Key) AppendBinary(b []byte) ([]byte, error) {
+	b = slices.Grow(b, 10)
+	b = binary.BigEndian.AppendUint16(b, uint16(k.Interval.Seconds()))
+	b = binary.BigEndian.AppendUint64(b, uint64(k.ProcessingTime.Unix()))
+	if len(k.Metadata) != 0 {
+		b = binary.AppendUvarint(b, uint64(len(k.Metadata)))
+		for _, kvs := range k.Metadata {
+			mk := kvs.Key
+			mvs := kvs.Values
 
-	binary.BigEndian.PutUint64(d[offset:], uint64(k.ProcessingTime.Unix()))
-
-	return d[:], nil
+			b = binary.AppendUvarint(b, uint64(len(mk)))
+			b = append(b, mk...)
+			b = binary.AppendUvarint(b, uint64(len(mvs)))
+			for _, mv := range mvs {
+				b = binary.AppendUvarint(b, uint64(len(mv)))
+				b = append(b, mv...)
+			}
+		}
+	}
+	return b, nil
 }
 
 // Unmarshal unmarshals the binary representation of the Key.
 func (k *Key) Unmarshal(d []byte) error {
-	if len(d) != 10 {
+	if len(d) < 10 {
 		return errors.New("failed to unmarshal key, invalid sized buffer provided")
 	}
 	k.Interval = time.Duration(binary.BigEndian.Uint16(d[:2])) * time.Second
 	k.ProcessingTime = time.Unix(int64(binary.BigEndian.Uint64(d[2:10])), 0)
+
+	d = d[10:]
+	if len(d) > 0 {
+		numKeys, n := binary.Uvarint(d)
+		if n <= 0 {
+			return fmt.Errorf("error reading number of metadata keys (n=%d)", n)
+		}
+		d = d[n:]
+		k.Metadata = make([]KeyValues, numKeys)
+
+		for i := range numKeys {
+			mklen, n := binary.Uvarint(d)
+			if n <= 0 {
+				return fmt.Errorf("error reading metadata key length (n=%d)", n)
+			}
+			d = d[n:]
+			mk := string(d[:mklen])
+			d = d[mklen:]
+
+			numValues, n := binary.Uvarint(d)
+			if n <= 0 {
+				return fmt.Errorf("error reading number of metadata values for %q (n=%d)", mk, n)
+			}
+			d = d[n:]
+			mvs := make([]string, numValues)
+			for i := range numValues {
+				mvlen, n := binary.Uvarint(d)
+				if n <= 0 {
+					return fmt.Errorf("error reading metadata value length for %q (n=%d)", mk, n)
+				}
+				d = d[n:]
+				mv := string(d[:mvlen])
+				d = d[mvlen:]
+				mvs[i] = mv
+			}
+
+			k.Metadata[i] = KeyValues{Key: mk, Values: mvs}
+		}
+	}
+
 	return nil
 }

processor/lsmintervalprocessor/internal/merger/key_test.go

@@ -28,29 +28,43 @@ import (
 
 func TestKey(t *testing.T) {
 	for _, tc := range []struct {
-		name           string
-		ivl            time.Duration
-		processingTime time.Time
+		name string
+		key  Key
 	}{
 		{
-			name:           "zero",
-			ivl:            0,
-			processingTime: time.Unix(0, 0),
+			name: "zero",
+			key: Key{
+				Interval:       0,
+				ProcessingTime: time.Unix(0, 0),
+			},
 		},
 		{
-			name:           "non_zero",
-			ivl:            time.Minute,
-			processingTime: time.Unix(time.Now().Unix(), 0),
+			name: "non_zero",
+			key: Key{
+				Interval:       time.Minute,
+				ProcessingTime: time.Unix(time.Now().Unix(), 0),
+			},
+		},
+		{
+			name: "with_metadata_keys",
+			key: Key{
+				Interval:       time.Minute,
+				ProcessingTime: time.Unix(time.Now().Unix(), 0),
+				Metadata: []KeyValues{
+					{Key: "empty", Values: []string{}},
+					{Key: "one_empty_value", Values: []string{""}},
+					{Key: "one_nonempty_value", Values: []string{"non-empty"}},
+					{Key: "mixed_values", Values: []string{"", "non-empty"}},
+				},
+			},
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			key := NewKey(tc.ivl, tc.processingTime)
-			assert.Equal(t, 10, key.SizeBinary())
-			b, err := key.Marshal()
+			b, err := tc.key.AppendBinary(nil)
 			assert.NoError(t, err)
 			var newKey Key
 			assert.NoError(t, newKey.Unmarshal(b))
-			assert.Equal(t, newKey, key)
+			assert.Equal(t, newKey, tc.key)
 		})
 	}
 }
@@ -60,14 +74,20 @@ func TestKeyOrdered(t *testing.T) {
 	ts := time.Unix(0, 0)
 	ivl := time.Minute
 
-	before := NewKey(ivl, ts)
+	before := Key{
+		Interval:       ivl,
+		ProcessingTime: ts,
+	}
 	for i := 0; i < 10; i++ {
-		beforeBytes, err := before.Marshal()
+		beforeBytes, err := before.AppendBinary(nil)
 		require.NoError(t, err)
 
 		ts = ts.Add(time.Minute)
-		after := NewKey(ivl, ts)
-		afterBytes, err := after.Marshal()
+		after := Key{
+			Interval:       ivl,
+			ProcessingTime: ts,
+		}
+		afterBytes, err := after.AppendBinary(nil)
 		require.NoError(t, err)
 
 		// before should always come first