Repository: GoogleCloudPlatform/opentelemetry-operations-go
Pull request #323 (2022-02-11 17:16:02)

Normalize sums prior to exporting them
This is based on the [normalizesumsprocessor](https://github.com/GoogleCloudPlatform/opentelemetry-operations-collector/tree/master/processor/normalizesumsprocessor), but has a few differences:

* Timeseries are garbage-collected if they are not used after 20 minutes to prevent it from leaking memory
* It converts non-monotonic sums to gauges instead of keeping track of the latest and inserting resets when the value decreases.
* It handles explicit reset points (where start time is after end time).

To implement garbage collection, i've extracted the cache out into its own package.


Patch:
diff --git a/exporter/collector/config.go b/exporter/collector/config.go
index 6f8b8bf..51ef79a 100644
--- a/exporter/collector/config.go
+++ b/exporter/collector/config.go
@@ -15,6 +15,7 @@
 package collector
 
 import (
+	"fmt"
 	"time"
 
 	"google.golang.org/api/option"
@@ -48,6 +49,18 @@ type ClientConfig struct {
 
 type TraceConfig struct {
 	ClientConfig ClientConfig `mapstructure:",squash"`
+	// AttributeMappings determines how to map from OpenTelemetry attribute
+	// keys to Google Cloud Trace keys.  By default, it changes http and
+	// service keys so that they appear more prominently in the UI.
+	AttributeMappings []AttributeMapping `mapstructure:"attribute_mappings"`
+}
+
+// AttributeMapping maps from an OpenTelemetry key to a Google Cloud Trace key.
+type AttributeMapping struct {
+	// Key is the OpenTelemetry attribute key
+	Key string `mapstructure:"key"`
+	// Replacement is the attribute sent to Google Cloud Trace
+	Replacement string `mapstructure:"replacement"`
 }
 
 type MetricConfig struct {
@@ -103,3 +116,20 @@ func DefaultConfig() Config {
 		},
 	}
 }
+
+// ValidateConfig returns an error if the provided configuration is invalid
+func ValidateConfig(cfg Config) error {
+	seenKeys := make(map[string]struct{}, len(cfg.TraceConfig.AttributeMappings))
+	seenReplacements := make(map[string]struct{}, len(cfg.TraceConfig.AttributeMappings))
+	for _, mapping := range cfg.TraceConfig.AttributeMappings {
+		if _, ok := seenKeys[mapping.Key]; ok {
+			return fmt.Errorf("duplicate key in traces.attribute_mappings: %q", mapping.Key)
+		}
+		seenKeys[mapping.Key] = struct{}{}
+		if _, ok := seenReplacements[mapping.Replacement]; ok {
+			return fmt.Errorf("duplicate replacement in traces.attribute_mappings: %q", mapping.Replacement)
+		}
+		seenReplacements[mapping.Replacement] = struct{}{}
+	}
+	return nil
+}
diff --git a/exporter/collector/internal/datapointstorage/datapointcache.go b/exporter/collector/internal/datapointstorage/datapointcache.go
new file mode 100644
index 0000000..c630ea0
--- /dev/null
+++ b/exporter/collector/internal/datapointstorage/datapointcache.go
@@ -0,0 +1,102 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datapointstorage
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"go.opentelemetry.io/collector/model/pdata"
+	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
+)
+
+const gcInterval = 20 * time.Minute
+
+type Cache map[string]usedPoint
+
+type usedPoint struct {
+	point *pdata.NumberDataPoint
+	used  bool
+}
+
+// New instantiates a cache and starts background processes
+func NewCache(shutdown <-chan struct{}) Cache {
+	c := make(Cache)
+	go func() {
+		ticker := time.NewTicker(gcInterval)
+		for c.gc(shutdown, ticker.C) {
+		}
+	}()
+	return c
+}
+
+// Get retrieves the point associated with the identifier, and whether
+// or not it was found
+func (c Cache) Get(identifier string) (*pdata.NumberDataPoint, bool) {
+	point, found := c[identifier]
+	if found {
+		point.used = true
+		c[identifier] = point
+	}
+	return point.point, found
+}
+
+// Set assigns the point to the identifier in the cache
+func (c Cache) Set(identifier string, point *pdata.NumberDataPoint) {
+	c[identifier] = usedPoint{point, true}
+}
+
+// gc garbage collects the cache after the ticker ticks
+func (c Cache) gc(shutdown <-chan struct{}, tickerCh <-chan time.Time) bool {
+	select {
+	case <-shutdown:
+		return false
+	case <-tickerCh:
+		// garbage collect the cache
+		for id, point := range c {
+			if point.used {
+				// for points that have been used, mark them as unused
+				point.used = false
+				c[id] = point
+			} else {
+				// for points that have not been used, delete points
+				delete(c, id)
+			}
+		}
+	}
+	return true
+}
+
+// Identifier returns the unique string identifier for a metric
+func Identifier(resource *monitoredrespb.MonitoredResource, extraLabels map[string]string, metric pdata.Metric, labels pdata.AttributeMap) string {
+	var b strings.Builder
+
+	// Resource identifiers
+	if resource != nil {
+		fmt.Fprintf(&b, "%v", resource.GetLabels())
+	}
+
+	// Instrumentation library labels and additional resource labels
+	fmt.Fprintf(&b, " - %v", extraLabels)
+
+	// Metric identifiers
+	fmt.Fprintf(&b, " - %s -", metric.Name())
+	labels.Sort().Range(func(k string, v pdata.AttributeValue) bool {
+		fmt.Fprintf(&b, " %s=%s", k, v.AsString())
+		return true
+	})
+	return b.String()
+}
diff --git a/exporter/collector/metrics.go b/exporter/collector/metrics.go
index 2095bb3..b663aa4 100644
--- a/exporter/collector/metrics.go
+++ b/exporter/collector/metrics.go
@@ -42,6 +42,8 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/datapointstorage"
 )
 
 // self-observability reporting meters/tracers/loggers.
