Repository: GoogleCloudPlatform/opentelemetry-operations-go
Pull request #417 (2022-06-09 15:49:25)

Make the datapoint storage cache safe for concurrent use
Fixes https://github.com/GoogleCloudPlatform/opentelemetry-operations-go/issues/405

This adds one RW lock for each cache.  For GetFoo() functions, it takes a RLock, and for SetFoo() functions, it takes a write Lock.  However, since the "used" bool is set in GetFoo() functions (it indicates if the item has been touched since the last GC), this still produced a data race.  I could have switched to a full write-lock for Get functions, but thought it would be more performant to use a concurrency-safe bool.  That also simplifies the implementation of GC using Compare and swap.

I added tests for concurrency to verify that there are no data races.

dashpole: @jsuereth you had mentioned wanting to review this at standup.

Patch:
diff --git a/exporter/collector/go.mod b/exporter/collector/go.mod
index cf0a20c..b85abb0 100644
--- a/exporter/collector/go.mod
+++ b/exporter/collector/go.mod
@@ -12,6 +12,7 @@ require (
 	go.opentelemetry.io/otel v1.6.3
 	go.opentelemetry.io/otel/sdk v1.6.2
 	go.opentelemetry.io/otel/trace v1.6.3
+	go.uber.org/atomic v1.9.0
 	google.golang.org/api v0.74.0
 	google.golang.org/genproto v0.0.0-20220405205423-9d709892a2bf
 	google.golang.org/grpc v1.46.0
@@ -46,7 +47,6 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	go.uber.org/atomic v1.9.0 // indirect
 	golang.org/x/net v0.0.0-20220325170049-de3da57026de // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886 // indirect
diff --git a/exporter/collector/internal/datapointstorage/datapointcache.go b/exporter/collector/internal/datapointstorage/datapointcache.go
index b40e0e9..ec6c585 100644
--- a/exporter/collector/internal/datapointstorage/datapointcache.go
+++ b/exporter/collector/internal/datapointstorage/datapointcache.go
@@ -17,10 +17,12 @@ package datapointstorage
 import (
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/atomic"
 	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
 )
 
@@ -31,31 +33,35 @@ type Cache struct {
 	summaryCache              map[string]usedSummaryPoint
 	histogramCache            map[string]usedHistogramPoint
 	exponentialHistogramCache map[string]usedExponentialHistogramPoint
+	numberLock                sync.RWMutex
+	summaryLock               sync.RWMutex
+	histogramLock             sync.RWMutex
+	exponentialHistogramLock  sync.RWMutex
 }
 
 type usedNumberPoint struct {
 	point *pmetric.NumberDataPoint
-	used  bool
+	used  *atomic.Bool
 }
 
 type usedSummaryPoint struct {
 	point *pmetric.SummaryDataPoint
-	used  bool
+	used  *atomic.Bool
 }
 
 type usedHistogramPoint struct {
 	point *pmetric.HistogramDataPoint
-	used  bool
+	used  *atomic.Bool
 }
 
 type usedExponentialHistogramPoint struct {
 	point *pmetric.ExponentialHistogramDataPoint
-	used  bool
+	used  *atomic.Bool
 }
 
 // NewCache instantiates a cache and starts background processes
-func NewCache(shutdown <-chan struct{}) Cache {
-	c := Cache{
+func NewCache(shutdown <-chan struct{}) *Cache {
+	c := &Cache{
 		numberCache:               make(map[string]usedNumberPoint),
 		summaryCache:              make(map[string]usedSummaryPoint),
 		histogramCache:            make(map[string]usedHistogramPoint),
@@ -71,118 +77,134 @@ func NewCache(shutdown <-chan struct{}) Cache {
 
 // GetNumberDataPoint retrieves the point associated with the identifier, and whether
 // or not it was found
-func (c Cache) GetNumberDataPoint(identifier string) (*pmetric.NumberDataPoint, bool) {
+func (c *Cache) GetNumberDataPoint(identifier string) (*pmetric.NumberDataPoint, bool) {
+	c.numberLock.RLock()
+	defer c.numberLock.RUnlock()
 	point, found := c.numberCache[identifier]
 	if found {
-		point.used = true
-		c.numberCache[identifier] = point
+		point.used.Store(true)
 	}
 	return point.point, found
 }
 
 // SetNumberDataPoint assigns the point to the identifier in the cache
-func (c Cache) SetNumberDataPoint(identifier string, point *pmetric.NumberDataPoint) {
-	c.numberCache[identifier] = usedNumberPoint{point, true}
+func (c *Cache) SetNumberDataPoint(identifier string, point *pmetric.NumberDataPoint) {
+	c.numberLock.Lock()
+	defer c.numberLock.Unlock()
+	c.numberCache[identifier] = usedNumberPoint{point, atomic.NewBool(true)}
 }
 
 // GetSummaryDataPoint retrieves the point associated with the identifier, and whether
 // or not it was found
-func (c Cache) GetSummaryDataPoint(identifier string) (*pmetric.SummaryDataPoint, bool) {
+func (c *Cache) GetSummaryDataPoint(identifier string) (*pmetric.SummaryDataPoint, bool) {
+	c.summaryLock.RLock()
+	defer c.summaryLock.RUnlock()
 	point, found := c.summaryCache[identifier]
 	if found {
-		point.used = true
-		c.summaryCache[identifier] = point
+		point.used.Store(true)
 	}
 	return point.point, found
 }
 
 // SetSummaryDataPoint assigns the point to the identifier in the cache
-func (c Cache) SetSummaryDataPoint(identifier string, point *pmetric.SummaryDataPoint) {
-	c.summaryCache[identifier] = usedSummaryPoint{point, true}
+func (c *Cache) SetSummaryDataPoint(identifier string, point *pmetric.SummaryDataPoint) {
+	c.summaryLock.Lock()
+	defer c.summaryLock.Unlock()
+	c.summaryCache[identifier] = usedSummaryPoint{point, atomic.NewBool(true)}
 }
 
 // GetHistogramDataPoint retrieves the point associated with the identifier, and whether
 // or not it was found
-func (c Cache) GetHistogramDataPoint(identifier string) (*pmetric.HistogramDataPoint, bool) {
+func (c *Cache) GetHistogramDataPoint(identifier string) (*pmetric.HistogramDataPoint, bool) {
+	c.histogramLock.RLock()
+	defer c.histogramLock.RUnlock()
 	point, found := c.histogramCache[identifier]
 	if found {
-		point.used = true
-		c.histogramCache[identifier] = point
+		point.used.Store(true)
 	}
 	return point.point, found
 }
 
 // SetHistogramDataPoint assigns the point to the identifier in the cache
-func (c Cache) SetHistogramDataPoint(identifier string, point *pmetric.HistogramDataPoint) {
-	c.histogramCache[identifier] = usedHistogramPoint{point, true}
+func (c *Cache) SetHistogramDataPoint(identifier string, point *pmetric.HistogramDataPoint) {
+	c.histogramLock.Lock()
+	defer c.histogramLock.Unlock()
+	c.histogramCache[identifier] = usedHistogramPoint{point, atomic.NewBool(true)}
 }
 
 // GetExponentialHistogramDataPoint retrieves the point associated with the identifier, and whether
 // or not it was found
-func (c Cache) GetExponentialHistogramDataPoint(identifier string) (*pmetric.ExponentialHistogramDataPoint, bool) {
+func (c *Cache) GetExponentialHistogramDataPoint(identifier string) (*pmetric.ExponentialHistogramDataPoint, bool) {
+	c.exponentialHistogramLock.RLock()
+	defer c.exponentialHistogramLock.RUnlock()
 	point, found := c.exponentialHistogramCache[identifier]
 	if found {
-		point.used = true
-		c.exponentialHistogramCache[identifier] = point
+		point.used.Store(true)
 	}
 	return point.point, found
 }
 
 // SetExponentialHistogramDataPoint assigns the point to the identifier in the cache
-func (c Cache) SetExponentialHistogramDataPoint(identifier string, point *pmetric.ExponentialHistogramDataPoint) {
-	c.exponentialHistogramCache[identifier] = usedExponentialHistogramPoint{point, true}
+func (c *Cache) SetExponentialHistogramDataPoint(identifier string, point *pmetric.ExponentialHistogramDataPoint) {
+	c.exponentialHistogramLock.Lock()
+	defer c.exponentialHistogramLock.Unlock()
+	c.exponentialHistogramCache[identifier] = usedExponentialHistogramPoint{point, atomic.NewBool(true)}
 }
 
 // gc garbage collects the cache after the ticker ticks
-func (c Cache) gc(shutdown <-chan struct{}, tickerCh <-chan time.Time) bool {
+func (c *Cache) gc(shutdown <-chan struct{}, tickerCh <-chan time.Time) bool {
 	select {
 	case <-shutdown:
 		return false
 	case <-tickerCh:
 		// garbage collect the numberCache
+		c.numberLock.Lock()
 		for id, point := range c.numberCache {
-			if point.used {
-				// for points that have been used, mark them as unused
-				point.used = false
-				c.numberCache[id] = point
+			// for points that have been used, mark them as unused
+			if point.used.Load() {
+				point.used.Store(false)
 			} else {
 				// for points that have not been used, delete points
 				delete(c.numberCache, id)
 			}
 		}
+		c.numberLock.Unlock()
 		// garbage collect the summaryCache
+		c.summaryLock.Lock()
 		for id, point := range c.summaryCache {
-			if point.used {
-				// for points that have been used, mark them as unused
-				point.used = false
-				c.summaryCache[id] = point
+			// for points that have been used, mark them as unused
+			if point.used.Load() {
+				point.used.Store(false)
 			} else {
 				// for points that have not been used, delete points
 				delete(c.summaryCache, id)
 			}
 		}
+		c.summaryLock.Unlock()
 		// garbage collect the histogramCache
+		c.histogramLock.Lock()
 		for id, point := range c.histogramCache {
-			if point.used {
-				// for points that have been used, mark them as unused
-				point.used = false
-				c.histogramCache[id] = point
+			// for points that have been used, mark them as unused
+			if point.used.Load() {
+				point.used.Store(false)
 			} else {
 				// for points that have not been used, delete points
 				delete(c.histogramCache, id)
 			}
 		}
+		c.histogramLock.Unlock()
 		// garbage collect the exponentialHistogramCache
+		c.exponentialHistogramLock.Lock()
 		for id, point := range c.exponentialHistogramCache {
-			if point.used {
-				// for points that have been used, mark them as unused
-				point.used = false
-				c.exponentialHistogramCache[id] = point
+			// for points that have been used, mark them as unused
+			if point.used.Load() {
+				point.used.Store(false)
 			} else {
 				// for points that have not been used, delete points
 				delete(c.exponentialHistogramCache, id)
 			}
 		}
+		c.exponentialHistogramLock.Unlock()
 	}
 	return true
 }
diff --git a/exporter/collector/internal/normalization/standard_normalizer.go b/exporter/collector/internal/normalization/standard_normalizer.go
index 59b30dd..3a9c372 100644
--- a/exporter/collector/internal/normalization/standard_normalizer.go
+++ b/exporter/collector/internal/normalization/standard_normalizer.go
@@ -40,8 +40,8 @@ func NewStandardNormalizer(shutdown <-chan struct{}, logger *zap.Logger) Normali
 }
 
 type standardNormalizer struct {
-	startCache    datapointstorage.Cache
-	previousCache datapointstorage.Cache
+	startCache    *datapointstorage.Cache
+	previousCache *datapointstorage.Cache
 	log           *zap.Logger
 }
 
diff --git a/exporter/trace/cloudtrace.go b/exporter/trace/cloudtrace.go
index ae4d493..e4810ba 100644
--- a/exporter/trace/cloudtrace.go
+++ b/exporter/trace/cloudtrace.go
@@ -62,21 +62,10 @@ type options struct {
 	// resource if the resource does not inherently belong to a specific
 	// project, e.g. on-premise resource like k8s_container or generic_task.
 	projectID string
-	// location is the identifier of the GCP or AWS cloud region/zone in which
-	// the data for a resource is stored.
-	// If not set, it will default to the location provided by the metadata server.
-	//
-	// It will be used in the location label of a Stackdriver monitored resource
-	// if the resource does not inherently belong to a specific project, e.g.
-	// on-premise resource like k8s_container or generic_task.
-	location string
 	// traceClientOptions are additional options to be passed
 	// to the underlying Stackdriver Trace API client.
 	// Optional.
 	traceClientOptions []option.ClientOption
-	// batchSpanProcessorOptions are additional options to be based
-	// to the underlying BatchSpanProcessor when call making a new export pipeline.
-	batchSpanProcessorOptions []sdktrace.BatchSpanProcessorOption
 	// timeout for all API calls. If not set, defaults to 5 seconds.
 	timeout time.Duration
 }
@@ -206,22 +195,3 @@ func (e *Exporter) ExportSpans(ctx context.Context, spanData []sdktrace.ReadOnly
 func (e *Exporter) Shutdown(ctx context.Context) error {
 	return e.traceExporter.Shutdown(ctx)
 }
-
-func createKeyValueAttributes(attr map[string]interface{}) []attribute.KeyValue {
-	kv := make([]attribute.KeyValue, 0, len(attr))
-
-	for k, v := range attr {
-		switch val := v.(type) {
-		case bool:
-			kv = append(kv, attribute.Bool(k, val))
-		case int64:
-			kv = append(kv, attribute.Int64(k, val))
-		case float64:
-			kv = append(kv, attribute.Float64(k, val))
-		case string:
-			kv = append(kv, attribute.String(k, val))
-		}
-	}
-
-	return kv
-}


Test patch:
diff --git a/exporter/collector/internal/datapointstorage/datapointcache_test.go b/exporter/collector/internal/datapointstorage/datapointcache_test.go
index 192e72f..f5ba824 100644
--- a/exporter/collector/internal/datapointstorage/datapointcache_test.go
+++ b/exporter/collector/internal/datapointstorage/datapointcache_test.go
@@ -15,6 +15,7 @@
 package datapointstorage
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -73,7 +74,7 @@ func TestGC(t *testing.T) {
 
 	// bar exists since we just set it
 	usedPoint, found := c.numberCache["bar"]
-	assert.True(t, usedPoint.used)
+	assert.True(t, usedPoint.used.Load())
 	assert.True(t, found)
 
 	// first gc tick marks bar stale
@@ -83,7 +84,7 @@ func TestGC(t *testing.T) {
 	cont := c.gc(shutdown, fakeTicker)
 	assert.True(t, cont)
 	usedPoint, found = c.numberCache["bar"]
-	assert.False(t, usedPoint.used)
+	assert.False(t, usedPoint.used.Load())
 	assert.True(t, found)
 
 	// second gc tick removes bar
@@ -130,6 +131,130 @@ func TestGetPreventsGC(t *testing.T) {
 	assert.True(t, found)
 }
 
+func TestConcurrentNumber(t *testing.T) {
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
+	setPoint := pmetric.NewNumberDataPoint()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			c.SetNumberDataPoint("bar", &setPoint)
+			point, found := c.GetNumberDataPoint("bar")
+			assert.Equal(t, point, &setPoint)
+			assert.True(t, found)
+			wg.Done()
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		dontShutdown := make(chan struct{})
+		fakeTick := make(chan time.Time)
+		go func() { fakeTick <- time.Now() }()
+		c.gc(dontShutdown, fakeTick)
+		wg.Done()
+	}()
+	wg.Wait()
+}
+
+func TestConcurrentSummary(t *testing.T) {
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
+	setPoint := pmetric.NewSummaryDataPoint()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			c.SetSummaryDataPoint("bar", &setPoint)
+			point, found := c.GetSummaryDataPoint("bar")
+			assert.Equal(t, point, &setPoint)
+			assert.True(t, found)
+			wg.Done()
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		dontShutdown := make(chan struct{})
+		fakeTick := make(chan time.Time)
+		go func() { fakeTick <- time.Now() }()
+		c.gc(dontShutdown, fakeTick)
+		wg.Done()
+	}()
+	wg.Wait()
+}
+
+func TestConcurrentHistogram(t *testing.T) {
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
+	setPoint := pmetric.NewHistogramDataPoint()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			c.SetHistogramDataPoint("bar", &setPoint)
+			point, found := c.GetHistogramDataPoint("bar")
+			assert.Equal(t, point, &setPoint)
+			assert.True(t, found)
+			wg.Done()
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		dontShutdown := make(chan struct{})
+		fakeTick := make(chan time.Time)
+		go func() { fakeTick <- time.Now() }()
+		c.gc(dontShutdown, fakeTick)
+		wg.Done()
+	}()
+	wg.Wait()
+}
+
+func TestConcurrentExponentialHistogram(t *testing.T) {
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
+	setPoint := pmetric.NewExponentialHistogramDataPoint()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			c.SetExponentialHistogramDataPoint("bar", &setPoint)
+			point, found := c.GetExponentialHistogramDataPoint("bar")
+			assert.Equal(t, point, &setPoint)
+			assert.True(t, found)
+			wg.Done()
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		dontShutdown := make(chan struct{})
+		fakeTick := make(chan time.Time)
+		go func() { fakeTick <- time.Now() }()
+		c.gc(dontShutdown, fakeTick)
+		wg.Done()
+	}()
+	wg.Wait()
+}
+
 func TestIdentifier(t *testing.T) {
 	metricWithName := pmetric.NewMetric()
 	metricWithName.SetName("custom.googleapis.com/test.metric")