Avoid race conditions when the input attribute slice is shared (#422)

CHANGELOG.md

@@ -8,6 +8,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## Unreleased
 
+## [1.15.1](https://github.com/lightstep/otel-launcher-go/releases/tag/v1.15.1) - 2023-04-03)
+
+- Correct race conditions related to sharing of the input attributes slice. [#422](https://github.com/lightstep/otel-launcher-go/pull/422)
+
 ## [1.15.0](https://github.com/lightstep/otel-launcher-go/releases/tag/v1.15.0) - 2023-04-03)
 
 - Remove access token length check. [#412](https://github.com/lightstep/otel-launcher-go/pull/412)

VERSION

@@ -1 +1 @@
-1.15.0
+1.15.1

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.15.0
-	github.com/lightstep/otel-launcher-go/pipelines v1.15.0
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.15.1
+	github.com/lightstep/otel-launcher-go/pipelines v1.15.1
 	github.com/sethvargo/go-envconfig v0.8.3
 	github.com/stretchr/testify v1.8.2
 	go.opentelemetry.io/otel v1.14.0
@@ -23,7 +23,7 @@ require (
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0 // indirect
 	github.com/lightstep/go-expohisto v1.0.0 // indirect
-	github.com/lightstep/otel-launcher-go/lightstep/instrumentation v1.15.0 // indirect
+	github.com/lightstep/otel-launcher-go/lightstep/instrumentation v1.15.1 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect

launcher/version.go

@@ -14,4 +14,4 @@
 
 package launcher
 
-const version = "1.15.0"
+const version = "1.15.1"

lightstep/sdk/metric/benchmark_test.go

@@ -56,7 +56,7 @@ func BenchmarkCounterAddNoAttrs(b *testing.B) {
 	}
 }
 
-func BenchmarkCounterAddOneAttr(b *testing.B) {
+func BenchmarkCounterAddOneAttrSafe(b *testing.B) {
 	ctx := context.Background()
 	rdr := NewManualReader("bench")
 	provider := NewMeterProvider(WithReader(rdr))
@@ -82,6 +82,38 @@ func BenchmarkCounterAddOneAttrUnsafe(b *testing.B) {
 	}
 }
 
+func BenchmarkCounterAddOneAttrSliceReuseSafe(b *testing.B) {
+	ctx := context.Background()
+	rdr := NewManualReader("bench")
+	provider := NewMeterProvider(WithReader(rdr))
+	b.ReportAllocs()
+
+	attrs := []attribute.KeyValue{
+		attribute.String("K", "V"),
+	}
+	cntr, _ := provider.Meter("test").Int64Counter("hello")
+
+	for i := 0; i < b.N; i++ {
+		cntr.Add(ctx, 1, attrs...)
+	}
+}
+
+func BenchmarkCounterAddOneAttrSliceReuseUnsafe(b *testing.B) {
+	ctx := context.Background()
+	rdr := NewManualReader("bench")
+	provider := NewMeterProvider(WithReader(rdr), unsafePerf)
+	b.ReportAllocs()
+
+	attrs := []attribute.KeyValue{
+		attribute.String("K", "V"),
+	}
+	cntr, _ := provider.Meter("test").Int64Counter("hello")
+
+	for i := 0; i < b.N; i++ {
+		cntr.Add(ctx, 1, attrs...)
+	}
+}
+
 func BenchmarkCounterAddOneInvalidAttr(b *testing.B) {
 	ctx := context.Background()
 	rdr := NewManualReader("bench")

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.15.0
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.15.1
 	github.com/lightstep/otel-launcher-go/pipelines v1.8.0
 	go.opentelemetry.io/proto/otlp v0.19.0
 )

lightstep/sdk/metric/internal/asyncstate/async.go

@@ -133,7 +133,11 @@ func (obs *Observer) getOrCreate(cs *callbackState, attrs []attribute.KeyValue)
 		cs.state.store[obs] = imap
 	}
 
-	aset := attribute.NewSet(attrs...)
+	// Copy the attribute list in case the caller has multiple
+	// concurrent callers.
+	acpy := make([]attribute.KeyValue, len(attrs))
+	copy(acpy, attrs)
+	aset := attribute.NewSet(acpy...)
 	se, has := imap[aset]
 	if !has {
 		se = comp.NewAccumulator(aset)

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

@@ -227,18 +227,13 @@ type record struct {
 	// to ensure that once.Do(initialize) is called.
 	accumulatorUnsafe viewstate.Accumulator
 
-	// attrsListInputUnsafe is a temporary copy of the caller's
-	// attribute list in its original order, possibly having duplicates.
-	// this is passed to attribute.NewSetWithSortable, which performs a
-	// stable-sort of the slice and resets the pointer to nil inside
-	// once.Do(initialize).  This field is used consistently regardless
-	// of IgnoreCollisions.
-	attrsListInputUnsafe attribute.Sortable
-
-	// attrsListCopy is a copy of the original attribute list, only
-	// used when checking collisions (i.e., IgnoreCollisions is false).
-	// set in computeAttrsUnderLock.
-	attrsListCopy []attribute.KeyValue
+	// attrsList is the caller's copy of the attribute list.  when
+	// IgnoreCollisions is false, this is copied immediately as we
+	// need it for reference after the call returns.  In both cases,
+	// when the entry is not found and a new record is initialized,
+	// a new copy of the attribute list will be created for the
+	// call to NewSet.
+	attrsList []attribute.KeyValue
 }
 
 // normalCollect equals conditionalCollect(false), is named
@@ -285,26 +280,36 @@ func (rec *record) readAccumulator() viewstate.Accumulator {
 //
 // readAccumulator() calls this inside a sync.Once.Do().
 func (rec *record) initialize() {
-	// Note that rec.attrsListInputUnsafe is set to nil in NewSetWithSortable().
-	aset := attribute.NewSetWithSortable(rec.attrsListInputUnsafe, &rec.attrsListInputUnsafe)
+	// We need another copy of the attribute list because NewSet()
+	// will sort it in place.
+	acpy := make([]attribute.KeyValue, len(rec.attrsList))
+	copy(acpy, rec.attrsList)
 
+	// When ignoring collisions, the list is no longer used.
+	if rec.inst.performance.IgnoreCollisions {
+		rec.attrsList = nil
+	}
+
+	aset := attribute.NewSet(acpy...)
 	rec.accumulatorUnsafe = rec.inst.compiled.NewAccumulator(aset)
 }
 
 // computeAttrsUnderLock sets the attribute.Set that will be used to
 // construct the accumulator.
 func (rec *record) computeAttrsUnderLock(attrs []attribute.KeyValue) {
-	// The work of NewSetWithSortable and NewAccumulator is
-	// deferred until once.Do(initialize) outside of the lock.
-	rec.attrsListInputUnsafe = attrs
-
+	// The work of NewSet and NewAccumulator is deferred until
+	// once.Do(initialize) outside of the lock but while the call
+	// is still in flight.
 	if rec.inst.performance.IgnoreCollisions {
+		rec.attrsList = attrs
 		return
 	}
 
+	// We have to copy the attributes here because the caller may
+	// modify their copy of the list after the call returns.
 	acpy := make([]attribute.KeyValue, len(attrs))
 	copy(acpy, attrs)
-	rec.attrsListCopy = acpy
+	rec.attrsList = acpy
 }
 
 func (inst *Observer) ObserveInt64(ctx context.Context, num int64, attrs ...attribute.KeyValue) {
@@ -460,7 +465,7 @@ func acquireReadLocked(inst *Observer, fp uint64, attrs []attribute.KeyValue, ov
 
 	// Potentially test for hash collisions.
 	if !inst.performance.IgnoreCollisions {
-		for rec != nil && !attributesEqual(attrs, rec.attrsListCopy) {
+		for rec != nil && !attributesEqual(attrs, rec.attrsList) {
 			rec = rec.next
 		}
 	}
@@ -533,7 +538,7 @@ func acquireWrite(inst *Observer, fp uint64, newRec *record) (*record, bool) {
 
 	for oldRec := inst.current[fp]; oldRec != nil; oldRec = oldRec.next {
 
-		if inst.performance.IgnoreCollisions || attributesEqual(oldRec.attrsListCopy, newRec.attrsListCopy) {
+		if inst.performance.IgnoreCollisions || attributesEqual(oldRec.attrsList, newRec.attrsList) {
 			if oldRec.refMapped.ref() {
 				return oldRec, true
 			}

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

@@ -1368,3 +1368,52 @@ func TestCardinalityOverflowOscillationDelta(t *testing.T) {
 		)
 	}
 }
+
+func TestInputAttributeSliceRaceCondition(t *testing.T) {
+	ctx := context.Background()
+	lib := instrumentation.Library{
+		Name: "testlib",
+	}
+	perf := sdkinstrument.Performance{}
+	vcs := make([]*viewstate.Compiler, 1)
+	vcs[0] = viewstate.New(lib, view.New("test", perf))
+
+	desc := test.Descriptor("c", sdkinstrument.SyncCounter, number.Float64Kind)
+
+	pipes := make(pipeline.Register[viewstate.Instrument], 1)
+	pipes[0], _ = vcs[0].Compile(desc)
+
+	inst := New(desc, perf, nil, pipes)
+	require.NotNil(t, inst)
+
+	// This attribute slice is shared by multiple concurrent callers.
+	attrs := []attribute.KeyValue{
+		{
+			Key:   "key1",
+			Value: attribute.StringValue("val1"),
+		},
+		{
+			Key:   "key2",
+			Value: attribute.StringValue("val2"),
+		},
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 1000; i++ {
+			inst.ObserveFloat64(ctx, 1, attrs...)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 1000; i++ {
+			inst.ObserveFloat64(ctx, 1, attrs...)
+		}
+	}()
+
+	wg.Wait()
+}

pipelines/go.mod

@@ -51,8 +51,8 @@ require (
 )
 
 require (
-	github.com/lightstep/otel-launcher-go/lightstep/instrumentation v1.15.0
-	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.15.0
+	github.com/lightstep/otel-launcher-go/lightstep/instrumentation v1.15.1
+	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.15.1
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.37.0
 )