@@ -77,8 +79,9 @@ type MetricsExporter struct {
 // metricMapper is the part that transforms metrics. Separate from MetricsExporter since it has
 // all pure functions.
 type metricMapper struct {
-	obs selfObservability
-	cfg Config
+	obs      selfObservability
+	cfg      Config
+	sumCache datapointstorage.Cache
 }
 
 // Constants we use when translating summary metrics into GCP.
@@ -139,11 +142,16 @@ func NewGoogleCloudMetricsExporter(
 		return nil, err
 	}
 	obs := selfObservability{log: log}
+	shutdown := make(chan struct{})
 	mExp := &MetricsExporter{
 		cfg:    cfg,
 		client: client,
 		obs:    obs,
-		mapper: metricMapper{obs, cfg},
+		mapper: metricMapper{
+			obs,
+			cfg,
+			datapointstorage.NewCache(shutdown),
+		},
 		// We create a buffered channel for metric descriptors.
 		// MetricDescritpors are asychronously sent and optimistic.
 		// We only get Unit/Description/Display name from them, so it's ok
@@ -151,7 +159,7 @@ func NewGoogleCloudMetricsExporter(
 		metricDescriptorC: make(chan *metricpb.MetricDescriptor, cfg.MetricConfig.CreateMetricDescriptorBufferSize),
 		mdCache:           make(map[string]*metricpb.MetricDescriptor),
 		timeSeriesC:       make(chan *monitoringpb.TimeSeries),
-		shutdownC:         make(chan struct{}),
+		shutdownC:         shutdown,
 	}
 
 	// Fire up the metric descriptor exporter.
@@ -334,7 +342,7 @@ func (m *metricMapper) metricToTimeSeries(
 		points := sum.DataPoints()
 		for i := 0; i < points.Len(); i++ {
 			ts := m.sumPointToTimeSeries(resource, extraLabels, metric, sum, points.At(i))
-			timeSeries = append(timeSeries, ts)
+			timeSeries = append(timeSeries, ts...)
 		}
 	case pdata.MetricDataTypeGauge:
 		gauge := metric.Gauge()
@@ -659,16 +667,24 @@ func (m *metricMapper) sumPointToTimeSeries(
 	metric pdata.Metric,
 	sum pdata.Sum,
 	point pdata.NumberDataPoint,
-) *monitoringpb.TimeSeries {
+) []*monitoringpb.TimeSeries {
 	metricKind := metricpb.MetricDescriptor_CUMULATIVE
-	startTime := timestamppb.New(point.StartTimestamp().AsTime())
-	if !sum.IsMonotonic() {
+	var startTime *timestamppb.Timestamp
+	if sum.IsMonotonic() {
+		metricIdentifier := datapointstorage.Identifier(resource, extraLabels, metric, point.Attributes())
+		normalizedPoint := m.normalizeNumberDataPoint(point, metricIdentifier)
+		if normalizedPoint == nil {
+			return nil
+		}
+		point = *normalizedPoint
+		startTime = timestamppb.New(normalizedPoint.StartTimestamp().AsTime())
+	} else {
 		metricKind = metricpb.MetricDescriptor_GAUGE
 		startTime = nil
 	}
 	value, valueType := numberDataPointToValue(point)
 
-	return &monitoringpb.TimeSeries{
+	return []*monitoringpb.TimeSeries{{
 		Resource:   resource,
 		Unit:       metric.Unit(),
 		MetricKind: metricKind,
@@ -687,7 +703,49 @@ func (m *metricMapper) sumPointToTimeSeries(
 				extraLabels,
 			),
 		},
+	}}
+}
+
+// normalizeNumberDataPoint returns a point which has been normalized against an initial
+// start point, or nil if the point should be dropped.
+func (m *metricMapper) normalizeNumberDataPoint(point pdata.NumberDataPoint, identifier string) *pdata.NumberDataPoint {
+	// if the point doesn't need to be normalized, use original point
+	normalizedPoint := &point
+	start, ok := m.sumCache.Get(identifier)
+	if ok {
+		if !start.StartTimestamp().AsTime().Before(point.Timestamp().AsTime()) {
+			// We found a cached start timestamp that wouldn't produce a valid point.
+			// Drop it and log.
+			m.obs.log.Info(
+				"data point being processed older than last recorded reset, will not be emitted",
+				zap.String("lastRecordedReset", start.Timestamp().String()),
+				zap.String("dataPoint", point.Timestamp().String()),
+			)
+			return nil
+		}
+		// Make a copy so we don't mutate underlying data
+		newPoint := pdata.NewNumberDataPoint()
+		point.CopyTo(newPoint)
+		// Use the start timestamp from the normalization point
+		newPoint.SetStartTimestamp(start.Timestamp())
+		// Adjust the value based on the start point's value
+		switch newPoint.ValueType() {
+		case pdata.MetricValueTypeInt:
+			newPoint.SetIntVal(point.IntVal() - start.IntVal())
+		case pdata.MetricValueTypeDouble:
+			newPoint.SetDoubleVal(point.DoubleVal() - start.DoubleVal())
+		}
+		normalizedPoint = &newPoint
+	}
+	if (!ok && point.StartTimestamp().AsTime().IsZero()) || !point.StartTimestamp().AsTime().Before(point.Timestamp().AsTime()) {
+		// This is the first time we've seen this metric, or we received
+		// an explicit reset point as described in
+		// https://github.com/open-telemetry/opentelemetry-specification/blob/9555f9594c7ffe5dc333b53da5e0f880026cead1/specification/metrics/datamodel.md#resets-and-gaps
+		// Record it in history and drop the point.
+		m.sumCache.Set(identifier, &point)
+		return nil
 	}
+	return normalizedPoint
 }
 
 func (m *metricMapper) gaugePointToTimeSeries(
@@ -818,7 +876,7 @@ func (m *metricMapper) labelDescriptors(
 	result := []*label.LabelDescriptor{}
 	for key := range extraLabels {
 		result = append(result, &label.LabelDescriptor{
-			Key: key,
+			Key: sanitizeKey(key),
 		})
 	}
 
@@ -826,13 +884,13 @@ func (m *metricMapper) labelDescriptors(
 	addAttributes := func(attr pdata.AttributeMap) {
 		attr.Range(func(key string, _ pdata.AttributeValue) bool {
 			// Skip keys that have already been set
-			if _, ok := seenKeys[key]; ok {
+			if _, ok := seenKeys[sanitizeKey(key)]; ok {
 				return true
 			}
 			result = append(result, &label.LabelDescriptor{
-				Key: key,
+				Key: sanitizeKey(key),
 			})
-			seenKeys[key] = struct{}{}
+			seenKeys[sanitizeKey(key)] = struct{}{}
 			return true
 		})
 	}
diff --git a/exporter/collector/traces.go b/exporter/collector/traces.go
index 84b3955..a626de7 100644
--- a/exporter/collector/traces.go
+++ b/exporter/collector/traces.go
@@ -27,6 +27,7 @@ import (
 	"go.opencensus.io/plugin/ocgrpc"
 	"go.opencensus.io/stats/view"
 	"go.opentelemetry.io/collector/model/pdata"
+	"go.opentelemetry.io/otel/attribute"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"golang.org/x/oauth2/google"
 	"google.golang.org/api/option"
@@ -104,6 +105,9 @@ func NewGoogleCloudTracesExporter(ctx context.Context, cfg Config, version strin
 		cloudtrace.WithProjectID(cfg.ProjectID),
 		cloudtrace.WithTimeout(timeout),
 	}
+	if cfg.TraceConfig.AttributeMappings != nil {
+		topts = append(topts, cloudtrace.WithAttributeMapping(mappingFuncFromAKM(cfg.TraceConfig.AttributeMappings)))
+	}
 
 	copts, err := generateClientOptions(&cfg.TraceConfig.ClientConfig, cfg.UserAgent)
 	if err != nil {
@@ -119,6 +123,22 @@ func NewGoogleCloudTracesExporter(ctx context.Context, cfg Config, version strin
 	return &TraceExporter{texporter: exp}, nil
 }
 
+func mappingFuncFromAKM(akm []AttributeMapping) func(attribute.Key) attribute.Key {
+	// convert list to map for easy lookups
+	mapFromConfig := make(map[string]string, len(akm))
+	for _, mapping := range akm {
+		mapFromConfig[mapping.Key] = mapping.Replacement
+	}
+	return func(input attribute.Key) attribute.Key {
+		// if a replacement was specified in the config, use it.
+		if replacement, ok := mapFromConfig[string(input)]; ok {
+			return attribute.Key(replacement)
+		}
+		// otherwise, leave the attribute as-is
+		return input
+	}
+}
+
 // PushTraces calls texporter.ExportSpan for each span in the given traces
 func (te *TraceExporter) PushTraces(ctx context.Context, td pdata.Traces) error {
 	resourceSpans := td.ResourceSpans()
diff --git a/exporter/trace/cloudtrace.go b/exporter/trace/cloudtrace.go
index 051c966..3a77f0a 100644
--- a/exporter/trace/cloudtrace.go
+++ b/exporter/trace/cloudtrace.go
@@ -83,6 +83,9 @@ type options struct {
 
 	// Timeout for all API calls. If not set, defaults to 5 seconds.
 	Timeout time.Duration
+
+	// mapAttribute maps otel attribute keys to cloud trace attribute keys
+	mapAttribute AttributeMapping
 }
 
 // WithProjectID sets Google Cloud Platform project as projectID.
@@ -127,6 +130,19 @@ func WithTimeout(t time.Duration) func(o *options) {
 	}
 }
 
+// AttributeMapping determines how to map from OpenTelemetry span attribute keys to
+// cloud trace attribute keys.
+type AttributeMapping func(attribute.Key) attribute.Key
+
+// WithAttributeMapping configures how to map OpenTelemetry span attributes
+// to google cloud trace span attributes.  By default, it maps to attributes
+// that are used prominently in the trace UI.
+func WithAttributeMapping(mapping AttributeMapping) func(o *options) {
+	return func(o *options) {
+		o.mapAttribute = mapping
+	}
+}
+
 func (o *options) handleError(err error) {
 	if o.errorHandler != nil {
 		o.errorHandler.Handle(err)
@@ -148,7 +164,10 @@ type Exporter struct {
 
 // New creates a new Exporter thats implements trace.Exporter.
 func New(opts ...Option) (*Exporter, error) {
-	o := options{Context: context.Background()}
+	o := options{
+		Context:      context.Background(),
+		mapAttribute: defaultAttributeMapping,
+	}
 	for _, opt := range opts {
 		opt(&o)
 	}
diff --git a/exporter/trace/trace.go b/exporter/trace/trace.go
index e9b007c..f8a9eb7 100644
--- a/exporter/trace/trace.go
+++ b/exporter/trace/trace.go
@@ -65,7 +65,7 @@ func (e *traceExporter) ExportSpans(ctx context.Context, spanData []sdktrace.Rea
 
 // ConvertSpan converts a ReadOnlySpan to Stackdriver Trace.
 func (e *traceExporter) ConvertSpan(_ context.Context, sd sdktrace.ReadOnlySpan) *tracepb.Span {
-	return protoFromReadOnlySpan(sd, e.projectID)
+	return e.protoFromReadOnlySpan(sd, e.projectID)
 }
 
 func (e *traceExporter) Shutdown(ctx context.Context) error {
diff --git a/exporter/trace/trace_proto.go b/exporter/trace/trace_proto.go
index 214b7e2..76fbb48 100644
--- a/exporter/trace/trace_proto.go
+++ b/exporter/trace/trace_proto.go
@@ -87,7 +87,7 @@ func attributeWithLabelsFromResources(sd sdktrace.ReadOnlySpan) []attribute.KeyV
 	return attributes
 }
 
-func protoFromReadOnlySpan(s sdktrace.ReadOnlySpan, projectID string) *tracepb.Span {
+func (e *traceExporter) protoFromReadOnlySpan(s sdktrace.ReadOnlySpan, projectID string) *tracepb.Span {
 	if s == nil {
 		return nil
 	}
@@ -118,20 +118,20 @@ func protoFromReadOnlySpan(s sdktrace.ReadOnlySpan, projectID string) *tracepb.S
 	}
 
 	attributes := attributeWithLabelsFromResources(s)
-	copyAttributes(&sp.Attributes, attributes)
+	e.copyAttributes(&sp.Attributes, attributes)
 	// NOTE(ymotongpoo): omitting copyMonitoringReesourceAttributes()
 
 	var annotations, droppedAnnotationsCount int
 	es := s.Events()
-	for i, e := range es {
+	for i, ev := range es {
 		if annotations >= maxAnnotationEventsPerSpan {
 			droppedAnnotationsCount = len(es) - i
 			break
 		}
-		annotation := &tracepb.Span_TimeEvent_Annotation{Description: trunc(e.Name, maxAttributeStringValue)}
-		copyAttributes(&annotation.Attributes, e.Attributes)
+		annotation := &tracepb.Span_TimeEvent_Annotation{Description: trunc(ev.Name, maxAttributeStringValue)}
+		e.copyAttributes(&annotation.Attributes, ev.Attributes)
 		event := &tracepb.Span_TimeEvent{
-			Time:  timestampProto(e.Time),
+			Time:  timestampProto(ev.Time),
 			Value: &tracepb.Span_TimeEvent_Annotation_{Annotation: annotation},
 		}
 		annotations++
@@ -171,7 +171,7 @@ func protoFromReadOnlySpan(s sdktrace.ReadOnlySpan, projectID string) *tracepb.S
 		sp.TimeEvents.DroppedAnnotationsCount = clip32(droppedAnnotationsCount)
 	}
 
-	sp.Links = linksProtoFromLinks(s.Links())
+	sp.Links = e.linksProtoFromLinks(s.Links())
 
 	return sp
 }
@@ -179,7 +179,7 @@ func protoFromReadOnlySpan(s sdktrace.ReadOnlySpan, projectID string) *tracepb.S
 // Converts OTel span links to Cloud Trace links proto in order. If there are
 // more than maxNumLinks links, the first maxNumLinks will be taken and the rest
 // dropped.
-func linksProtoFromLinks(links []sdktrace.Link) *tracepb.Span_Links {
+func (e *traceExporter) linksProtoFromLinks(links []sdktrace.Link) *tracepb.Span_Links {
 	numLinks := len(links)
 	if numLinks == 0 {
 		return nil
@@ -197,7 +197,7 @@ func linksProtoFromLinks(links []sdktrace.Link) *tracepb.Span_Links {
 			SpanId:  link.SpanContext.SpanID().String(),
 			Type:    tracepb.Span_Link_TYPE_UNSPECIFIED,
 		}
-		copyAttributes(&linkPb.Attributes, link.Attributes)
+		e.copyAttributes(&linkPb.Attributes, link.Attributes)
 		linksPb.Link = append(linksPb.Link, linkPb)
 	}
 	linksPb.DroppedLinksCount = clip32(numLinks - numLinksToKeep)
@@ -215,7 +215,7 @@ func timestampProto(t time.Time) *timestamppb.Timestamp {
 
 // copyAttributes copies a map of attributes to a proto map field.
 // It creates the map if it is nil.
-func copyAttributes(out **tracepb.Span_Attributes, in []attribute.KeyValue) {
+func (e *traceExporter) copyAttributes(out **tracepb.Span_Attributes, in []attribute.KeyValue) {
 	if len(in) == 0 {
 		return
 	}
@@ -231,30 +231,36 @@ func copyAttributes(out **tracepb.Span_Attributes, in []attribute.KeyValue) {
 		if av == nil {
 			continue
 		}
-		switch kv.Key {
-		case pathAttribute:
-			(*out).AttributeMap[labelHTTPPath] = av
-		case hostAttribute:
-			(*out).AttributeMap[labelHTTPHost] = av
-		case methodAttribute:
-			(*out).AttributeMap[labelHTTPMethod] = av
-		case userAgentAttribute:
-			(*out).AttributeMap[labelHTTPUserAgent] = av
-		case statusCodeAttribute:
-			(*out).AttributeMap[labelHTTPStatusCode] = av
-		case serviceAttribute:
-			(*out).AttributeMap[labelService] = av
-		default:
-			if len(kv.Key) > 128 {
-				dropped++
-				continue
-			}
-			(*out).AttributeMap[string(kv.Key)] = av
+		key := e.o.mapAttribute(kv.Key)
+		if len(key) > 128 {
+			dropped++
+			continue
 		}
+		(*out).AttributeMap[string(key)] = av
 	}
 	(*out).DroppedAttributesCount = dropped
 }
 
+// defaultAttributeMapping maps attributes to trace attributes which are
+// used by cloud trace for prominent UI functions, and keeps all others.
+func defaultAttributeMapping(k attribute.Key) attribute.Key {
+	switch k {
+	case pathAttribute:
+		return labelHTTPPath
+	case hostAttribute:
+		return labelHTTPHost
+	case methodAttribute:
+		return labelHTTPMethod
+	case userAgentAttribute:
+		return labelHTTPUserAgent
+	case statusCodeAttribute:
+		return labelHTTPStatusCode
+	case serviceAttribute:
+		return labelService
+	}
+	return k
+}
+
 func attributeValue(keyValue attribute.KeyValue) *tracepb.AttributeValue {
 	v := keyValue.Value
 	switch v.Type() {


Test patch:
diff --git a/exporter/collector/config_test.go b/exporter/collector/config_test.go
new file mode 100644
index 0000000..fdd60af
--- /dev/null
+++ b/exporter/collector/config_test.go
@@ -0,0 +1,77 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package collector
+
+import "testing"
+
+func TestValidateConfig(t *testing.T) {
+	for _, tc := range []struct {
+		desc        string
+		input       Config
+		expectedErr bool
+	}{
+		{
+			desc:  "Empty",
+			input: Config{},
+		},
+		{
+			desc:  "Default",
+			input: DefaultConfig(),
+		},
+		{
+			desc: "Duplicate attribute keys",
+			input: Config{
+				TraceConfig: TraceConfig{
+					AttributeMappings: []AttributeMapping{
+						{
+							Key:         "foo",
+							Replacement: "bar",
+						},
+						{
+							Key:         "foo",
+							Replacement: "baz",
+						},
+					},
+				},
+			},
+			expectedErr: true,
+		},
+		{
+			desc: "Duplicate attribute replacements",
+			input: Config{
+				TraceConfig: TraceConfig{
+					AttributeMappings: []AttributeMapping{
+						{
+							Key:         "key1",
+							Replacement: "same",
+						},
+						{
+							Key:         "key2",
+							Replacement: "same",
+						},
+					},
+				},
+			},
+			expectedErr: true,
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			err := ValidateConfig(tc.input)
+			if (err != nil && !tc.expectedErr) || (err == nil && tc.expectedErr) {
+				t.Errorf("ValidateConfig(%v) = %v; want no error", tc.input, err)
+			}
+		})
+	}
+}
diff --git a/exporter/collector/internal/datapointstorage/datapointcache_test.go b/exporter/collector/internal/datapointstorage/datapointcache_test.go
new file mode 100644
index 0000000..2b79566
--- /dev/null
+++ b/exporter/collector/internal/datapointstorage/datapointcache_test.go
@@ -0,0 +1,192 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package datapointstorage
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/model/pdata"
+	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
+)
+
+func TestSetAndGet(t *testing.T) {
+	c := make(Cache)
+	c.Set("foo", nil)
+	point, found := c.Get("foo")
+	assert.Nil(t, point)
+	assert.True(t, found)
+
+	point, found = c.Get("bar")
+	assert.Nil(t, point)
+	assert.False(t, found)
+
+	setPoint := pdata.NewNumberDataPoint()
+	c.Set("bar", &setPoint)
+
+	point, found = c.Get("bar")
+	assert.Equal(t, point, &setPoint)
+	assert.True(t, found)
+}
+
+func TestShutdown(t *testing.T) {
+	shutdown := make(chan struct{})
+	c := make(Cache)
+	close(shutdown)
+	// gc should return after shutdown is closed
+	cont := c.gc(shutdown, make(chan time.Time))
+	assert.False(t, cont)
+}
+
+func TestGC(t *testing.T) {
+	shutdown := make(chan struct{})
+	c := make(Cache)
+	fakeTicker := make(chan time.Time)
+
+	c.Set("bar", nil)
+
+	// bar exists since we just set it
+	usedPoint, found := c["bar"]
+	assert.True(t, usedPoint.used)
+	assert.True(t, found)
+
+	// first gc tick marks bar stale
+	go func() {
+		fakeTicker <- time.Now()
+	}()
+	cont := c.gc(shutdown, fakeTicker)
+	assert.True(t, cont)
+	usedPoint, found = c["bar"]
+	assert.False(t, usedPoint.used)
+	assert.True(t, found)
+
+	// second gc tick removes bar
+	go func() {
+		fakeTicker <- time.Now()
+	}()
+	cont = c.gc(shutdown, fakeTicker)
+	assert.True(t, cont)
+	_, found = c["bar"]
+	assert.False(t, found)
+}
+
+func TestGetPreventsGC(t *testing.T) {
+	shutdown := make(chan struct{})
+	c := make(Cache)
+	fakeTicker := make(chan time.Time)
+
+	setPoint := pdata.NewNumberDataPoint()
+	c.Set("bar", &setPoint)
+
+	// bar exists since we just set it
+	_, found := c["bar"]
+	assert.True(t, found)
+
+	// first gc tick marks bar stale
+	go func() {
+		fakeTicker <- time.Now()
+	}()
+	cont := c.gc(shutdown, fakeTicker)
+	assert.True(t, cont)
+	// calling Get() marks it fresh again.
+	_, found = c.Get("bar")
+	assert.True(t, found)
+
+	// second gc tick does not remove bar
+	go func() {
+		fakeTicker <- time.Now()
+	}()
+	cont = c.gc(shutdown, fakeTicker)
+	assert.True(t, cont)
+	_, found = c["bar"]
+	assert.True(t, found)
+}
+
+func TestIdentifier(t *testing.T) {
+	metricWithName := pdata.NewMetric()
+	metricWithName.SetName("custom.googleapis.com/test.metric")
+	dpWithAttributes := pdata.NewNumberDataPoint()
+	dpWithAttributes.Attributes().Insert("string", pdata.NewAttributeValueString("strval"))
+	dpWithAttributes.Attributes().Insert("bool", pdata.NewAttributeValueBool(true))
+	dpWithAttributes.Attributes().Insert("int", pdata.NewAttributeValueInt(123))
+	monitoredResource := &monitoredrespb.MonitoredResource{
+		Type: "generic_task",
+		Labels: map[string]string{
+			"location": "us-central1-b",
+			"project":  "project-foo",
+		},
+	}
+	extraLabels := map[string]string{
+		"foo":   "bar",
+		"hello": "world",
+	}
+	for _, tc := range []struct {
+		desc        string
+		want        string
+		resource    *monitoredrespb.MonitoredResource
+		extraLabels map[string]string
+		metric      pdata.Metric
+		labels      pdata.AttributeMap
+	}{
+		{
+			desc:   "empty",
+			want:   " - map[] -  -",
+			metric: pdata.NewMetric(),
+			labels: pdata.NewNumberDataPoint().Attributes(),
+		},
+		{
+			desc:   "with name",
+			want:   " - map[] - custom.googleapis.com/test.metric -",
+			metric: metricWithName,
+			labels: pdata.NewNumberDataPoint().Attributes(),
+		},
+		{
+			desc:   "with attributes",
+			want:   " - map[] -  - bool=true int=123 string=strval",
+			metric: pdata.NewMetric(),
+			labels: dpWithAttributes.Attributes(),
+		},
+		{
+			desc:     "with resource",
+			want:     "map[location:us-central1-b project:project-foo] - map[] -  -",
+			resource: monitoredResource,
+			metric:   pdata.NewMetric(),
+			labels:   pdata.NewNumberDataPoint().Attributes(),
+		},
+		{
+			desc:        "with extra labels",
+			want:        " - map[foo:bar hello:world] -  -",
+			metric:      pdata.NewMetric(),
+			labels:      pdata.NewNumberDataPoint().Attributes(),
+			extraLabels: extraLabels,
+		},
+		{
+			desc:        "with all",
+			want:        "map[location:us-central1-b project:project-foo] - map[foo:bar hello:world] - custom.googleapis.com/test.metric - bool=true int=123 string=strval",
+			metric:      metricWithName,
+			labels:      dpWithAttributes.Attributes(),
+			extraLabels: extraLabels,
+			resource:    monitoredResource,
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			got := Identifier(tc.resource, tc.extraLabels, tc.metric, tc.labels)
+			if tc.want != got {
+				t.Errorf("Identifier() = %q; want %q", got, tc.want)
+			}
+		})
+	}
+}
diff --git a/exporter/collector/metrics_test.go b/exporter/collector/metrics_test.go
index c1bf2d2..dc86465 100644
--- a/exporter/collector/metrics_test.go
+++ b/exporter/collector/metrics_test.go
@@ -30,31 +30,50 @@ import (
 	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
 	"google.golang.org/protobuf/testing/protocmp"
 	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/datapointstorage"
 )
 
 var (
 	start = time.Unix(0, 0)
 )
 
-func newTestMetricMapper() metricMapper {
+func newTestMetricMapper() (metricMapper, func()) {
 	obs := selfObservability{log: zap.NewNop()}
-	return metricMapper{obs, DefaultConfig()}
+	s := make(chan struct{})
+	return metricMapper{
+		obs,
+		DefaultConfig(),
+		datapointstorage.NewCache(s),
+	}, func() { close(s) }
 }
 
 func TestMetricToTimeSeries(t *testing.T) {
 	mr := &monitoredrespb.MonitoredResource{}
 
 	t.Run("Sum", func(t *testing.T) {
-		mapper := newTestMetricMapper()
+		mapper, shutdown := newTestMetricMapper()
+		defer shutdown()
 		metric := pdata.NewMetric()
 		metric.SetDataType(pdata.MetricDataTypeSum)
 		sum := metric.Sum()
 		sum.SetIsMonotonic(true)
 		sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityCumulative)
+		startTs := pdata.NewTimestampFromTime(start)
+		endTs := pdata.NewTimestampFromTime(start.Add(time.Hour))
 		// Add three points
-		sum.DataPoints().AppendEmpty().SetDoubleVal(10)
-		sum.DataPoints().AppendEmpty().SetDoubleVal(15)
-		sum.DataPoints().AppendEmpty().SetDoubleVal(16)
+		point := sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(10)
+		point.SetStartTimestamp(startTs)
+		point.SetTimestamp(endTs)
+		point = sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(15)
+		point.SetStartTimestamp(startTs)
+		point.SetTimestamp(endTs)
+		point = sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(16)
+		point.SetStartTimestamp(startTs)
+		point.SetTimestamp(endTs)
 
 		ts := mapper.metricToTimeSeries(
 			mr,
@@ -65,8 +84,79 @@ func TestMetricToTimeSeries(t *testing.T) {
 		require.Same(t, ts[0].Resource, mr, "Should assign the passed in monitored resource")
 	})
 
+	t.Run("Sum without timestamps", func(t *testing.T) {
+		mapper, shutdown := newTestMetricMapper()
+		defer shutdown()
+		metric := pdata.NewMetric()
+		metric.SetDataType(pdata.MetricDataTypeSum)
+		sum := metric.Sum()
+		sum.SetIsMonotonic(true)
+		sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityCumulative)
+		startTs := pdata.NewTimestampFromTime(start)
+		endTs := pdata.NewTimestampFromTime(start.Add(time.Hour))
+		// Add three points without start timestamps.
+		// The first one should be dropped to set the start timestamp for the rest
+		point := sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(10)
+		point.SetTimestamp(startTs)
+		point = sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(15)
+		point.SetTimestamp(endTs)
+		point = sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(16)
+		point.SetTimestamp(endTs)
+
+		ts := mapper.metricToTimeSeries(
+			mr,
+			labels{},
+			metric,
+		)
+		require.Len(t, ts, 2, "Should create one timeseries for each sum point")
+		require.Same(t, ts[0].Resource, mr, "Should assign the passed in monitored resource")
+		require.Equal(t, ts[0].Points[0].Value.GetDoubleValue(), 5.0, "Should normalize the resulting sum")
+		require.Equal(t, ts[0].Points[0].Interval.StartTime, timestamppb.New(start), "Should use the first timestamp as the start time for the rest of the points")
+	})
+
+	t.Run("Sum with reset timestamp", func(t *testing.T) {
+		mapper, shutdown := newTestMetricMapper()
+		defer shutdown()
+		metric := pdata.NewMetric()
+		metric.SetDataType(pdata.MetricDataTypeSum)
+		sum := metric.Sum()
+		sum.SetIsMonotonic(true)
+		sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityCumulative)
+		startTs := pdata.NewTimestampFromTime(start)
+		middleTs := pdata.NewTimestampFromTime(start.Add(30 * time.Minute))
+		endTs := pdata.NewTimestampFromTime(start.Add(time.Hour))
+		// Add three points
+		point := sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(10)
+		point.SetStartTimestamp(startTs)
+		point.SetTimestamp(middleTs)
+		point = sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(15)
+		// identical start and end indicates a reset point
+		// this point is expected to be dropped, and to normalize subsequent points
+		point.SetStartTimestamp(middleTs)
+		point.SetTimestamp(middleTs)
+		point = sum.DataPoints().AppendEmpty()
+		point.SetDoubleVal(16)
+		point.SetStartTimestamp(middleTs)
+		point.SetTimestamp(endTs)
+
+		ts := mapper.metricToTimeSeries(
+			mr,
+			labels{},
+			metric,
+		)
+		require.Len(t, ts, 2, "Should create one timeseries for each sum point")
+		require.Same(t, ts[0].Resource, mr, "Should assign the passed in monitored resource")
+		require.Equal(t, ts[1].Points[0].Value.GetDoubleValue(), 1.0, "Should normalize the point after the reset")
+	})
+
 	t.Run("Gauge", func(t *testing.T) {
-		mapper := newTestMetricMapper()
+		mapper, shutdown := newTestMetricMapper()
+		defer shutdown()
 		metric := pdata.NewMetric()
 		metric.SetDataType(pdata.MetricDataTypeGauge)
 		gauge := metric.Gauge()
@@ -104,7 +194,8 @@ func TestMergeLabels(t *testing.T) {
 }
 
 func TestHistogramPointToTimeSeries(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	mapper.cfg.ProjectID = "myproject"
 	mr := &monitoredrespb.MonitoredResource{}
 	metric := pdata.NewMetric()
@@ -167,7 +258,8 @@ func TestHistogramPointToTimeSeries(t *testing.T) {
 }
 
 func TestExponentialHistogramPointToTimeSeries(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	mapper.cfg.ProjectID = "myproject"
 	mr := &monitoredrespb.MonitoredResource{}
 	metric := pdata.NewMetric()
@@ -234,7 +326,8 @@ func TestExponentialHistogramPointToTimeSeries(t *testing.T) {
 }
 
 func TestExemplarNoAttachements(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	exemplar := pdata.NewExemplar()
 	exemplar.SetTimestamp(pdata.NewTimestampFromTime(start))
 	exemplar.SetDoubleVal(1)
@@ -246,7 +339,8 @@ func TestExemplarNoAttachements(t *testing.T) {
 }
 
 func TestExemplarOnlyDroppedLabels(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	exemplar := pdata.NewExemplar()
 	exemplar.SetTimestamp(pdata.NewTimestampFromTime(start))
 	exemplar.SetDoubleVal(1)
@@ -264,7 +358,8 @@ func TestExemplarOnlyDroppedLabels(t *testing.T) {
 }
 
 func TestExemplarOnlyTraceId(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	mapper.cfg.ProjectID = "p"
 	exemplar := pdata.NewExemplar()
 	exemplar.SetTimestamp(pdata.NewTimestampFromTime(start))
@@ -288,7 +383,8 @@ func TestExemplarOnlyTraceId(t *testing.T) {
 }
 
 func TestSumPointToTimeSeries(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	mr := &monitoredrespb.MonitoredResource{}
 
 	newCase := func() (pdata.Metric, pdata.Sum, pdata.NumberDataPoint) {
@@ -312,7 +408,9 @@ func TestSumPointToTimeSeries(t *testing.T) {
 		point.SetStartTimestamp(pdata.NewTimestampFromTime(start))
 		point.SetTimestamp(pdata.NewTimestampFromTime(end))
 
-		ts := mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		tsl := mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		assert.Equal(t, 1, len(tsl))
+		ts := tsl[0]
 		assert.Equal(t, ts.MetricKind, metricpb.MetricDescriptor_CUMULATIVE)
 		assert.Equal(t, ts.ValueType, metricpb.MetricDescriptor_INT64)
 		assert.Equal(t, ts.Unit, unit)
@@ -332,7 +430,9 @@ func TestSumPointToTimeSeries(t *testing.T) {
 
 		// Test double as well
 		point.SetDoubleVal(float64(value))
-		ts = mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		tsl = mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		assert.Equal(t, 1, len(tsl))
+		ts = tsl[0]
 		assert.Equal(t, ts.MetricKind, metricpb.MetricDescriptor_CUMULATIVE)
 		assert.Equal(t, ts.ValueType, metricpb.MetricDescriptor_DOUBLE)
 		assert.Equal(t, ts.Points[0].Value.GetDoubleValue(), float64(value))
@@ -349,7 +449,9 @@ func TestSumPointToTimeSeries(t *testing.T) {
 		point.SetTimestamp(pdata.NewTimestampFromTime(end))
 
 		// Should output a "pseudo-cumulative" with same interval as the delta
-		ts := mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		tsl := mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		assert.Equal(t, 1, len(tsl))
+		ts := tsl[0]
 		assert.Equal(t, ts.MetricKind, metricpb.MetricDescriptor_CUMULATIVE)
 		assert.Equal(t, ts.Points[0].Interval, &monitoringpb.TimeInterval{
 			StartTime: timestamppb.New(start),
@@ -368,14 +470,18 @@ func TestSumPointToTimeSeries(t *testing.T) {
 		point.SetTimestamp(pdata.NewTimestampFromTime(end))
 
 		// Should output a gauge regardless of temporality, only setting end time
-		ts := mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		tsl := mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		assert.Equal(t, 1, len(tsl))
+		ts := tsl[0]
 		assert.Equal(t, ts.MetricKind, metricpb.MetricDescriptor_GAUGE)
 		assert.Equal(t, ts.Points[0].Interval, &monitoringpb.TimeInterval{
 			EndTime: timestamppb.New(end),
 		})
 
 		sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityDelta)
-		ts = mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		tsl = mapper.sumPointToTimeSeries(mr, labels{}, metric, sum, point)
+		assert.Equal(t, 1, len(tsl))
+		ts = tsl[0]
 		assert.Equal(t, ts.MetricKind, metricpb.MetricDescriptor_GAUGE)
 		assert.Equal(t, ts.Points[0].Interval, &monitoringpb.TimeInterval{
 			EndTime: timestamppb.New(end),
@@ -384,19 +490,27 @@ func TestSumPointToTimeSeries(t *testing.T) {
 
 	t.Run("Add labels", func(t *testing.T) {
 		metric, sum, point := newCase()
+		end := start.Add(time.Hour)
+		point.SetStartTimestamp(pdata.NewTimestampFromTime(start))
+		point.SetTimestamp(pdata.NewTimestampFromTime(end))
 		extraLabels := map[string]string{"foo": "bar"}
-		ts := mapper.sumPointToTimeSeries(mr, labels(extraLabels), metric, sum, point)
+		tsl := mapper.sumPointToTimeSeries(mr, labels(extraLabels), metric, sum, point)
+		assert.Equal(t, 1, len(tsl))
+		ts := tsl[0]
 		assert.Equal(t, ts.Metric.Labels, extraLabels)
 
 		// Full set of labels
 		point.Attributes().InsertString("baz", "bar")
-		ts = mapper.sumPointToTimeSeries(mr, labels(extraLabels), metric, sum, point)
+		tsl = mapper.sumPointToTimeSeries(mr, labels(extraLabels), metric, sum, point)
+		assert.Equal(t, 1, len(tsl))
+		ts = tsl[0]
 		assert.Equal(t, ts.Metric.Labels, map[string]string{"foo": "bar", "baz": "bar"})
 	})
 }
 
 func TestGaugePointToTimeSeries(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	mr := &monitoredrespb.MonitoredResource{}
 
 	newCase := func() (pdata.Metric, pdata.Gauge, pdata.NumberDataPoint) {
@@ -452,7 +566,8 @@ func TestGaugePointToTimeSeries(t *testing.T) {
 }
 
 func TestSummaryPointToTimeSeries(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	mr := &monitoredrespb.MonitoredResource{}
 
 	metric := pdata.NewMetric()
@@ -524,7 +639,8 @@ func TestSummaryPointToTimeSeries(t *testing.T) {
 }
 
 func TestMetricNameToType(t *testing.T) {
-	mapper := newTestMetricMapper()
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
 	assert.Equal(
 		t,
 		mapper.metricNameToType("foo"),
@@ -629,7 +745,7 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				gauge := metric.Gauge()
 				point := gauge.DataPoints().AppendEmpty()
 				point.SetDoubleVal(10)
-				point.Attributes().InsertString("test_label", "test_value")
+				point.Attributes().InsertString("test.label", "test_value")
 				return metric
 			},
 			expected: []*metricpb.MetricDescriptor{
@@ -662,7 +778,7 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityCumulative)
 				point := sum.DataPoints().AppendEmpty()
 				point.SetDoubleVal(10)
-				point.Attributes().InsertString("test_label", "test_value")
+				point.Attributes().InsertString("test.label", "test_value")
 				return metric
 			},
 			expected: []*metricpb.MetricDescriptor{
@@ -695,7 +811,7 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityDelta)
 				point := sum.DataPoints().AppendEmpty()
 				point.SetDoubleVal(10)
-				point.Attributes().InsertString("test_label", "test_value")
+				point.Attributes().InsertString("test.label", "test_value")
 				return metric
 			},
 			expected: []*metricpb.MetricDescriptor{
@@ -728,7 +844,7 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityCumulative)
 				point := sum.DataPoints().AppendEmpty()
 				point.SetDoubleVal(10)
-				point.Attributes().InsertString("test_label", "test_value")
+				point.Attributes().InsertString("test.label", "test_value")
 				return metric
 			},
 			expected: []*metricpb.MetricDescriptor{
@@ -759,7 +875,7 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				histogram := metric.Histogram()
 				histogram.SetAggregationTemporality(pdata.MetricAggregationTemporalityCumulative)
 				point := histogram.DataPoints().AppendEmpty()
-				point.Attributes().InsertString("test_label", "test_value")
+				point.Attributes().InsertString("test.label", "test_value")
 				return metric
 			},
 			expected: []*metricpb.MetricDescriptor{
@@ -814,7 +930,7 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				metric.SetUnit("1")
 				summary := metric.Summary()
 				point := summary.DataPoints().AppendEmpty()
-				point.Attributes().InsertString("test_label", "value")
+				point.Attributes().InsertString("test.label", "value")
 				return metric
 			},
 			expected: []*metricpb.MetricDescriptor{
@@ -876,7 +992,7 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				for i := 0; i < 5; i++ {
 					point := gauge.DataPoints().AppendEmpty()
 					point.SetDoubleVal(10)
-					point.Attributes().InsertString("test_label", "test_value")
+					point.Attributes().InsertString("test.label", "test_value")
 				}
 				return metric
 			},
@@ -910,13 +1026,13 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				sum.SetAggregationTemporality(pdata.MetricAggregationTemporalityCumulative)
 				point := sum.DataPoints().AppendEmpty()
 				point.SetDoubleVal(10)
-				point.Attributes().InsertString("test_label", "test_value")
+				point.Attributes().InsertString("test.label", "test_value")
 				return metric
 			},
 			extraLabels: labels{
-				"service_name":        "myservice",
-				"service_instance_id": "abcdef",
-				"service_namespace":   "myns",
+				"service.name":        "myservice",
+				"service.instance_id": "abcdef",
+				"service.namespace":   "myns",
 			},
 			expected: []*metricpb.MetricDescriptor{
 				{
@@ -944,10 +1060,45 @@ func TestMetricDescriptorMapping(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "Attributes that collide",
+			metricCreator: func() pdata.Metric {
+				metric := pdata.NewMetric()
+				metric.SetDataType(pdata.MetricDataTypeGauge)
+				metric.SetName("custom.googleapis.com/test.metric")
+				metric.SetDescription("Description")
+				metric.SetUnit("1")
+				gauge := metric.Gauge()
+				point := gauge.DataPoints().AppendEmpty()
+				point.SetDoubleVal(10)
+				point.Attributes().InsertString("test.label", "test_value")
+				point.Attributes().InsertString("test_label", "other_value")
+				return metric
+			},
+			expected: []*metricpb.MetricDescriptor{
+				{
+					Name:        "custom.googleapis.com/test.metric",
+					DisplayName: "test.metric",
+					Type:        "custom.googleapis.com/test.metric",
+					MetricKind:  metricpb.MetricDescriptor_GAUGE,
+					ValueType:   metricpb.MetricDescriptor_DOUBLE,
+					Unit:        "1",
+					Description: "Description",
+					Labels: []*label.LabelDescriptor{
+						{
+							// even though we had two attributes originally,
+							// they are a single label after sanitization
+							Key: "test_label",
+						},
+					},
+				},
+			},
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			mapper := newTestMetricMapper()
+			mapper, shutdown := newTestMetricMapper()
+			defer shutdown()
 			metric := test.metricCreator()
 			md := mapper.metricDescriptor(metric, test.extraLabels)
 			diff := cmp.Diff(
@@ -1004,7 +1155,8 @@ func TestKnownDomains(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(fmt.Sprintf("%v to %v", test.name, test.metricType), func(t *testing.T) {
-			mapper := newTestMetricMapper()
+			mapper, shutdown := newTestMetricMapper()
+			defer shutdown()
 			if len(test.knownDomains) > 0 {
 				mapper.cfg.MetricConfig.KnownDomains = test.knownDomains
 			}
@@ -1046,7 +1198,8 @@ func TestInstrumentationLibraryToLabels(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			m := newTestMetricMapper()
+			m, shutdown := newTestMetricMapper()
+			defer shutdown()
 			m.cfg.MetricConfig = test.metricConfig
 
 			out := m.instrumentationLibraryToLabels(test.instrumentationLibrary)
diff --git a/exporter/collector/traces_test.go b/exporter/collector/traces_test.go
index b38d219..35f8569 100644
--- a/exporter/collector/traces_test.go
+++ b/exporter/collector/traces_test.go
@@ -57,9 +57,10 @@ func (ts *testServer) CreateSpan(context.Context, *cloudtracepb.Span) (*cloudtra
 
 func TestGoogleCloudTraceExport(t *testing.T) {
 	type testCase struct {
-		name        string
-		cfg         Config
-		expectedErr string
+		name               string
+		cfg                Config
+		expectedErr        string
+		expectedServiceKey string
 	}
 
 	testCases := []testCase{
@@ -74,6 +75,21 @@ func TestGoogleCloudTraceExport(t *testing.T) {
 					},
 				},
 			},
+			expectedServiceKey: "g.co/gae/app/module",
+		},
+		{
+			name: "With Empty Mapping",
+			cfg: Config{
+				ProjectID: "idk",
+				TraceConfig: TraceConfig{
+					ClientConfig: ClientConfig{
+						Endpoint:    "127.0.0.1:8080",
+						UseInsecure: true,
+					},
+					AttributeMappings: []AttributeMapping{},
+				},
+			},
+			expectedServiceKey: "service.name",
 		},
 	}
 
@@ -108,12 +124,15 @@ func TestGoogleCloudTraceExport(t *testing.T) {
 			span := ispans.Spans().AppendEmpty()
 			span.SetName(spanName)
 			span.SetStartTimestamp(pdata.NewTimestampFromTime(testTime))
+			span.Attributes().InsertString("service.name", "myservice")
 			err = sde.PushTraces(ctx, traces)
 			assert.NoError(t, err)
 
 			r := <-reqCh
 			assert.Len(t, r.Spans, 1)
 			assert.Equal(t, spanName, r.Spans[0].GetDisplayName().Value)
+			_, ok := r.Spans[0].GetAttributes().GetAttributeMap()[test.expectedServiceKey]
+			assert.True(t, ok)
 			assert.Equal(t, timestamppb.New(testTime), r.Spans[0].StartTime)
 		})
 	}
diff --git a/exporter/trace/trace_proto_test.go b/exporter/trace/trace_proto_test.go
index 66b2d25..a624f26 100644
--- a/exporter/trace/trace_proto_test.go
+++ b/exporter/trace/trace_proto_test.go
@@ -25,6 +25,15 @@ import (
 	tracepb "google.golang.org/genproto/googleapis/devtools/cloudtrace/v2"
 )
 
+func testExporter() *traceExporter {
+	return &traceExporter{
+		o: &options{
+			Context:      context.Background(),
+			mapAttribute: defaultAttributeMapping,
+		},
+	}
+}
+
 func genSpanContext() trace.SpanContext {
 	tracer := sdktrace.NewTracerProvider().Tracer("")
 	_, span := tracer.Start(context.Background(), "")
@@ -33,10 +42,12 @@ func genSpanContext() trace.SpanContext {
 
 func TestTraceProto_linksProtoFromLinks(t *testing.T) {
 	t.Run("Should be nil when no links", func(t *testing.T) {
-		assert.Nil(t, linksProtoFromLinks([]sdktrace.Link{}))
+		e := testExporter()
+		assert.Nil(t, e.linksProtoFromLinks([]sdktrace.Link{}))
 	})
 
 	t.Run("Can convert one link", func(t *testing.T) {
+		e := testExporter()
 		spanContext := genSpanContext()
 		link := sdktrace.Link{
 			SpanContext: spanContext,
@@ -44,7 +55,7 @@ func TestTraceProto_linksProtoFromLinks(t *testing.T) {
 				attribute.String("hello", "world"),
 			},
 		}
-		linksPb := linksProtoFromLinks([]sdktrace.Link{link})
+		linksPb := e.linksProtoFromLinks([]sdktrace.Link{link})
 
 		assert.NotNil(t, linksPb)
 		assert.EqualValues(t, linksPb.DroppedLinksCount, 0)
@@ -62,6 +73,7 @@ func TestTraceProto_linksProtoFromLinks(t *testing.T) {
 	})
 
 	t.Run("Drops links when there are more than 128", func(t *testing.T) {
+		e := testExporter()
 		var links []sdktrace.Link
 		for i := 0; i < 148; i++ {
 			links = append(
@@ -73,7 +85,7 @@ func TestTraceProto_linksProtoFromLinks(t *testing.T) {
 					},
 				})
 		}
-		linksPb := linksProtoFromLinks(links)
+		linksPb := e.linksProtoFromLinks(links)
 		assert.NotNil(t, linksPb)
 		assert.EqualValues(t, linksPb.DroppedLinksCount, 20)
 		assert.Len(t, linksPb.Link, maxNumLinks)