Add measurement processor support (#493)

Author: kristinapathak

CHANGELOG.md

@@ -8,7 +8,11 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## Unreleased
 
-## [1.18.5](https://github.com/lightstep/otel-launcher-go/releases/tag/v1.18.4) - 2023-06-26)
+## [1.18.6](https://github.com/lightstep/otel-launcher-go/releases/tag/v1.18.6) - 2023-06-27)
+
+- Support experimental `MeasurementProcessor` in the LS metrics SDK. [#493](https://github.com/lightstep/otel-launcher-go/pull/493)
+
+## [1.18.5](https://github.com/lightstep/otel-launcher-go/releases/tag/v1.18.5) - 2023-06-26)
 
 - Correct the translation of Span StatusCode. [#495](https://github.com/lightstep/otel-launcher-go/pull/495)
 

VERSION

@@ -1 +1 @@
-1.18.5
+1.18.6

go.mod

@@ -3,8 +3,8 @@ module github.com/lightstep/otel-launcher-go
 go 1.18
 
 require (
-	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.18.5
-	github.com/lightstep/otel-launcher-go/pipelines v1.18.5
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.18.6
+	github.com/lightstep/otel-launcher-go/pipelines v1.18.6
 	github.com/sethvargo/go-envconfig v0.8.3
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/otel v1.16.0
@@ -42,8 +42,8 @@ require (
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/knadh/koanf v1.5.0 // indirect
 	github.com/lightstep/go-expohisto v1.0.0 // indirect
-	github.com/lightstep/otel-launcher-go/lightstep/instrumentation v1.18.5 // indirect
-	github.com/lightstep/otel-launcher-go/lightstep/sdk/internal v1.18.5 // indirect
+	github.com/lightstep/otel-launcher-go/lightstep/instrumentation v1.18.6 // indirect
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/internal v1.18.6 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
 	github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect

launcher/version.go

@@ -14,4 +14,4 @@
 
 package launcher
 
-const version = "1.18.5"
+const version = "1.18.6"

lightstep/sdk/metric/README.md

@@ -182,3 +182,14 @@ attribute filters are applied.
 When the aggregator's output grows to `AggregatorCardinalityLimit`,
 new attribute sets will be replaced by the overflow attribute set,
 which is `{ otel.metric.overflow=true }`.
+
+#### MeasurementProcessor
+
+The `MeasurementProcessor` interface that makes it possible to extend
+the set of attributes from synchronous instrument events, which allows
+metric attributes to be generated from the OpenTelemetry request
+context and/or W3C Tracecontext baggage.
+
+This hook also supports removing attributes from metric events based
+on attribute value before they are aggregated, for example to
+dynamically configure allowed cardinality values.

lightstep/sdk/metric/example/go.mod

@@ -3,7 +3,7 @@ module github.com/lightstep/otel-launcher-go/lightstep/sdk/metric/example
 go 1.18
 
 require (
-	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.18.5
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.18.6
 	github.com/lightstep/otel-launcher-go/pipelines v1.8.0
 	go.opentelemetry.io/proto/otlp v0.20.0
 )
@@ -35,7 +35,7 @@ require (
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/knadh/koanf v1.5.0 // indirect
 	github.com/lightstep/go-expohisto v1.0.0 // indirect
-	github.com/lightstep/otel-launcher-go/lightstep/sdk/internal v1.18.5 // indirect
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/internal v1.18.6 // indirect
 	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
 	github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect

lightstep/sdk/metric/go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/google/go-cmp v0.5.9
 	github.com/lightstep/go-expohisto v1.0.0
-	github.com/lightstep/otel-launcher-go/lightstep/sdk/internal v1.18.5
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/internal v1.18.6
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/collector v0.79.0
 	go.opentelemetry.io/collector/component v0.79.0

lightstep/sdk/metric/internal/syncstate/sync.go

@@ -327,7 +327,7 @@ func (inst *Observer) ObserveFloat64(ctx context.Context, num float64, cfg OpCon
 }
 
 // Observe performs a generic update for any synchronous instrument.
-func Observe[N number.Any, Traits number.Traits[N]](_ context.Context, inst *Observer, num N, cfg OpConfig) {
+func Observe[N number.Any, Traits number.Traits[N]](ctx context.Context, inst *Observer, num N, cfg OpConfig) {
 	if inst == nil {
 		// Instrument was completely disabled by the view.
 		return
@@ -339,15 +339,20 @@ func Observe[N number.Any, Traits number.Traits[N]](_ context.Context, inst *Obs
 		return
 	}
 
-	var rec *recordKV
+	var keyValues []attribute.KeyValue
 	if cfg.KeyValues != nil {
-		rec = acquireUninitializedKV[N](inst, cfg.KeyValues)
+		keyValues = cfg.KeyValues
 	} else {
 		// TODO: This is a new code path for optimization,
 		// for now fall back to the slow path.
-		rec = acquireUninitializedKV[N](inst, cfg.Attributes.ToSlice())
+		keyValues = cfg.Attributes.ToSlice()
 	}
 
+	if inst.performance.MeasurementProcessor != nil {
+		keyValues = inst.performance.MeasurementProcessor.Process(ctx, keyValues)
+	}
+	rec := acquireUninitializedKV[N](inst, keyValues)
+
 	defer rec.refMapped.unref()
 
 	rec.readAccumulator().(viewstate.Updater[N]).Update(num)

lightstep/sdk/metric/internal/syncstate/sync_test.go

@@ -53,9 +53,16 @@ var (
 		Now:   endTime,
 	}
 
+	testKeyVal = attribute.String("test", "test-value")
+
 	safePerf = sdkinstrument.Performance{
 		IgnoreCollisions:          false,
 		InactiveCollectionPeriods: 1,
+		MeasurementProcessor: &testMeasurementProcessor{
+			attrs: []attribute.KeyValue{
+				testKeyVal,
+			},
+		},
 	}
 
 	unsafePerf = sdkinstrument.Performance{
@@ -99,6 +106,16 @@ var (
 	)
 )
 
+type testMeasurementProcessor struct {
+	attrs []attribute.KeyValue
+}
+
+func (mp *testMeasurementProcessor) Process(ctx context.Context, inAttrs []attribute.KeyValue) []attribute.KeyValue {
+	outAttrs := make([]attribute.KeyValue, 0, len(mp.attrs)+len(inAttrs))
+	outAttrs = append(outAttrs, inAttrs...)
+	return append(outAttrs, mp.attrs...)
+}
+
 func TestSyncStateDeltaConcurrencyInt(t *testing.T) {
 	testSyncStateConcurrency[int64, number.Int64Traits](t, deltaUpdate[int64], deltaSelector)
 }
@@ -310,6 +327,7 @@ func TestSyncStatePartialNoopInstrument(t *testing.T) {
 			test.Point(startTime, endTime,
 				expectHist,
 				aggregation.CumulativeTemporality,
+				testKeyVal,
 			),
 		),
 	)
@@ -397,6 +415,7 @@ func TestOutOfRangeValues(t *testing.T) {
 				startTime, endTime,
 				negOne,
 				aggregation.CumulativeTemporality,
+				testKeyVal,
 			))
 		}
 

lightstep/sdk/metric/sdkinstrument/performance.go

@@ -14,6 +14,12 @@
 
 package sdkinstrument
 
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+)
+
 // DefaultInactiveCollectionPeriods is how many collection periods to
 // delay before removing records from memory.
 const DefaultInactiveCollectionPeriods = 10
@@ -48,6 +54,16 @@ type Performance struct {
 	// AggregatorCardinalityLimit is a hard limit on output
 	// cardinality for all aggregators in the SDK.
 	AggregatorCardinalityLimit uint32
+
+	// MeasurementProcessor supports modifying the attributes
+	// based on context.  Only applies to synchronous instruments.
+	MeasurementProcessor MeasurementProcessor
+}
+
+// MeasurementProcessor allows applications to extend metric events
+// based on context.
+type MeasurementProcessor interface {
+	Process(ctx context.Context, inAttrs []attribute.KeyValue) (outAttrs []attribute.KeyValue)
 }
 
 // Validate returns a Performance object with 0 values replaced by