Repository: GoogleCloudPlatform/opentelemetry-operations-go
Pull request #360 (2022-04-18 21:16:55)

Normalize histograms, summaries, and exponential histograms
## Background

Normalization is done for cumulative streams of metrics.  For example, if we have a counter that counts the number of seconds since 1970, we wouldn't want to report that number as having happened since we started reading it, since that would produce an enormous rate  (>1 billion over 10 seconds).  Instead, the first time we see a metric, we record its current value.  Then, each time we see the metric again, we "subtract" the original value from the new one.  That is trivial for a counter, and we already did that in https://github.com/GoogleCloudPlatform/opentelemetry-operations-go/pull/323.  It is harder for other data types (how do you subtract a histogram from another histogram?).

## Changes

This PR extends tracking for the first time we saw a metric by extending the data point cache for other types.  It then copies logic for setting the start time for intervals from the sum implementation.  Finally, it implements "subtraction" for each of the remaining types: summaries, histograms, and exponential histograms.

For histograms, just subtract the previous sum, previous count, and subtract each bucket from the previous point.

For summaries, leave the quantiles as they are (treat them like gauges passing through), and just subtract previous sum and previous count.

For exponential histograms, subtract the previous sum, previous count.  Also subtract each bucket from the previous point, taking the offset into account.

quentinmit: Commenting on the logic from your description, without reading the code:

- Histograms and summaries should only be normalized as described if they represent a cumulative "aggregation temporality". (Side note: I don't see a gauge aggregation temporality - that would also of course not require any normalization.) Subtracting the original sum/count/bucket counts seems correct, but make sure you are subtracting the correct buckets - if the bucket boundaries change you need to either reset or check if they're a superset of the previous point.
- For summaries I don't see any way to normalize quantiles (what does a quantile-over-cumulative even mean in this context?)
- Exponential histograms seem the same as regular histograms, except you can more reliably line up corresponding buckets

Patch:
diff --git a/example/trace/http/go.mod b/example/trace/http/go.mod
index c1b18c1..148d14f 100644
--- a/example/trace/http/go.mod
+++ b/example/trace/http/go.mod
@@ -4,18 +4,21 @@ go 1.17
 
 replace github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace => ../../../exporter/trace
 
+replace github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping => ../../../internal/resourcemapping
+
 require (
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace v1.4.0
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0
-	go.opentelemetry.io/otel v1.6.2
+	go.opentelemetry.io/otel v1.6.3
 	go.opentelemetry.io/otel/sdk v1.6.2
-	go.opentelemetry.io/otel/trace v1.6.2
+	go.opentelemetry.io/otel/trace v1.6.3
 )
 
 require (
 	cloud.google.com/go/compute v1.5.0 // indirect
 	cloud.google.com/go/trace v1.2.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.0.0-00010101000000-000000000000 // indirect
 	github.com/felixge/httpsnoop v1.0.2 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
diff --git a/example/trace/http/go.sum b/example/trace/http/go.sum
index d9bc849..a5e2a39 100644
--- a/example/trace/http/go.sum
+++ b/example/trace/http/go.sum
@@ -212,16 +212,18 @@ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0 h1:woM+Mb4
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0/go.mod h1:PFmBsWbldL1kiWZk9+0LBZz2brhByaGsvp6pRICMlPE=
 go.opentelemetry.io/otel v1.6.0/go.mod h1:bfJD2DZVw0LBxghOTlgnlI0CV3hLDu9XF/QKOUXMTQQ=
 go.opentelemetry.io/otel v1.6.1/go.mod h1:blzUabWHkX6LJewxvadmzafgh/wnvBSDBdOuwkAtrWQ=
-go.opentelemetry.io/otel v1.6.2 h1:iLrj1F5ydCSVc4CGjBX2ou/QRav3C8xHjHlVoch6g4w=
 go.opentelemetry.io/otel v1.6.2/go.mod h1:MUBZHaB2cm6CahEBHQPq9Anos7IXynP/noVpjsxQTSc=
+go.opentelemetry.io/otel v1.6.3 h1:FLOfo8f9JzFVFVyU+MSRJc2HdEAXQgm7pIv2uFKRSZE=
+go.opentelemetry.io/otel v1.6.3/go.mod h1:7BgNga5fNlF/iZjG06hM3yofffp0ofKCDwSXx1GC4dI=
 go.opentelemetry.io/otel/metric v0.28.0 h1:o5YNh+jxACMODoAo1bI7OES0RUW4jAMae0Vgs2etWAQ=
 go.opentelemetry.io/otel/metric v0.28.0/go.mod h1:TrzsfQAmQaB1PDcdhBauLMk7nyyg9hm+GoQq/ekE9Iw=
 go.opentelemetry.io/otel/sdk v1.6.2 h1:wxY+YrfpGJfjxtm7SFBMJp9APDMZjDG+ErZOs/wkubg=
 go.opentelemetry.io/otel/sdk v1.6.2/go.mod h1:M2r4VCm1Yurk4E+fWtP2p+QzFDHMFEqhGdbtQ7zRf+k=
 go.opentelemetry.io/otel/trace v1.6.0/go.mod h1:qs7BrU5cZ8dXQHBGxHMOxwME/27YH2qEp4/+tZLLwJE=
 go.opentelemetry.io/otel/trace v1.6.1/go.mod h1:RkFRM1m0puWIq10oxImnGEduNBzxiN7TXluRBtE+5j0=
-go.opentelemetry.io/otel/trace v1.6.2 h1:oY7i1k6XD/ozlGo7ASy+H1UdkNcj9cPfuklaYSXtoFk=
 go.opentelemetry.io/otel/trace v1.6.2/go.mod h1:RMqfw8Mclba1p7sXDmEDBvrB8jw65F6GOoN1fyyXTzk=
+go.opentelemetry.io/otel/trace v1.6.3 h1:IqN4L+5b0mPNjdXIiZ90Ni4Bl5BRkDQywePLWemd9bc=
+go.opentelemetry.io/otel/trace v1.6.3/go.mod h1:GNJQusJlUgZl9/TQBPKU/Y/ty+0iVB5fjhKeJGZPGFs=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
diff --git a/exporter/collector/config.go b/exporter/collector/config.go
index dcaf368..19bbfc0 100644
--- a/exporter/collector/config.go
+++ b/exporter/collector/config.go
@@ -94,6 +94,12 @@ type MetricConfig struct {
 	// service_resource_labels option operates independently from resource_filters.
 	ResourceFilters []ResourceFilter `mapstructure:"resource_filters"`
 
+	// CumulativeNormalization normalizes cumulative metrics without start times or with
+	// explicit reset points by subtracting subsequent points from the initial point.
+	// It is enabled by default. Since it caches starting points, it may result in
+	// increased memory usage.
+	CumulativeNormalization bool `mapstructure:"cumulative_normalization"`
+
 	// This enables calculation of an estimated sum of squared deviation.  It isn't correct,
 	// so we don't send it by default, and don't expose it to users. For some uses, it is
 	// expected, however.
@@ -118,6 +124,7 @@ func DefaultConfig() Config {
 			CreateMetricDescriptorBufferSize: 10,
 			InstrumentationLibraryLabels:     true,
 			ServiceResourceLabels:            true,
+			CumulativeNormalization:          true,
 		},
 	}
 }
diff --git a/exporter/collector/go.mod b/exporter/collector/go.mod
index c2af2b8..d719a7d 100644
--- a/exporter/collector/go.mod
+++ b/exporter/collector/go.mod
@@ -9,9 +9,9 @@ require (
 	github.com/stretchr/testify v1.7.1
 	go.opencensus.io v0.23.0
 	go.opentelemetry.io/collector/model v0.49.0
-	go.opentelemetry.io/otel v1.6.2
+	go.opentelemetry.io/otel v1.6.3
 	go.opentelemetry.io/otel/sdk v1.6.2
-	go.opentelemetry.io/otel/trace v1.6.2
+	go.opentelemetry.io/otel/trace v1.6.3
 	google.golang.org/api v0.74.0
 	google.golang.org/genproto v0.0.0-20220405205423-9d709892a2bf
 	google.golang.org/grpc v1.45.0
@@ -21,6 +21,7 @@ require (
 require (
 	cloud.google.com/go/monitoring v1.4.0
 	cloud.google.com/go/trace v1.2.0
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.0.0-00010101000000-000000000000
 	github.com/google/go-cmp v0.5.7
 	go.opentelemetry.io/collector/pdata v0.49.0
 	go.uber.org/zap v1.21.0
@@ -50,3 +51,5 @@ require (
 )
 
 replace github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace => ../trace
+
+replace github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping => ../../internal/resourcemapping
diff --git a/exporter/collector/go.sum b/exporter/collector/go.sum
index c537583..8821528 100644
--- a/exporter/collector/go.sum
+++ b/exporter/collector/go.sum
@@ -232,13 +232,15 @@ go.opentelemetry.io/collector/model v0.49.0/go.mod h1:nOYQv9KoFPs6ihJwOi24qB209E
 go.opentelemetry.io/collector/pdata v0.49.0 h1:aYj5rOlRC0x7lGXbc185LMsMMoY/pjOTXr5s1O2SzXs=
 go.opentelemetry.io/collector/pdata v0.49.0/go.mod h1:YwmKuiFhNgtmhRdpi8Q8FAWPa0AwJTCSlssSsAtuRcY=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0 h1:woM+Mb4d0A+Dxa3rYPenSN5ZeS9qHUvE8rlObiLRXTY=
-go.opentelemetry.io/otel v1.6.2 h1:iLrj1F5ydCSVc4CGjBX2ou/QRav3C8xHjHlVoch6g4w=
 go.opentelemetry.io/otel v1.6.2/go.mod h1:MUBZHaB2cm6CahEBHQPq9Anos7IXynP/noVpjsxQTSc=
+go.opentelemetry.io/otel v1.6.3 h1:FLOfo8f9JzFVFVyU+MSRJc2HdEAXQgm7pIv2uFKRSZE=
+go.opentelemetry.io/otel v1.6.3/go.mod h1:7BgNga5fNlF/iZjG06hM3yofffp0ofKCDwSXx1GC4dI=
 go.opentelemetry.io/otel/metric v0.28.0 h1:o5YNh+jxACMODoAo1bI7OES0RUW4jAMae0Vgs2etWAQ=
 go.opentelemetry.io/otel/sdk v1.6.2 h1:wxY+YrfpGJfjxtm7SFBMJp9APDMZjDG+ErZOs/wkubg=
 go.opentelemetry.io/otel/sdk v1.6.2/go.mod h1:M2r4VCm1Yurk4E+fWtP2p+QzFDHMFEqhGdbtQ7zRf+k=
-go.opentelemetry.io/otel/trace v1.6.2 h1:oY7i1k6XD/ozlGo7ASy+H1UdkNcj9cPfuklaYSXtoFk=
 go.opentelemetry.io/otel/trace v1.6.2/go.mod h1:RMqfw8Mclba1p7sXDmEDBvrB8jw65F6GOoN1fyyXTzk=
+go.opentelemetry.io/otel/trace v1.6.3 h1:IqN4L+5b0mPNjdXIiZ90Ni4Bl5BRkDQywePLWemd9bc=
+go.opentelemetry.io/otel/trace v1.6.3/go.mod h1:GNJQusJlUgZl9/TQBPKU/Y/ty+0iVB5fjhKeJGZPGFs=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
diff --git a/exporter/collector/internal/datapointstorage/datapointcache.go b/exporter/collector/internal/datapointstorage/datapointcache.go
index a371468..c0051c9 100644
--- a/exporter/collector/internal/datapointstorage/datapointcache.go
+++ b/exporter/collector/internal/datapointstorage/datapointcache.go
@@ -25,16 +25,41 @@ import (
 
 const gcInterval = 20 * time.Minute
 
-type Cache map[string]usedPoint
+type Cache struct {
+	numberCache               map[string]usedNumberPoint
+	summaryCache              map[string]usedSummaryPoint
+	histogramCache            map[string]usedHistogramPoint
+	exponentialHistogramCache map[string]usedExponentialHistogramPoint
+}
 
-type usedPoint struct {
+type usedNumberPoint struct {
 	point *pdata.NumberDataPoint
 	used  bool
 }
 
-// New instantiates a cache and starts background processes
+type usedSummaryPoint struct {
+	point *pdata.SummaryDataPoint
+	used  bool
+}
+
+type usedHistogramPoint struct {
+	point *pdata.HistogramDataPoint
+	used  bool
+}
+
+type usedExponentialHistogramPoint struct {
+	point *pdata.ExponentialHistogramDataPoint
+	used  bool
+}
+
+// NewCache instantiates a cache and starts background processes
 func NewCache(shutdown <-chan struct{}) Cache {
-	c := make(Cache)
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
 	go func() {
 		ticker := time.NewTicker(gcInterval)
 		for c.gc(shutdown, ticker.C) {
@@ -43,20 +68,68 @@ func NewCache(shutdown <-chan struct{}) Cache {
 	return c
 }
 
-// Get retrieves the point associated with the identifier, and whether
+// GetNumberDataPoint retrieves the point associated with the identifier, and whether
+// or not it was found
+func (c Cache) GetNumberDataPoint(identifier string) (*pdata.NumberDataPoint, bool) {
+	point, found := c.numberCache[identifier]
+	if found {
+		point.used = true
+		c.numberCache[identifier] = point
+	}
+	return point.point, found
+}
+
+// SetNumberDataPoint assigns the point to the identifier in the cache
+func (c Cache) SetNumberDataPoint(identifier string, point *pdata.NumberDataPoint) {
+	c.numberCache[identifier] = usedNumberPoint{point, true}
+}
+
+// GetSummaryDataPoint 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]
+func (c Cache) GetSummaryDataPoint(identifier string) (*pdata.SummaryDataPoint, bool) {
+	point, found := c.summaryCache[identifier]
 	if found {
 		point.used = true
-		c[identifier] = point
+		c.summaryCache[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}
+// SetSummaryDataPoint assigns the point to the identifier in the cache
+func (c Cache) SetSummaryDataPoint(identifier string, point *pdata.SummaryDataPoint) {
+	c.summaryCache[identifier] = usedSummaryPoint{point, true}
+}
+
+// GetHistogramDataPoint retrieves the point associated with the identifier, and whether
+// or not it was found
+func (c Cache) GetHistogramDataPoint(identifier string) (*pdata.HistogramDataPoint, bool) {
+	point, found := c.histogramCache[identifier]
+	if found {
+		point.used = true
+		c.histogramCache[identifier] = point
+	}
+	return point.point, found
+}
+
+// SetHistogramDataPoint assigns the point to the identifier in the cache
+func (c Cache) SetHistogramDataPoint(identifier string, point *pdata.HistogramDataPoint) {
+	c.histogramCache[identifier] = usedHistogramPoint{point, true}
+}
+
+// GetExponentialHistogramDataPoint retrieves the point associated with the identifier, and whether
+// or not it was found
+func (c Cache) GetExponentialHistogramDataPoint(identifier string) (*pdata.ExponentialHistogramDataPoint, bool) {
+	point, found := c.exponentialHistogramCache[identifier]
+	if found {
+		point.used = true
+		c.exponentialHistogramCache[identifier] = point
+	}
+	return point.point, found
+}
+
+// SetExponentialHistogramDataPoint assigns the point to the identifier in the cache
+func (c Cache) SetExponentialHistogramDataPoint(identifier string, point *pdata.ExponentialHistogramDataPoint) {
+	c.exponentialHistogramCache[identifier] = usedExponentialHistogramPoint{point, true}
 }
 
 // gc garbage collects the cache after the ticker ticks
@@ -65,15 +138,48 @@ func (c Cache) gc(shutdown <-chan struct{}, tickerCh <-chan time.Time) bool {
 	case <-shutdown:
 		return false
 	case <-tickerCh:
-		// garbage collect the cache
-		for id, point := range c {
+		// garbage collect the numberCache
+		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
+			} else {
+				// for points that have not been used, delete points
+				delete(c.numberCache, id)
+			}
+		}
+		// garbage collect the summaryCache
+		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
+			} else {
+				// for points that have not been used, delete points
+				delete(c.summaryCache, id)
+			}
+		}
+		// garbage collect the histogramCache
+		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
+			} else {
+				// for points that have not been used, delete points
+				delete(c.histogramCache, id)
+			}
+		}
+		// garbage collect the exponentialHistogramCache
+		for id, point := range c.exponentialHistogramCache {
 			if point.used {
 				// for points that have been used, mark them as unused
 				point.used = false
-				c[id] = point
+				c.exponentialHistogramCache[id] = point
 			} else {
 				// for points that have not been used, delete points
-				delete(c, id)
+				delete(c.exponentialHistogramCache, id)
 			}
 		}
 	}
diff --git a/exporter/collector/internal/normalization/disabled_normalizer.go b/exporter/collector/internal/normalization/disabled_normalizer.go
new file mode 100644
index 0000000..5e46890
--- /dev/null
+++ b/exporter/collector/internal/normalization/disabled_normalizer.go
@@ -0,0 +1,46 @@
+// 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 normalization
+
+import "go.opentelemetry.io/collector/model/pdata"
+
+// NewDisabledNormalizer returns a Normalizer which does not perform any
+// normalization. This may be useful if standard normalization consumes too
+// much memory.
+func NewDisabledNormalizer() Normalizer {
+	return &disabledNormalizer{}
+}
+
+type disabledNormalizer struct{}
+
+// NormalizeExponentialHistogramDataPoint returns the point without normalizing.
+func (d *disabledNormalizer) NormalizeExponentialHistogramDataPoint(point pdata.ExponentialHistogramDataPoint, _ string) *pdata.ExponentialHistogramDataPoint {
+	return &point
+}
+
+// NormalizeHistogramDataPoint returns the point without normalizing.
+func (d *disabledNormalizer) NormalizeHistogramDataPoint(point pdata.HistogramDataPoint, _ string) *pdata.HistogramDataPoint {
+	return &point
+}
+
+// NormalizeNumberDataPoint returns the point without normalizing.
+func (d *disabledNormalizer) NormalizeNumberDataPoint(point pdata.NumberDataPoint, _ string) *pdata.NumberDataPoint {
+	return &point
+}
+
+// NormalizeSummaryDataPoint returns the point without normalizing.
+func (d *disabledNormalizer) NormalizeSummaryDataPoint(point pdata.SummaryDataPoint, _ string) *pdata.SummaryDataPoint {
+	return &point
+}
diff --git a/exporter/collector/internal/normalization/standard_normalizer.go b/exporter/collector/internal/normalization/standard_normalizer.go
new file mode 100644
index 0000000..75a1450
--- /dev/null
+++ b/exporter/collector/internal/normalization/standard_normalizer.go
@@ -0,0 +1,247 @@
+// 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 normalization
+
+import (
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
+
+	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/datapointstorage"
+)
+
+// NewStandardNormalizer performs normalization on cumulative points which:
+//  (a) don't have a start time, OR
+//  (b) have been sent a preceding "reset" point as described in https://github.com/open-telemetry/opentelemetry-specification/blob/9555f9594c7ffe5dc333b53da5e0f880026cead1/specification/metrics/datamodel.md#resets-and-gaps
+// The first point without a start time or the reset point is cached, and is
+// NOT exported. Subsequent points "subtract" the initial point prior to exporting.
+func NewStandardNormalizer(shutdown <-chan struct{}, logger *zap.Logger) Normalizer {
+	return &standardNormalizer{
+		cache: datapointstorage.NewCache(shutdown),
+		log:   logger,
+	}
+}
+
+type standardNormalizer struct {
+	cache datapointstorage.Cache
+	log   *zap.Logger
+}
+
+func (s *standardNormalizer) NormalizeExponentialHistogramDataPoint(point pdata.ExponentialHistogramDataPoint, identifier string) *pdata.ExponentialHistogramDataPoint {
+	// if the point doesn't need to be normalized, use original point
+	normalizedPoint := &point
+	start, ok := s.cache.GetExponentialHistogramDataPoint(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.
+			s.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
+		}
+		if point.Scale() != start.Scale() {
+			// TODO(#366): It is possible, but difficult to compare exponential
+			// histograms with different scales. For now, treat a change in
+			// scale as a reset.
+			s.cache.SetExponentialHistogramDataPoint(identifier, &point)
+			return nil
+		}
+		// Make a copy so we don't mutate underlying data
+		newPoint := pdata.NewExponentialHistogramDataPoint()
+		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
+		newPoint.SetCount(point.Count() - start.Count())
+		// We drop points without a sum, so no need to check here.
+		newPoint.SetSum(point.Sum() - start.Sum())
+		newPoint.SetZeroCount(point.ZeroCount() - start.ZeroCount())
+		normalizeExponentialBuckets(newPoint.Positive(), start.Positive())
+		normalizeExponentialBuckets(newPoint.Negative(), start.Negative())
+		normalizedPoint = &newPoint
+	}
+	if (!ok && point.StartTimestamp() == 0) || !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
+		//
+		// Record it in history and drop the point.
+		s.cache.SetExponentialHistogramDataPoint(identifier, &point)
+		return nil
+	}
+	return normalizedPoint
+}
+
+func normalizeExponentialBuckets(pointBuckets, startBuckets pdata.Buckets) {
+	newBuckets := make([]uint64, len(pointBuckets.BucketCounts()))
+	offsetDiff := int(pointBuckets.Offset() - startBuckets.Offset())
+	for i := range pointBuckets.BucketCounts() {
+		startOffset := i + offsetDiff
+		// if there is no corresponding bucket for the starting bucketcounts, don't normalize
+		if startOffset < 0 || startOffset >= len(startBuckets.BucketCounts()) {
+			newBuckets[i] = pointBuckets.BucketCounts()[i]
+		} else {
+			newBuckets[i] = pointBuckets.BucketCounts()[i] - startBuckets.BucketCounts()[startOffset]
+		}
+	}
+	pointBuckets.SetOffset(pointBuckets.Offset())
+	pointBuckets.SetBucketCounts(newBuckets)
+}
+
+func (s *standardNormalizer) NormalizeHistogramDataPoint(point pdata.HistogramDataPoint, identifier string) *pdata.HistogramDataPoint {
+	// if the point doesn't need to be normalized, use original point
+	normalizedPoint := &point
+	start, ok := s.cache.GetHistogramDataPoint(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.
+			s.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.NewHistogramDataPoint()
+		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
+		newPoint.SetCount(point.Count() - start.Count())
+		// We drop points without a sum, so no need to check here.
+		newPoint.SetSum(point.Sum() - start.Sum())
+		pointBuckets := point.BucketCounts()
+		startBuckets := start.BucketCounts()
+		if !bucketBoundariesEqual(point.ExplicitBounds(), start.ExplicitBounds()) {
+			// The number of buckets changed, so we can't normalize points anymore.
+			// Treat this as a reset by recording and dropping this point.
+			s.cache.SetHistogramDataPoint(identifier, &point)
+			return nil
+		}
+		newBuckets := make([]uint64, len(pointBuckets))
+		for i := range pointBuckets {
+			newBuckets[i] = pointBuckets[i] - startBuckets[i]
+		}
+		newPoint.SetBucketCounts(newBuckets)
+		normalizedPoint = &newPoint
+	}
+	if (!ok && point.StartTimestamp() == 0) || !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.
+		s.cache.SetHistogramDataPoint(identifier, &point)
+		return nil
+	}
+	return normalizedPoint
+}
+
+func bucketBoundariesEqual(a, b []float64) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// NormalizeNumberDataPoint normalizes a cumulative, monotonic sum.
+// It returns the normalized point, or nil if the point should be dropped.
+func (s *standardNormalizer) NormalizeNumberDataPoint(point pdata.NumberDataPoint, identifier string) *pdata.NumberDataPoint {
+	// if the point doesn't need to be normalized, use original point
+	normalizedPoint := &point
+	start, ok := s.cache.GetNumberDataPoint(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.
+			s.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 pmetric.MetricValueTypeInt:
+			newPoint.SetIntVal(point.IntVal() - start.IntVal())
+		case pmetric.MetricValueTypeDouble:
+			newPoint.SetDoubleVal(point.DoubleVal() - start.DoubleVal())
+		}
+		normalizedPoint = &newPoint
+	}
+	if (!ok && point.StartTimestamp() == 0) || !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.
+		s.cache.SetNumberDataPoint(identifier, &point)
+		return nil
+	}
+	return normalizedPoint
+}
+
+func (s *standardNormalizer) NormalizeSummaryDataPoint(point pdata.SummaryDataPoint, identifier string) *pdata.SummaryDataPoint {
+	// if the point doesn't need to be normalized, use original point
+	normalizedPoint := &point
+	start, ok := s.cache.GetSummaryDataPoint(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.
+			s.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.NewSummaryDataPoint()
+		// Quantile values are copied, and are not modified. Quantiles are
+		// computed over the same time period as sum and count, but it isn't
+		// possible to normalize them.
+		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
+		newPoint.SetCount(point.Count() - start.Count())
+		// We drop points without a sum, so no need to check here.
+		newPoint.SetSum(point.Sum() - start.Sum())
+		normalizedPoint = &newPoint
+	}
+	if (!ok && point.StartTimestamp() == 0) || !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.
+		s.cache.SetSummaryDataPoint(identifier, &point)
+		return nil
+	}
+	return normalizedPoint
+}
diff --git a/exporter/collector/internal/normalization/types.go b/exporter/collector/internal/normalization/types.go
new file mode 100644
index 0000000..9a55fef
--- /dev/null
+++ b/exporter/collector/internal/normalization/types.go
@@ -0,0 +1,33 @@
+// 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 normalization
+
+import "go.opentelemetry.io/collector/model/pdata"
+
+// Normalizer can normalize data points to handle cases in which the start time is unknown.
+type Normalizer interface {
+	// NormalizeExponentialHistogramDataPoint normalizes an exponential histogram.
+	// It returns the normalized point, or nil if the point should be dropped.
+	NormalizeExponentialHistogramDataPoint(point pdata.ExponentialHistogramDataPoint, identifier string) *pdata.ExponentialHistogramDataPoint
+	// NormalizeHistogramDataPoint normalizes a cumulative histogram.
+	// It returns the normalized point, or nil if the point should be dropped.
+	NormalizeHistogramDataPoint(point pdata.HistogramDataPoint, identifier string) *pdata.HistogramDataPoint
+	// NormalizeNumberDataPoint normalizes a cumulative, monotonic sum.
+	// It returns the normalized point, or nil if the point should be dropped.
+	NormalizeNumberDataPoint(point pdata.NumberDataPoint, identifier string) *pdata.NumberDataPoint
+	// NormalizeSummaryDataPoint normalizes a summary.
+	// It returns the normalized point, or nil if the point should be dropped.
+	NormalizeSummaryDataPoint(point pdata.SummaryDataPoint, identifier string) *pdata.SummaryDataPoint
+}
diff --git a/exporter/collector/metrics.go b/exporter/collector/metrics.go
index c17995d..ed23644 100644
--- a/exporter/collector/metrics.go
+++ b/exporter/collector/metrics.go
@@ -45,6 +45,7 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/datapointstorage"
+	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/normalization"
 )
 
 // self-observability reporting meters/tracers/loggers.
@@ -74,9 +75,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
-	sumCache datapointstorage.Cache
+	obs        selfObservability
+	cfg        Config
+	normalizer normalization.Normalizer
 }
 
 // Constants we use when translating summary metrics into GCP.
@@ -133,6 +134,10 @@ func NewGoogleCloudMetricsExporter(
 	}
 	obs := selfObservability{log: log}
 	shutdown := make(chan struct{})
+	normalizer := normalization.NewDisabledNormalizer()
+	if cfg.MetricConfig.CumulativeNormalization {
+		normalizer = normalization.NewStandardNormalizer(shutdown, log)
+	}
 	mExp := &MetricsExporter{
 		cfg:    cfg,
 		client: client,
@@ -140,7 +145,7 @@ func NewGoogleCloudMetricsExporter(
 		mapper: metricMapper{
 			obs,
 			cfg,
-			datapointstorage.NewCache(shutdown),
+			normalizer,
 		},
 		// We create a buffered channel for metric descriptors.
 		// MetricDescritpors are asychronously sent and optimistic.
@@ -372,6 +377,13 @@ func (m *metricMapper) summaryPointToTimeSeries(
 		// Drop points without a value.
 		return nil
 	}
+	// Normalize the summary point.
+	metricIdentifier := datapointstorage.Identifier(resource, extraLabels, metric, point.Attributes())
+	normalizedPoint := m.normalizer.NormalizeSummaryDataPoint(point, metricIdentifier)
+	if normalizedPoint == nil {
+		return nil
+	}
+	point = *normalizedPoint
 	sumName, countName := summaryMetricNames(metric.Name())
 	startTime := timestamppb.New(point.StartTimestamp().AsTime())
 	endTime := timestamppb.New(point.Timestamp().AsTime())
@@ -605,13 +617,23 @@ func (m *metricMapper) histogramToTimeSeries(
 	resource *monitoredrespb.MonitoredResource,
 	extraLabels labels,
 	metric pdata.Metric,
-	_ pdata.Histogram,
+	hist pdata.Histogram,
 	point pdata.HistogramDataPoint,
 ) []*monitoringpb.TimeSeries {
 	if point.Flags().HasFlag(pmetric.MetricDataPointFlagNoRecordedValue) || !point.HasSum() {
 		// Drop points without a value or without a sum
 		return nil
 	}
+	if hist.AggregationTemporality() == pmetric.MetricAggregationTemporalityCumulative {
+		// Normalize cumulative histogram points.
+		metricIdentifier := datapointstorage.Identifier(resource, extraLabels, metric, point.Attributes())
+		normalizedPoint := m.normalizer.NormalizeHistogramDataPoint(point, metricIdentifier)
+		if normalizedPoint == nil {
+			return nil
+		}
+		point = *normalizedPoint
+	}
+
 	// We treat deltas as cumulatives w/ resets.
 	metricKind := metricpb.MetricDescriptor_CUMULATIVE
 	startTime := timestamppb.New(point.StartTimestamp().AsTime())
@@ -643,13 +665,22 @@ func (m *metricMapper) exponentialHistogramToTimeSeries(
 	resource *monitoredrespb.MonitoredResource,
 	extraLabels labels,
 	metric pdata.Metric,
-	_ pdata.ExponentialHistogram,
+	exponentialHist pdata.ExponentialHistogram,
 	point pdata.ExponentialHistogramDataPoint,
 ) []*monitoringpb.TimeSeries {
 	if point.Flags().HasFlag(pmetric.MetricDataPointFlagNoRecordedValue) {
 		// Drop points without a value.
 		return nil
 	}
+	if exponentialHist.AggregationTemporality() == pmetric.MetricAggregationTemporalityCumulative {
+		// Normalize the histogram point.
+		metricIdentifier := datapointstorage.Identifier(resource, extraLabels, metric, point.Attributes())
+		normalizedPoint := m.normalizer.NormalizeExponentialHistogramDataPoint(point, metricIdentifier)
+		if normalizedPoint == nil {
+			return nil
+		}
+		point = *normalizedPoint
+	}
 	// We treat deltas as cumulatives w/ resets.
 	metricKind := metricpb.MetricDescriptor_CUMULATIVE
 	startTime := timestamppb.New(point.StartTimestamp().AsTime())
@@ -692,13 +723,15 @@ func (m *metricMapper) sumPointToTimeSeries(
 		return nil
 	}
 	if sum.IsMonotonic() {
-		metricIdentifier := datapointstorage.Identifier(resource, extraLabels, metric, point.Attributes())
-		normalizedPoint := m.normalizeNumberDataPoint(point, metricIdentifier)
-		if normalizedPoint == nil {
-			return nil
+		if sum.AggregationTemporality() == pmetric.MetricAggregationTemporalityCumulative {
+			metricIdentifier := datapointstorage.Identifier(resource, extraLabels, metric, point.Attributes())
+			normalizedPoint := m.normalizer.NormalizeNumberDataPoint(point, metricIdentifier)
+			if normalizedPoint == nil {
+				return nil
+			}
+			point = *normalizedPoint
 		}
-		point = *normalizedPoint
-		startTime = timestamppb.New(normalizedPoint.StartTimestamp().AsTime())
+		startTime = timestamppb.New(point.StartTimestamp().AsTime())
 	} else {
 		metricKind = metricpb.MetricDescriptor_GAUGE
 		startTime = nil
@@ -727,48 +760,6 @@ func (m *metricMapper) sumPointToTimeSeries(
 	}}
 }
 
-// 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 pmetric.MetricValueTypeInt:
-			newPoint.SetIntVal(point.IntVal() - start.IntVal())
-		case pmetric.MetricValueTypeDouble:
-			newPoint.SetDoubleVal(point.DoubleVal() - start.DoubleVal())
-		}
-		normalizedPoint = &newPoint
-	}
-	if (!ok && point.StartTimestamp() == 0) || !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(
 	resource *monitoredrespb.MonitoredResource,
 	extraLabels labels,
diff --git a/exporter/collector/monitoredresource.go b/exporter/collector/monitoredresource.go
index 535e947..8999300 100644
--- a/exporter/collector/monitoredresource.go
+++ b/exporter/collector/monitoredresource.go
@@ -21,100 +21,21 @@ import (
 	semconv "go.opentelemetry.io/collector/model/semconv/v1.8.0"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
-)
 
-const (
-	awsAccount     = "aws_account"
-	awsEc2Instance = "aws_ec2_instance"
-	clusterName    = "cluster_name"
-	containerName  = "container_name"
-	gceInstance    = "gce_instance"
-	genericNode    = "generic_node"
-	genericTask    = "generic_task"
-	instanceID     = "instance_id"
-	job            = "job"
-	k8sCluster     = "k8s_cluster"
-	k8sContainer   = "k8s_container"
-	k8sNode        = "k8s_node"
-	k8sPod         = "k8s_pod"
-	location       = "location"
-	namespace      = "namespace"
-	namespaceName  = "namespace_name"
-	nodeID         = "node_id"
-	nodeName       = "node_name"
-	podName        = "pod_name"
-	region         = "region"
-	taskID         = "task_id"
-	zone           = "zone"
+	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping"
 )
 
-var (
-	// monitoredResourceMappings contains mappings of GCM resource label keys onto mapping config from OTel
-	// resource for a given monitored resource type.
-	monitoredResourceMappings = map[string]map[string]struct {
-		// OTel resource keys to try and populate the resource label from. For entries with
-		// multiple OTel resource keys, the keys' values will be coalesced in order until there
-		// is a non-empty value.
-		otelKeys []string
-		// If none of the otelKeys are present in the Resource, fallback to this literal value
-		fallbackLiteral string
-	}{
-		gceInstance: {
-			zone:       {otelKeys: []string{semconv.AttributeCloudAvailabilityZone}},
-			instanceID: {otelKeys: []string{semconv.AttributeHostID}},
-		},
-		k8sContainer: {
-			location:      {otelKeys: []string{semconv.AttributeCloudAvailabilityZone}},
-			clusterName:   {otelKeys: []string{semconv.AttributeK8SClusterName}},
-			namespaceName: {otelKeys: []string{semconv.AttributeK8SNamespaceName}},
-			podName:       {otelKeys: []string{semconv.AttributeK8SPodName}},
-			containerName: {otelKeys: []string{semconv.AttributeK8SContainerName}},
-		},
-		k8sPod: {
-			location:      {otelKeys: []string{semconv.AttributeCloudAvailabilityZone}},
-			clusterName:   {otelKeys: []string{semconv.AttributeK8SClusterName}},
-			namespaceName: {otelKeys: []string{semconv.AttributeK8SNamespaceName}},
-			podName:       {otelKeys: []string{semconv.AttributeK8SPodName}},
-		},
-		k8sNode: {
-			location:    {otelKeys: []string{semconv.AttributeCloudAvailabilityZone}},
-			clusterName: {otelKeys: []string{semconv.AttributeK8SClusterName}},
-			nodeName:    {otelKeys: []string{semconv.AttributeK8SNodeName}},
-		},
-		k8sCluster: {
-			location:    {otelKeys: []string{semconv.AttributeCloudAvailabilityZone}},
-			clusterName: {otelKeys: []string{semconv.AttributeK8SClusterName}},
-		},
-		awsEc2Instance: {
-			instanceID: {otelKeys: []string{semconv.AttributeHostID}},
-			region:     {otelKeys: []string{semconv.AttributeCloudAvailabilityZone}},
-			awsAccount: {otelKeys: []string{semconv.AttributeCloudAccountID}},
-		},
-		genericTask: {
-			location: {
-				otelKeys: []string{
-					semconv.AttributeCloudAvailabilityZone,
-					semconv.AttributeCloudRegion,
-				},
-				fallbackLiteral: "global",
-			},
-			namespace: {otelKeys: []string{semconv.AttributeServiceNamespace}},
-			job:       {otelKeys: []string{semconv.AttributeServiceName}},
-			taskID:    {otelKeys: []string{semconv.AttributeServiceInstanceID}},
-		},
-		genericNode: {
-			location: {
-				otelKeys: []string{
-					semconv.AttributeCloudAvailabilityZone,
-					semconv.AttributeCloudRegion,
-				},
-				fallbackLiteral: "global",
-			},
-			namespace: {otelKeys: []string{semconv.AttributeServiceNamespace}},
-			nodeID:    {otelKeys: []string{semconv.AttributeHostID, semconv.AttributeHostName}},
-		},
+type attributes struct {
+	Attrs *pdata.Map
+}
+
+func (attrs *attributes) GetString(key string) (string, bool) {
+	value, ok := attrs.Attrs.Get(key)
+	if ok {
+		return value.AsString(), ok
 	}
-)
+	return "", false
+}
 
 // Transforms pdata Resource to a GCM Monitored Resource. Any additional metric labels from the
 // resource are also returned.
@@ -122,66 +43,16 @@ func (m *metricMapper) resourceToMonitoredResource(
 	resource pdata.Resource,
 ) (*monitoredrespb.MonitoredResource, labels) {
 	attrs := resource.Attributes()
-	cloudPlatform := getStringOrEmpty(attrs, semconv.AttributeCloudPlatform)
-
-	var mr *monitoredrespb.MonitoredResource
-	switch cloudPlatform {
-	case semconv.AttributeCloudPlatformGCPComputeEngine:
-		mr = createMonitoredResource(gceInstance, attrs)
-	case semconv.AttributeCloudPlatformGCPKubernetesEngine:
-		// Try for most to least specific k8s_container, k8s_pod, etc
-		if _, ok := attrs.Get(semconv.AttributeK8SContainerName); ok {
-			mr = createMonitoredResource(k8sContainer, attrs)
-		} else if _, ok := attrs.Get(semconv.AttributeK8SPodName); ok {
-			mr = createMonitoredResource(k8sPod, attrs)
-		} else if _, ok := attrs.Get(semconv.AttributeK8SNodeName); ok {
-			mr = createMonitoredResource(k8sNode, attrs)
-		} else {
-			mr = createMonitoredResource(k8sCluster, attrs)
-		}
-	case semconv.AttributeCloudPlatformAWSEC2:
-		mr = createMonitoredResource(awsEc2Instance, attrs)
-	default:
-		// Fallback to generic_task
-		_, hasServiceName := attrs.Get(semconv.AttributeServiceName)
-		_, hasServiceInstanceID := attrs.Get(semconv.AttributeServiceInstanceID)
-		if hasServiceName && hasServiceInstanceID {
-			mr = createMonitoredResource(genericTask, attrs)
-		} else {
-			// If not possible, fallback to generic_node
-			mr = createMonitoredResource(genericNode, attrs)
-		}
+	gmr := resourcemapping.ResourceAttributesToMonitoredResource(&attributes{
+		Attrs: &attrs,
+	})
+	mr := &monitoredrespb.MonitoredResource{
+		Type:   gmr.Type,
+		Labels: gmr.Labels,
 	}
 	return mr, m.resourceToMetricLabels(resource)
 }
 
-func createMonitoredResource(
-	monitoredResourceType string,
-	resourceAttrs pdata.Map,
-) *monitoredrespb.MonitoredResource {
-	mappings := monitoredResourceMappings[monitoredResourceType]
-	mrLabels := make(map[string]string, len(mappings))
-
-	for mrKey, mappingConfig := range mappings {
-		mrValue := ""
-		// Coalesce the possible keys in order
-		for _, otelKey := range mappingConfig.otelKeys {
-			mrValue = getStringOrEmpty(resourceAttrs, otelKey)
-			if mrValue != "" {
-				break
-			}
-		}
-		if mrValue == "" {
-			mrValue = mappingConfig.fallbackLiteral
-		}
-		mrLabels[mrKey] = mrValue
-	}
-	return &monitoredrespb.MonitoredResource{
-		Type:   monitoredResourceType,
-		Labels: mrLabels,
-	}
-}
-
 // resourceToMetricLabels converts the Resource into metric labels needed to uniquely identify
 // the timeseries.
 func (m *metricMapper) resourceToMetricLabels(
@@ -208,10 +79,3 @@ func (m *metricMapper) resourceToMetricLabels(
 	})
 	return attributesToLabels(attrs)
 }
-
-func getStringOrEmpty(attributes pdata.Map, key string) string {
-	if val, ok := attributes.Get(key); ok {
-		return val.StringVal()
-	}
-	return ""
-}
diff --git a/exporter/trace/go.mod b/exporter/trace/go.mod
index e1a8663..f65c527 100644
--- a/exporter/trace/go.mod
+++ b/exporter/trace/go.mod
@@ -9,9 +9,9 @@ require (
 	github.com/googleinterns/cloud-operations-api-mock v0.0.0-20200709193332-a1e58c29bdd3
 	github.com/stretchr/testify v1.7.1
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0
-	go.opentelemetry.io/otel v1.6.2
+	go.opentelemetry.io/otel v1.6.3
 	go.opentelemetry.io/otel/sdk v1.6.2
-	go.opentelemetry.io/otel/trace v1.6.2
+	go.opentelemetry.io/otel/trace v1.6.3
 	golang.org/x/net v0.0.0-20220325170049-de3da57026de // indirect
 	golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a
 	golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886 // indirect
@@ -21,6 +21,8 @@ require (
 	google.golang.org/protobuf v1.28.0
 )
 
+require github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.0.0-00010101000000-000000000000
+
 require (
 	cloud.google.com/go/compute v1.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.0 // indirect
@@ -36,3 +38,5 @@ require (
 	google.golang.org/appengine v1.6.7 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )
+
+replace github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping => ../../internal/resourcemapping
diff --git a/exporter/trace/go.sum b/exporter/trace/go.sum
index 9f4965f..8813312 100644
--- a/exporter/trace/go.sum
+++ b/exporter/trace/go.sum
@@ -214,16 +214,18 @@ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0 h1:woM+Mb4
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0/go.mod h1:PFmBsWbldL1kiWZk9+0LBZz2brhByaGsvp6pRICMlPE=
 go.opentelemetry.io/otel v1.6.0/go.mod h1:bfJD2DZVw0LBxghOTlgnlI0CV3hLDu9XF/QKOUXMTQQ=
 go.opentelemetry.io/otel v1.6.1/go.mod h1:blzUabWHkX6LJewxvadmzafgh/wnvBSDBdOuwkAtrWQ=
-go.opentelemetry.io/otel v1.6.2 h1:iLrj1F5ydCSVc4CGjBX2ou/QRav3C8xHjHlVoch6g4w=
 go.opentelemetry.io/otel v1.6.2/go.mod h1:MUBZHaB2cm6CahEBHQPq9Anos7IXynP/noVpjsxQTSc=
+go.opentelemetry.io/otel v1.6.3 h1:FLOfo8f9JzFVFVyU+MSRJc2HdEAXQgm7pIv2uFKRSZE=
+go.opentelemetry.io/otel v1.6.3/go.mod h1:7BgNga5fNlF/iZjG06hM3yofffp0ofKCDwSXx1GC4dI=
 go.opentelemetry.io/otel/metric v0.28.0 h1:o5YNh+jxACMODoAo1bI7OES0RUW4jAMae0Vgs2etWAQ=
 go.opentelemetry.io/otel/metric v0.28.0/go.mod h1:TrzsfQAmQaB1PDcdhBauLMk7nyyg9hm+GoQq/ekE9Iw=
 go.opentelemetry.io/otel/sdk v1.6.2 h1:wxY+YrfpGJfjxtm7SFBMJp9APDMZjDG+ErZOs/wkubg=
 go.opentelemetry.io/otel/sdk v1.6.2/go.mod h1:M2r4VCm1Yurk4E+fWtP2p+QzFDHMFEqhGdbtQ7zRf+k=
 go.opentelemetry.io/otel/trace v1.6.0/go.mod h1:qs7BrU5cZ8dXQHBGxHMOxwME/27YH2qEp4/+tZLLwJE=
 go.opentelemetry.io/otel/trace v1.6.1/go.mod h1:RkFRM1m0puWIq10oxImnGEduNBzxiN7TXluRBtE+5j0=
-go.opentelemetry.io/otel/trace v1.6.2 h1:oY7i1k6XD/ozlGo7ASy+H1UdkNcj9cPfuklaYSXtoFk=
 go.opentelemetry.io/otel/trace v1.6.2/go.mod h1:RMqfw8Mclba1p7sXDmEDBvrB8jw65F6GOoN1fyyXTzk=
+go.opentelemetry.io/otel/trace v1.6.3 h1:IqN4L+5b0mPNjdXIiZ90Ni4Bl5BRkDQywePLWemd9bc=
+go.opentelemetry.io/otel/trace v1.6.3/go.mod h1:GNJQusJlUgZl9/TQBPKU/Y/ty+0iVB5fjhKeJGZPGFs=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
diff --git a/exporter/trace/trace_proto.go b/exporter/trace/trace_proto.go
index 76fbb48..22a3aad 100644
--- a/exporter/trace/trace_proto.go
+++ b/exporter/trace/trace_proto.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"math"
 	"strconv"
+	"strings"
 	"time"
 	"unicode/utf8"
 
@@ -33,6 +34,8 @@ import (
 	tracepb "google.golang.org/genproto/googleapis/devtools/cloudtrace/v2"
 	codepb "google.golang.org/genproto/googleapis/rpc/code"
 	statuspb "google.golang.org/genproto/googleapis/rpc/status"
+
+	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping"
 )
 
 const (
@@ -61,10 +64,27 @@ const (
 	// This is prefixed for google app engine, but translates to the service
 	// in the trace UI
 	labelService = `g.co/gae/app/module`
+
+	instrumentationScopeNameAttribute    = "otel.scope.name"
+	instrumentationScopeVersionAttribute = "otel.scope.version"
 )
 
 var userAgent = fmt.Sprintf("opentelemetry-go %s; google-cloud-trace-exporter %s", otel.Version(), Version())
 
+// Adapters for using resourcemapping library
+type attrs struct {
+	Attrs []attribute.KeyValue
+}
+
+func (a *attrs) GetString(key string) (string, bool) {
+	for _, kv := range a.Attrs {
+		if kv.Key == attribute.Key(key) {
+			return kv.Value.AsString(), true
+		}
+	}
+	return "", false
+}
+
 // If there are duplicate keys present in the list of attributes,
 // then the first value found for the key is preserved.
 func attributeWithLabelsFromResources(sd sdktrace.ReadOnlySpan) []attribute.KeyValue {
@@ -73,9 +93,11 @@ func attributeWithLabelsFromResources(sd sdktrace.ReadOnlySpan) []attribute.KeyV
 		return attributes
 	}
 	uniqueAttrs := make(map[attribute.Key]bool, len(sd.Attributes()))
+	// Span Attributes take precedence
 	for _, attr := range sd.Attributes() {
 		uniqueAttrs[attr.Key] = true
 	}
+	// Raw resource attributes are next.
 	for _, attr := range sd.Resource().Attributes() {
 		if uniqueAttrs[attr.Key] {
 			continue // skip resource attributes which conflict with span attributes
@@ -83,7 +105,26 @@ func attributeWithLabelsFromResources(sd sdktrace.ReadOnlySpan) []attribute.KeyV
 		uniqueAttrs[attr.Key] = true
 		attributes = append(attributes, attr)
 	}
+	// Instrumentation Scope attributes come next.
+	if !uniqueAttrs[instrumentationScopeNameAttribute] {
+		uniqueAttrs[instrumentationScopeNameAttribute] = true
+		scopeNameAttrs := attribute.String(instrumentationScopeNameAttribute, sd.InstrumentationLibrary().Name)
+		attributes = append(attributes, scopeNameAttrs)
+	}
+	if !uniqueAttrs[instrumentationScopeVersionAttribute] && strings.Compare("", sd.InstrumentationLibrary().Version) != 0 {
+		uniqueAttrs[instrumentationScopeVersionAttribute] = true
+		scopeVersionAttrs := attribute.String(instrumentationScopeVersionAttribute, sd.InstrumentationLibrary().Version)
+		attributes = append(attributes, scopeVersionAttrs)
+	}
 
+	// Monitored resource attributes (`g.co/r/{resource_type}/{resource_label}`) come next.
+	gceResource := resourcemapping.ResourceAttributesToMonitoredResource(&attrs{
+		Attrs: sd.Resource().Attributes(),
+	})
+	for key, value := range gceResource.Labels {
+		name := fmt.Sprintf("g.co/r/%v/%v", gceResource.Type, key)
+		attributes = append(attributes, attribute.String(name, value))
+	}
 	return attributes
 }
 
diff --git a/internal/resourcemapping/go.mod b/internal/resourcemapping/go.mod
new file mode 100644
index 0000000..7b6ce97
--- /dev/null
+++ b/internal/resourcemapping/go.mod
@@ -0,0 +1,7 @@
+module github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping
+
+go 1.17
+
+require go.opentelemetry.io/otel v1.6.3
+
+require go.opentelemetry.io/otel/trace v1.6.3 // indirect
diff --git a/internal/resourcemapping/go.sum b/internal/resourcemapping/go.sum
new file mode 100644
index 0000000..ba74074
--- /dev/null
+++ b/internal/resourcemapping/go.sum
@@ -0,0 +1,20 @@
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
+github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+go.opentelemetry.io/otel v1.6.3 h1:FLOfo8f9JzFVFVyU+MSRJc2HdEAXQgm7pIv2uFKRSZE=
+go.opentelemetry.io/otel v1.6.3/go.mod h1:7BgNga5fNlF/iZjG06hM3yofffp0ofKCDwSXx1GC4dI=
+go.opentelemetry.io/otel/trace v1.6.3 h1:IqN4L+5b0mPNjdXIiZ90Ni4Bl5BRkDQywePLWemd9bc=
+go.opentelemetry.io/otel/trace v1.6.3/go.mod h1:GNJQusJlUgZl9/TQBPKU/Y/ty+0iVB5fjhKeJGZPGFs=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/internal/resourcemapping/resourcemapping.go b/internal/resourcemapping/resourcemapping.go
new file mode 100644
index 0000000..646a456
--- /dev/null
+++ b/internal/resourcemapping/resourcemapping.go
@@ -0,0 +1,187 @@
+// 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 resourcemapping
+
+import (
+	semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
+)
+
+const (
+	awsAccount     = "aws_account"
+	awsEc2Instance = "aws_ec2_instance"
+	clusterName    = "cluster_name"
+	containerName  = "container_name"
+	gceInstance    = "gce_instance"
+	genericNode    = "generic_node"
+	genericTask    = "generic_task"
+	instanceID     = "instance_id"
+	job            = "job"
+	k8sCluster     = "k8s_cluster"
+	k8sContainer   = "k8s_container"
+	k8sNode        = "k8s_node"
+	k8sPod         = "k8s_pod"
+	location       = "location"
+	namespace      = "namespace"
+	namespaceName  = "namespace_name"
+	nodeID         = "node_id"
+	nodeName       = "node_name"
+	podName        = "pod_name"
+	region         = "region"
+	taskID         = "task_id"
+	zone           = "zone"
+)
+
+var (
+	// monitoredResourceMappings contains mappings of GCM resource label keys onto mapping config from OTel
+	// resource for a given monitored resource type.
+	monitoredResourceMappings = map[string]map[string]struct {
+		// OTel resource keys to try and populate the resource label from. For entries with
+		// multiple OTel resource keys, the keys' values will be coalesced in order until there
+		// is a non-empty value.
+		otelKeys []string
+		// If none of the otelKeys are present in the Resource, fallback to this literal value
+		fallbackLiteral string
+	}{
+		gceInstance: {
+			zone:       {otelKeys: []string{string(semconv.CloudAvailabilityZoneKey)}},
+			instanceID: {otelKeys: []string{string(semconv.HostIDKey)}},
+		},
+		k8sContainer: {
+			location:      {otelKeys: []string{string(semconv.CloudAvailabilityZoneKey)}},
+			clusterName:   {otelKeys: []string{string(semconv.K8SClusterNameKey)}},
+			namespaceName: {otelKeys: []string{string(semconv.K8SNamespaceNameKey)}},
+			podName:       {otelKeys: []string{string(semconv.K8SPodNameKey)}},
+			containerName: {otelKeys: []string{string(semconv.K8SContainerNameKey)}},
+		},
+		k8sPod: {
+			location:      {otelKeys: []string{string(semconv.CloudAvailabilityZoneKey)}},
+			clusterName:   {otelKeys: []string{string(semconv.K8SClusterNameKey)}},
+			namespaceName: {otelKeys: []string{string(semconv.K8SNamespaceNameKey)}},
+			podName:       {otelKeys: []string{string(semconv.K8SPodNameKey)}},
+		},
+		k8sNode: {
+			location:    {otelKeys: []string{string(semconv.CloudAvailabilityZoneKey)}},
+			clusterName: {otelKeys: []string{string(semconv.K8SClusterNameKey)}},
+			nodeName:    {otelKeys: []string{string(semconv.K8SNodeNameKey)}},
+		},
+		k8sCluster: {
+			location:    {otelKeys: []string{string(semconv.CloudAvailabilityZoneKey)}},
+			clusterName: {otelKeys: []string{string(semconv.K8SClusterNameKey)}},
+		},
+		awsEc2Instance: {
+			instanceID: {otelKeys: []string{string(semconv.HostIDKey)}},
+			region:     {otelKeys: []string{string(semconv.CloudAvailabilityZoneKey)}},
+			awsAccount: {otelKeys: []string{string(semconv.CloudAccountIDKey)}},
+		},
+		genericTask: {
+			location: {
+				otelKeys: []string{
+					string(semconv.CloudAvailabilityZoneKey),
+					string(semconv.CloudRegionKey),
+				},
+				fallbackLiteral: "global",
+			},
+			namespace: {otelKeys: []string{string(semconv.ServiceNamespaceKey)}},
+			job:       {otelKeys: []string{string(semconv.ServiceNameKey)}},
+			taskID:    {otelKeys: []string{string(semconv.ServiceInstanceIDKey)}},
+		},
+		genericNode: {
+			location: {
+				otelKeys: []string{
+					string(semconv.CloudAvailabilityZoneKey),
+					string(semconv.CloudRegionKey),
+				},
+				fallbackLiteral: "global",
+			},
+			namespace: {otelKeys: []string{string(semconv.ServiceNamespaceKey)}},
+			nodeID:    {otelKeys: []string{string(semconv.HostIDKey), string(semconv.HostNameKey)}},
+		},
+	}
+)
+
+type GceResource struct {
+	Type   string
+	Labels map[string]string
+}
+
+// ReadOnlyAttributes is an interface to abstract between pulling attributes from PData library or OTEL SDK.
+type ReadOnlyAttributes interface {
+	GetString(string) (string, bool)
+}
+
+// ResourceAttributesToMonitoredResource converts from a set of OTEL resource attributes into a
+// GCP monitored resource type and label set.
+// E.g.
+// This may output `gce_instance` type with appropriate labels.
+func ResourceAttributesToMonitoredResource(attrs ReadOnlyAttributes) *GceResource {
+	cloudPlatform, _ := attrs.GetString(string(semconv.CloudPlatformKey))
+	var mr *GceResource
+	switch cloudPlatform {
+	case semconv.CloudPlatformGCPComputeEngine.Value.AsString():
+		mr = createMonitoredResource(gceInstance, attrs)
+	case semconv.CloudPlatformGCPKubernetesEngine.Value.AsString():
+		// Try for most to least specific k8s_container, k8s_pod, etc
+		if _, ok := attrs.GetString(string(semconv.K8SContainerNameKey)); ok {
+			mr = createMonitoredResource(k8sContainer, attrs)
+		} else if _, ok := attrs.GetString(string(semconv.K8SPodNameKey)); ok {
+			mr = createMonitoredResource(k8sPod, attrs)
+		} else if _, ok := attrs.GetString(string(semconv.K8SNodeNameKey)); ok {
+			mr = createMonitoredResource(k8sNode, attrs)
+		} else {
+			mr = createMonitoredResource(k8sCluster, attrs)
+		}
+	case semconv.CloudPlatformAWSEC2.Value.AsString():
+		mr = createMonitoredResource(awsEc2Instance, attrs)
+	default:
+		// Fallback to generic_task
+		_, hasServiceName := attrs.GetString(string(semconv.ServiceNameKey))
+		_, hasServiceInstanceID := attrs.GetString(string(semconv.ServiceInstanceIDKey))
+		if hasServiceName && hasServiceInstanceID {
+			mr = createMonitoredResource(genericTask, attrs)
+		} else {
+			// If not possible, fallback to generic_node
+			mr = createMonitoredResource(genericNode, attrs)
+		}
+	}
+	return mr
+}
+
+func createMonitoredResource(
+	monitoredResourceType string,
+	resourceAttrs ReadOnlyAttributes,
+) *GceResource {
+	mappings := monitoredResourceMappings[monitoredResourceType]
+	mrLabels := make(map[string]string, len(mappings))
+
+	for mrKey, mappingConfig := range mappings {
+		mrValue := ""
+		ok := false
+		// Coalesce the possible keys in order
+		for _, otelKey := range mappingConfig.otelKeys {
+			mrValue, ok = resourceAttrs.GetString(otelKey)
+			if mrValue != "" {
+				break
+			}
+		}
+		if !ok || mrValue == "" {
+			mrValue = mappingConfig.fallbackLiteral
+		}
+		mrLabels[mrKey] = mrValue
+	}
+	return &GceResource{
+		Type:   monitoredResourceType,
+		Labels: mrLabels,
+	}
+}


Test patch:
diff --git a/e2e-test-server/Dockerfile b/e2e-test-server/Dockerfile
index 8648385..01792a4 100644
--- a/e2e-test-server/Dockerfile
+++ b/e2e-test-server/Dockerfile
@@ -18,6 +18,7 @@ WORKDIR /workspace/e2e-test-server/
 
 # In-repo dependencies
 COPY exporter/trace/ ../exporter/trace/
+COPY internal/resourcemapping/ ../internal/resourcemapping
 
 # cache deps before copying source so that source changes don't invalidate our
 # downloaded layer and we don't need to re-download as much.
diff --git a/e2e-test-server/go.mod b/e2e-test-server/go.mod
index 25482d2..4545cc2 100644
--- a/e2e-test-server/go.mod
+++ b/e2e-test-server/go.mod
@@ -6,9 +6,9 @@ require (
 	cloud.google.com/go/iam v0.1.1 // indirect
 	cloud.google.com/go/pubsub v1.19.0
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace v1.4.0
-	go.opentelemetry.io/otel v1.6.2
+	go.opentelemetry.io/otel v1.6.3
 	go.opentelemetry.io/otel/sdk v1.6.2
-	go.opentelemetry.io/otel/trace v1.6.2
+	go.opentelemetry.io/otel/trace v1.6.3
 	google.golang.org/genproto v0.0.0-20220405205423-9d709892a2bf
 )
 
@@ -16,6 +16,7 @@ require (
 	cloud.google.com/go v0.100.2 // indirect
 	cloud.google.com/go/compute v1.5.0 // indirect
 	cloud.google.com/go/trace v1.2.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.0.0-00010101000000-000000000000 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
@@ -36,4 +37,6 @@ require (
 
 replace github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace => ../exporter/trace
 
+replace github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping => ../internal/resourcemapping
+
 retract v1.0.0-RC1
diff --git a/e2e-test-server/go.sum b/e2e-test-server/go.sum
index ce8dc4d..43b96cb 100644
--- a/e2e-test-server/go.sum
+++ b/e2e-test-server/go.sum
@@ -215,13 +215,15 @@ go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
 go.opencensus.io v0.23.0 h1:gqCw0LfLxScz8irSi8exQc7fyQ0fKQU/qnC/X8+V/1M=
 go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.31.0 h1:woM+Mb4d0A+Dxa3rYPenSN5ZeS9qHUvE8rlObiLRXTY=
-go.opentelemetry.io/otel v1.6.2 h1:iLrj1F5ydCSVc4CGjBX2ou/QRav3C8xHjHlVoch6g4w=
 go.opentelemetry.io/otel v1.6.2/go.mod h1:MUBZHaB2cm6CahEBHQPq9Anos7IXynP/noVpjsxQTSc=
+go.opentelemetry.io/otel v1.6.3 h1:FLOfo8f9JzFVFVyU+MSRJc2HdEAXQgm7pIv2uFKRSZE=
+go.opentelemetry.io/otel v1.6.3/go.mod h1:7BgNga5fNlF/iZjG06hM3yofffp0ofKCDwSXx1GC4dI=
 go.opentelemetry.io/otel/metric v0.28.0 h1:o5YNh+jxACMODoAo1bI7OES0RUW4jAMae0Vgs2etWAQ=
 go.opentelemetry.io/otel/sdk v1.6.2 h1:wxY+YrfpGJfjxtm7SFBMJp9APDMZjDG+ErZOs/wkubg=
 go.opentelemetry.io/otel/sdk v1.6.2/go.mod h1:M2r4VCm1Yurk4E+fWtP2p+QzFDHMFEqhGdbtQ7zRf+k=
-go.opentelemetry.io/otel/trace v1.6.2 h1:oY7i1k6XD/ozlGo7ASy+H1UdkNcj9cPfuklaYSXtoFk=
 go.opentelemetry.io/otel/trace v1.6.2/go.mod h1:RMqfw8Mclba1p7sXDmEDBvrB8jw65F6GOoN1fyyXTzk=
+go.opentelemetry.io/otel/trace v1.6.3 h1:IqN4L+5b0mPNjdXIiZ90Ni4Bl5BRkDQywePLWemd9bc=
+go.opentelemetry.io/otel/trace v1.6.3/go.mod h1:GNJQusJlUgZl9/TQBPKU/Y/ty+0iVB5fjhKeJGZPGFs=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
diff --git a/exporter/collector/integrationtest/config/config_test.go b/exporter/collector/integrationtest/config/config_test.go
index 1ad4bc0..62626d3 100644
--- a/exporter/collector/integrationtest/config/config_test.go
+++ b/exporter/collector/integrationtest/config/config_test.go
@@ -74,6 +74,7 @@ func TestLoadConfig(t *testing.T) {
 					InstrumentationLibraryLabels:     true,
 					CreateMetricDescriptorBufferSize: 10,
 					ServiceResourceLabels:            true,
+					CumulativeNormalization:          true,
 				},
 			},
 		})
diff --git a/exporter/collector/integrationtest/go.mod b/exporter/collector/integrationtest/go.mod
index b00d8ce..5834f29 100644
--- a/exporter/collector/integrationtest/go.mod
+++ b/exporter/collector/integrationtest/go.mod
@@ -5,7 +5,6 @@ go 1.17
 require (
 	contrib.go.opencensus.io/exporter/stackdriver v0.13.11
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector v0.28.0
-	github.com/aws/aws-sdk-go v1.42.49 // indirect
 	github.com/google/go-cmp v0.5.7
 	github.com/stretchr/testify v1.7.1
 	go.opencensus.io v0.23.0
@@ -24,6 +23,8 @@ require (
 	cloud.google.com/go/monitoring v1.4.0 // indirect
 	cloud.google.com/go/trace v1.2.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace v1.4.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.0.0-00010101000000-000000000000 // indirect
+	github.com/aws/aws-sdk-go v1.42.49 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logr/logr v1.2.3 // indirect
@@ -58,4 +59,5 @@ require (
 replace (
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector => ../../collector
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace => ../../trace
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping => ../../../internal/resourcemapping
 )
diff --git a/exporter/collector/internal/datapointstorage/datapointcache_test.go b/exporter/collector/internal/datapointstorage/datapointcache_test.go
index 3066882..8cf743d 100644
--- a/exporter/collector/internal/datapointstorage/datapointcache_test.go
+++ b/exporter/collector/internal/datapointstorage/datapointcache_test.go
@@ -25,27 +25,34 @@ import (
 )
 
 func TestSetAndGet(t *testing.T) {
-	c := make(Cache)
-	c.Set("foo", nil)
-	point, found := c.Get("foo")
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
+	c.SetNumberDataPoint("foo", nil)
+	point, found := c.GetNumberDataPoint("foo")
 	assert.Nil(t, point)
 	assert.True(t, found)
-
-	point, found = c.Get("bar")
+	point, found = c.GetNumberDataPoint("bar")
 	assert.Nil(t, point)
 	assert.False(t, found)
-
 	setPoint := pdata.NewNumberDataPoint()
-	c.Set("bar", &setPoint)
-
-	point, found = c.Get("bar")
+	c.SetNumberDataPoint("bar", &setPoint)
+	point, found = c.GetNumberDataPoint("bar")
 	assert.Equal(t, point, &setPoint)
 	assert.True(t, found)
 }
 
 func TestShutdown(t *testing.T) {
 	shutdown := make(chan struct{})
-	c := make(Cache)
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
 	close(shutdown)
 	// gc should return after shutdown is closed
 	cont := c.gc(shutdown, make(chan time.Time))
@@ -54,13 +61,18 @@ func TestShutdown(t *testing.T) {
 
 func TestGC(t *testing.T) {
 	shutdown := make(chan struct{})
-	c := make(Cache)
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
 	fakeTicker := make(chan time.Time)
 
-	c.Set("bar", nil)
+	c.SetNumberDataPoint("bar", nil)
 
 	// bar exists since we just set it
-	usedPoint, found := c["bar"]
+	usedPoint, found := c.numberCache["bar"]
 	assert.True(t, usedPoint.used)
 	assert.True(t, found)
 
@@ -70,7 +82,7 @@ func TestGC(t *testing.T) {
 	}()
 	cont := c.gc(shutdown, fakeTicker)
 	assert.True(t, cont)
-	usedPoint, found = c["bar"]
+	usedPoint, found = c.numberCache["bar"]
 	assert.False(t, usedPoint.used)
 	assert.True(t, found)
 
@@ -80,22 +92,25 @@ func TestGC(t *testing.T) {
 	}()
 	cont = c.gc(shutdown, fakeTicker)
 	assert.True(t, cont)
-	_, found = c["bar"]
+	_, found = c.numberCache["bar"]
 	assert.False(t, found)
 }
 
 func TestGetPreventsGC(t *testing.T) {
 	shutdown := make(chan struct{})
-	c := make(Cache)
+	c := Cache{
+		numberCache:               make(map[string]usedNumberPoint),
+		summaryCache:              make(map[string]usedSummaryPoint),
+		histogramCache:            make(map[string]usedHistogramPoint),
+		exponentialHistogramCache: make(map[string]usedExponentialHistogramPoint),
+	}
 	fakeTicker := make(chan time.Time)
 
 	setPoint := pdata.NewNumberDataPoint()
-	c.Set("bar", &setPoint)
-
+	c.SetNumberDataPoint("bar", &setPoint)
 	// bar exists since we just set it
-	_, found := c["bar"]
+	_, found := c.numberCache["bar"]
 	assert.True(t, found)
-
 	// first gc tick marks bar stale
 	go func() {
 		fakeTicker <- time.Now()
@@ -103,16 +118,15 @@ func TestGetPreventsGC(t *testing.T) {
 	cont := c.gc(shutdown, fakeTicker)
 	assert.True(t, cont)
 	// calling Get() marks it fresh again.
-	_, found = c.Get("bar")
+	_, found = c.GetNumberDataPoint("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"]
+	_, found = c.numberCache["bar"]
 	assert.True(t, found)
 }
 
diff --git a/exporter/collector/metrics_test.go b/exporter/collector/metrics_test.go
index 3a171d6..ae19856 100644
--- a/exporter/collector/metrics_test.go
+++ b/exporter/collector/metrics_test.go
@@ -34,7 +34,7 @@ import (
 	"google.golang.org/protobuf/testing/protocmp"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
-	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/datapointstorage"
+	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/normalization"
 )
 
 var (
@@ -49,7 +49,7 @@ func newTestMetricMapper() (metricMapper, func()) {
 	return metricMapper{
 		obs,
 		cfg,
-		datapointstorage.NewCache(s),
+		normalization.NewStandardNormalizer(s, zap.NewNop()),
 	}, func() { close(s) }
 }
 
@@ -322,6 +322,143 @@ func TestHistogramPointToTimeSeries(t *testing.T) {
 	assert.Equal(t, map[string]string{"test": "extra"}, dropped.Label)
 }
 
+func TestHistogramPointWithoutTimestampToTimeSeries(t *testing.T) {
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
+	mapper.cfg.ProjectID = "myproject"
+	mr := &monitoredrespb.MonitoredResource{}
+	metric := pdata.NewMetric()
+	metric.SetName("myhist")
+	metric.SetDataType(pmetric.MetricDataTypeHistogram)
+	unit := "1"
+	metric.SetUnit(unit)
+	hist := metric.Histogram()
+	hist.SetAggregationTemporality(pmetric.MetricAggregationTemporalityCumulative)
+	point := hist.DataPoints().AppendEmpty()
+	// leave start time unset with the intended start time
+	point.SetTimestamp(pdata.NewTimestampFromTime(start))
+	point.SetBucketCounts([]uint64{1, 0, 0, 0, 0})
+	point.SetCount(1)
+	point.SetSum(2)
+	point.SetExplicitBounds([]float64{10, 20, 30, 40})
+	exemplar := point.Exemplars().AppendEmpty()
+	exemplar.SetDoubleVal(2)
+	exemplar.SetTimestamp(pdata.NewTimestampFromTime(start))
+	exemplar.SetTraceID(pdata.NewTraceID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6}))
+	exemplar.SetSpanID(pdata.NewSpanID([8]byte{0, 1, 2, 3, 4, 5, 6, 7}))
+	exemplar.FilteredAttributes().InsertString("test", "extra")
+
+	// second point
+	point = hist.DataPoints().AppendEmpty()
+	end := start.Add(time.Hour)
+	// leave start time unset
+	point.SetTimestamp(pdata.NewTimestampFromTime(end))
+	point.SetBucketCounts([]uint64{2, 2, 3, 4, 5})
+	point.SetCount(16)
+	point.SetSum(44)
+	point.SetExplicitBounds([]float64{10, 20, 30, 40})
+	exemplar = point.Exemplars().AppendEmpty()
+	exemplar.SetDoubleVal(2)
+	exemplar.SetTimestamp(pdata.NewTimestampFromTime(end))
+	exemplar.SetTraceID(pdata.NewTraceID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6}))
+	exemplar.SetSpanID(pdata.NewSpanID([8]byte{0, 1, 2, 3, 4, 5, 6, 7}))
+	exemplar.FilteredAttributes().InsertString("test", "extra")
+
+	// third point
+	point = hist.DataPoints().AppendEmpty()
+	end2 := end.Add(time.Hour)
+	// leave start time unset
+	point.SetTimestamp(pdata.NewTimestampFromTime(end2))
+	point.SetBucketCounts([]uint64{3, 2, 3, 4, 5})
+	point.SetCount(17)
+	point.SetSum(445)
+	point.SetExplicitBounds([]float64{10, 20, 30, 40})
+	exemplar = point.Exemplars().AppendEmpty()
+	exemplar.SetDoubleVal(2)
+	exemplar.SetTimestamp(pdata.NewTimestampFromTime(end2))
+	exemplar.SetTraceID(pdata.NewTraceID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6}))
+	exemplar.SetSpanID(pdata.NewSpanID([8]byte{0, 1, 2, 3, 4, 5, 6, 7}))
+	exemplar.FilteredAttributes().InsertString("test", "extra")
+
+	tsl := mapper.metricToTimeSeries(mr, labels{}, metric)
+	// the first point should be dropped, so we expect 2 points
+	assert.Len(t, tsl, 2)
+	ts := tsl[0]
+	// Verify aspects
+	assert.Equal(t, metricpb.MetricDescriptor_CUMULATIVE, ts.MetricKind)
+	assert.Equal(t, metricpb.MetricDescriptor_DISTRIBUTION, ts.ValueType)
+	assert.Equal(t, unit, ts.Unit)
+	assert.Same(t, mr, ts.Resource)
+
+	assert.Equal(t, "workload.googleapis.com/myhist", ts.Metric.Type)
+	assert.Equal(t, map[string]string{}, ts.Metric.Labels)
+
+	assert.Nil(t, ts.Metadata)
+
+	assert.Len(t, ts.Points, 1)
+	assert.Equal(t, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end),
+	}, ts.Points[0].Interval)
+	hdp := ts.Points[0].Value.GetDistributionValue()
+	assert.Equal(t, int64(15), hdp.Count)
+	assert.ElementsMatch(t, []int64{1, 2, 3, 4, 5}, hdp.BucketCounts)
+	assert.Equal(t, float64(2.8), hdp.Mean)
+	assert.Equal(t, []float64{10, 20, 30, 40}, hdp.BucketOptions.GetExplicitBuckets().Bounds)
+	assert.Len(t, hdp.Exemplars, 1)
+	ex := hdp.Exemplars[0]
+	assert.Equal(t, float64(2), ex.Value)
+	assert.Equal(t, timestamppb.New(end), ex.Timestamp)
+	// We should see trace + dropped labels
+	assert.Len(t, ex.Attachments, 2)
+	spanctx := &monitoringpb.SpanContext{}
+	err := ex.Attachments[0].UnmarshalTo(spanctx)
+	assert.Nil(t, err)
+	assert.Equal(t, "projects/myproject/traces/00010203040506070809010203040506/spans/0001020304050607", spanctx.SpanName)
+	dropped := &monitoringpb.DroppedLabels{}
+	err = ex.Attachments[1].UnmarshalTo(dropped)
+	assert.Nil(t, err)
+	assert.Equal(t, map[string]string{"test": "extra"}, dropped.Label)
+
+	// verify second point
+	ts = tsl[1]
+	// Verify aspects
+	assert.Equal(t, metricpb.MetricDescriptor_CUMULATIVE, ts.MetricKind)
+	assert.Equal(t, metricpb.MetricDescriptor_DISTRIBUTION, ts.ValueType)
+	assert.Equal(t, unit, ts.Unit)
+	assert.Same(t, mr, ts.Resource)
+
+	assert.Equal(t, "workload.googleapis.com/myhist", ts.Metric.Type)
+	assert.Equal(t, map[string]string{}, ts.Metric.Labels)
+
+	assert.Nil(t, ts.Metadata)
+
+	assert.Len(t, ts.Points, 1)
+	assert.Equal(t, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end2),
+	}, ts.Points[0].Interval)
+	hdp = ts.Points[0].Value.GetDistributionValue()
+	assert.Equal(t, int64(16), hdp.Count)
+	assert.ElementsMatch(t, []int64{2, 2, 3, 4, 5}, hdp.BucketCounts)
+	assert.Equal(t, float64(27.6875), hdp.Mean)
+	assert.Equal(t, []float64{10, 20, 30, 40}, hdp.BucketOptions.GetExplicitBuckets().Bounds)
+	assert.Len(t, hdp.Exemplars, 1)
+	ex = hdp.Exemplars[0]
+	assert.Equal(t, float64(2), ex.Value)
+	assert.Equal(t, timestamppb.New(end2), ex.Timestamp)
+	// We should see trace + dropped labels
+	assert.Len(t, ex.Attachments, 2)
+	spanctx = &monitoringpb.SpanContext{}
+	err = ex.Attachments[0].UnmarshalTo(spanctx)
+	assert.Nil(t, err)
+	assert.Equal(t, "projects/myproject/traces/00010203040506070809010203040506/spans/0001020304050607", spanctx.SpanName)
+	dropped = &monitoringpb.DroppedLabels{}
+	err = ex.Attachments[1].UnmarshalTo(dropped)
+	assert.Nil(t, err)
+	assert.Equal(t, map[string]string{"test": "extra"}, dropped.Label)
+}
+
 func TestNoValueHistogramPointToTimeSeries(t *testing.T) {
 	mapper, shutdown := newTestMetricMapper()
 	defer shutdown()
@@ -498,7 +635,7 @@ func TestExponentialHistogramPointToTimeSeries(t *testing.T) {
 	// Add a second point with no value
 	hist.DataPoints().AppendEmpty().SetFlags(pmetric.MetricDataPointFlags(pmetric.MetricDataPointFlagNoRecordedValue))
 
-	tsl := mapper.exponentialHistogramToTimeSeries(mr, labels{}, metric, hist, point)
+	tsl := mapper.metricToTimeSeries(mr, labels{}, metric)
 	assert.Len(t, tsl, 1)
 	ts := tsl[0]
 	// Verify aspects
@@ -540,6 +677,161 @@ func TestExponentialHistogramPointToTimeSeries(t *testing.T) {
 	assert.Equal(t, map[string]string{"test": "extra"}, dropped.Label)
 }
 
+func TestExponentialHistogramPointWithoutStartTimeToTimeSeries(t *testing.T) {
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
+	mapper.cfg.ProjectID = "myproject"
+	mr := &monitoredrespb.MonitoredResource{}
+	metric := pdata.NewMetric()
+	metric.SetName("myexphist")
+	metric.SetDataType(pmetric.MetricDataTypeExponentialHistogram)
+	unit := "1"
+	metric.SetUnit(unit)
+	hist := metric.ExponentialHistogram()
+	hist.SetAggregationTemporality(pmetric.MetricAggregationTemporalityCumulative)
+
+	// First point will be dropped, since it has no start time.
+	point := hist.DataPoints().AppendEmpty()
+	// Omit start timestamp
+	point.SetTimestamp(pdata.NewTimestampFromTime(start))
+	point.Positive().SetOffset(2)
+	point.Positive().SetBucketCounts([]uint64{1, 1, 1})
+	point.Negative().SetOffset(1)
+	point.Negative().SetBucketCounts([]uint64{1, 1})
+	point.SetZeroCount(2)
+	point.SetCount(7)
+	point.SetScale(-1)
+	point.SetSum(10)
+	exemplar := point.Exemplars().AppendEmpty()
+	exemplar.SetDoubleVal(2)
+	exemplar.SetTimestamp(pdata.NewTimestampFromTime(start))
+	exemplar.SetTraceID(pdata.NewTraceID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6}))
+	exemplar.SetSpanID(pdata.NewSpanID([8]byte{0, 1, 2, 3, 4, 5, 6, 7}))
+	exemplar.FilteredAttributes().InsertString("test", "extra")
+
+	// Second point
+	point = hist.DataPoints().AppendEmpty()
+	end := start.Add(time.Hour)
+	// Omit start timestamp
+	point.SetTimestamp(pdata.NewTimestampFromTime(end))
+	point.Positive().SetOffset(1)
+	point.Positive().SetBucketCounts([]uint64{1, 2, 3, 4})
+	point.Negative().SetOffset(0)
+	point.Negative().SetBucketCounts([]uint64{1, 5, 6})
+	point.SetZeroCount(8)
+	point.SetCount(30)
+	point.SetScale(-1)
+	point.SetSum(56)
+	exemplar = point.Exemplars().AppendEmpty()
+	exemplar.SetDoubleVal(2)
+	exemplar.SetTimestamp(pdata.NewTimestampFromTime(end))
+	exemplar.SetTraceID(pdata.NewTraceID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6}))
+	exemplar.SetSpanID(pdata.NewSpanID([8]byte{0, 1, 2, 3, 4, 5, 6, 7}))
+	exemplar.FilteredAttributes().InsertString("test", "extra")
+
+	// Third point
+	point = hist.DataPoints().AppendEmpty()
+	end2 := end.Add(time.Hour)
+	// Omit start timestamp
+	point.SetTimestamp(pdata.NewTimestampFromTime(end2))
+	point.Positive().SetOffset(0)
+	point.Positive().SetBucketCounts([]uint64{1, 1, 3, 4, 5})
+	point.Negative().SetOffset(0)
+	point.Negative().SetBucketCounts([]uint64{1, 6, 7})
+	point.SetZeroCount(10)
+	point.SetCount(38)
+	point.SetScale(-1)
+	point.SetSum(72)
+	exemplar = point.Exemplars().AppendEmpty()
+	exemplar.SetDoubleVal(2)
+	exemplar.SetTimestamp(pdata.NewTimestampFromTime(end2))
+	exemplar.SetTraceID(pdata.NewTraceID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6}))
+	exemplar.SetSpanID(pdata.NewSpanID([8]byte{0, 1, 2, 3, 4, 5, 6, 7}))
+	exemplar.FilteredAttributes().InsertString("test", "extra")
+
+	tsl := mapper.metricToTimeSeries(mr, labels{}, metric)
+	// expect 2 timeseries, since the first is dropped
+	assert.Len(t, tsl, 2)
+	ts := tsl[0]
+	// Verify aspects
+	assert.Equal(t, metricpb.MetricDescriptor_CUMULATIVE, ts.MetricKind)
+	assert.Equal(t, metricpb.MetricDescriptor_DISTRIBUTION, ts.ValueType)
+	assert.Equal(t, unit, ts.Unit)
+	assert.Same(t, mr, ts.Resource)
+
+	assert.Equal(t, "workload.googleapis.com/myexphist", ts.Metric.Type)
+	assert.Equal(t, map[string]string{}, ts.Metric.Labels)
+
+	assert.Nil(t, ts.Metadata)
+
+	assert.Len(t, ts.Points, 1)
+	assert.Equal(t, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end),
+	}, ts.Points[0].Interval)
+	hdp := ts.Points[0].Value.GetDistributionValue()
+	assert.Equal(t, int64(23), hdp.Count)
+	assert.ElementsMatch(t, []int64{16, 1, 1, 2, 3, 0}, hdp.BucketCounts)
+	assert.Equal(t, float64(2), hdp.Mean)
+	assert.Equal(t, float64(4), hdp.BucketOptions.GetExponentialBuckets().GrowthFactor)
+	assert.Equal(t, float64(4), hdp.BucketOptions.GetExponentialBuckets().Scale)
+	assert.Equal(t, int32(4), hdp.BucketOptions.GetExponentialBuckets().NumFiniteBuckets)
+	assert.Len(t, hdp.Exemplars, 1)
+	ex := hdp.Exemplars[0]
+	assert.Equal(t, float64(2), ex.Value)
+	assert.Equal(t, timestamppb.New(end), ex.Timestamp)
+	// We should see trace + dropped labels
+	assert.Len(t, ex.Attachments, 2)
+	spanctx := &monitoringpb.SpanContext{}
+	err := ex.Attachments[0].UnmarshalTo(spanctx)
+	assert.Nil(t, err)
+	assert.Equal(t, "projects/myproject/traces/00010203040506070809010203040506/spans/0001020304050607", spanctx.SpanName)
+	dropped := &monitoringpb.DroppedLabels{}
+	err = ex.Attachments[1].UnmarshalTo(dropped)
+	assert.Nil(t, err)
+	assert.Equal(t, map[string]string{"test": "extra"}, dropped.Label)
+
+	// Check the second point
+	ts = tsl[1]
+	// Verify aspects
+	assert.Equal(t, metricpb.MetricDescriptor_CUMULATIVE, ts.MetricKind)
+	assert.Equal(t, metricpb.MetricDescriptor_DISTRIBUTION, ts.ValueType)
+	assert.Equal(t, unit, ts.Unit)
+	assert.Same(t, mr, ts.Resource)
+
+	assert.Equal(t, "workload.googleapis.com/myexphist", ts.Metric.Type)
+	assert.Equal(t, map[string]string{}, ts.Metric.Labels)
+
+	assert.Nil(t, ts.Metadata)
+
+	assert.Len(t, ts.Points, 1)
+	assert.Equal(t, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end2),
+	}, ts.Points[0].Interval)
+	hdp = ts.Points[0].Value.GetDistributionValue()
+	assert.Equal(t, int64(31), hdp.Count)
+	assert.ElementsMatch(t, []int64{20, 1, 1, 2, 3, 4, 0}, hdp.BucketCounts)
+	assert.Equal(t, float64(2), hdp.Mean)
+	assert.Equal(t, float64(4), hdp.BucketOptions.GetExponentialBuckets().GrowthFactor)
+	assert.Equal(t, float64(1), hdp.BucketOptions.GetExponentialBuckets().Scale)
+	assert.Equal(t, int32(5), hdp.BucketOptions.GetExponentialBuckets().NumFiniteBuckets)
+	assert.Len(t, hdp.Exemplars, 1)
+	ex = hdp.Exemplars[0]
+	assert.Equal(t, float64(2), ex.Value)
+	assert.Equal(t, timestamppb.New(end2), ex.Timestamp)
+	// We should see trace + dropped labels
+	assert.Len(t, ex.Attachments, 2)
+	spanctx = &monitoringpb.SpanContext{}
+	err = ex.Attachments[0].UnmarshalTo(spanctx)
+	assert.Nil(t, err)
+	assert.Equal(t, "projects/myproject/traces/00010203040506070809010203040506/spans/0001020304050607", spanctx.SpanName)
+	dropped = &monitoringpb.DroppedLabels{}
+	err = ex.Attachments[1].UnmarshalTo(dropped)
+	assert.Nil(t, err)
+	assert.Equal(t, map[string]string{"test": "extra"}, dropped.Label)
+}
+
 func TestNaNSumExponentialHistogramPointToTimeSeries(t *testing.T) {
 	mapper, shutdown := newTestMetricMapper()
 	defer shutdown()
@@ -970,6 +1262,149 @@ func TestSummaryPointToTimeSeries(t *testing.T) {
 	assert.Equal(t, quantileResult.Points[0].Value.GetDoubleValue(), 1.0)
 }
 
+func TestSummaryPointWithoutStartTimeToTimeSeries(t *testing.T) {
+	mapper, shutdown := newTestMetricMapper()
+	defer shutdown()
+	mr := &monitoredrespb.MonitoredResource{}
+
+	metric := pdata.NewMetric()
+	metric.SetDataType(pmetric.MetricDataTypeSummary)
+	summary := metric.Summary()
+	point := summary.DataPoints().AppendEmpty()
+
+	metric.SetName("mysummary")
+	metric.SetUnit("1")
+	point.SetCount(10)
+	point.SetSum(100.0)
+	quantile := point.QuantileValues().AppendEmpty()
+	quantile.SetQuantile(1.0)
+	quantile.SetValue(1.0)
+	// Don't set start timestamp.  This point will be dropped
+	point.SetTimestamp(pdata.NewTimestampFromTime(start))
+
+	point = summary.DataPoints().AppendEmpty()
+	metric.SetName("mysummary")
+	metric.SetUnit("1")
+	point.SetCount(20)
+	point.SetSum(200.0)
+	quantile = point.QuantileValues().AppendEmpty()
+	quantile.SetQuantile(1.0)
+	quantile.SetValue(1.0)
+	end := start.Add(time.Hour)
+	// Don't set start timestamp.  This point will be normalized
+	point.SetTimestamp(pdata.NewTimestampFromTime(end))
+
+	point = summary.DataPoints().AppendEmpty()
+	metric.SetName("mysummary")
+	metric.SetUnit("1")
+	point.SetCount(30)
+	point.SetSum(300.0)
+	quantile = point.QuantileValues().AppendEmpty()
+	quantile.SetQuantile(1.0)
+	quantile.SetValue(1.0)
+	end2 := end.Add(time.Hour)
+	// Don't set start timestamp.  This point will be normalized
+	point.SetTimestamp(pdata.NewTimestampFromTime(end2))
+
+	ts := mapper.metricToTimeSeries(mr, labels{}, metric)
+	assert.Len(t, ts, 6)
+	sumResult := ts[0]
+	countResult := ts[1]
+	quantileResult := ts[2]
+
+	// Test sum mapping
+	assert.Equal(t, sumResult.MetricKind, metricpb.MetricDescriptor_CUMULATIVE)
+	assert.Equal(t, sumResult.ValueType, metricpb.MetricDescriptor_DOUBLE)
+	assert.Equal(t, sumResult.Unit, "1")
+	assert.Same(t, sumResult.Resource, mr)
+	assert.Equal(t, sumResult.Metric.Type, "workload.googleapis.com/mysummary_sum")
+	assert.Equal(t, sumResult.Metric.Labels, map[string]string{})
+	assert.Nil(t, sumResult.Metadata)
+	assert.Len(t, sumResult.Points, 1)
+	assert.Equal(t, sumResult.Points[0].Interval, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end),
+	})
+	assert.Equal(t, sumResult.Points[0].Value.GetDoubleValue(), 100.0)
+	// Test count mapping
+	assert.Equal(t, countResult.MetricKind, metricpb.MetricDescriptor_CUMULATIVE)
+	assert.Equal(t, countResult.ValueType, metricpb.MetricDescriptor_DOUBLE)
+	assert.Equal(t, countResult.Unit, "1")
+	assert.Same(t, countResult.Resource, mr)
+	assert.Equal(t, countResult.Metric.Type, "workload.googleapis.com/mysummary_count")
+	assert.Equal(t, countResult.Metric.Labels, map[string]string{})
+	assert.Nil(t, countResult.Metadata)
+	assert.Len(t, countResult.Points, 1)
+	assert.Equal(t, countResult.Points[0].Interval, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end),
+	})
+	assert.Equal(t, countResult.Points[0].Value.GetDoubleValue(), float64(10))
+	// Test quantile mapping
+	assert.Equal(t, quantileResult.MetricKind, metricpb.MetricDescriptor_GAUGE)
+	assert.Equal(t, quantileResult.ValueType, metricpb.MetricDescriptor_DOUBLE)
+	assert.Equal(t, quantileResult.Unit, "1")
+	assert.Same(t, quantileResult.Resource, mr)
+	assert.Equal(t, quantileResult.Metric.Type, "workload.googleapis.com/mysummary")
+	assert.Equal(t, quantileResult.Metric.Labels, map[string]string{
+		"quantile": "1",
+	})
+	assert.Nil(t, quantileResult.Metadata)
+	assert.Len(t, quantileResult.Points, 1)
+	assert.Equal(t, quantileResult.Points[0].Interval, &monitoringpb.TimeInterval{
+		EndTime: timestamppb.New(end),
+	})
+	assert.Equal(t, quantileResult.Points[0].Value.GetDoubleValue(), 1.0)
+
+	sumResult = ts[3]
+	countResult = ts[4]
+	quantileResult = ts[5]
+
+	// Test sum mapping
+	assert.Equal(t, sumResult.MetricKind, metricpb.MetricDescriptor_CUMULATIVE)
+	assert.Equal(t, sumResult.ValueType, metricpb.MetricDescriptor_DOUBLE)
+	assert.Equal(t, sumResult.Unit, "1")
+	assert.Same(t, sumResult.Resource, mr)
+	assert.Equal(t, sumResult.Metric.Type, "workload.googleapis.com/mysummary_sum")
+	assert.Equal(t, sumResult.Metric.Labels, map[string]string{})
+	assert.Nil(t, sumResult.Metadata)
+	assert.Len(t, sumResult.Points, 1)
+	assert.Equal(t, sumResult.Points[0].Interval, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end2),
+	})
+	assert.Equal(t, sumResult.Points[0].Value.GetDoubleValue(), 200.0)
+	// Test count mapping
+	assert.Equal(t, countResult.MetricKind, metricpb.MetricDescriptor_CUMULATIVE)
+	assert.Equal(t, countResult.ValueType, metricpb.MetricDescriptor_DOUBLE)
+	assert.Equal(t, countResult.Unit, "1")
+	assert.Same(t, countResult.Resource, mr)
+	assert.Equal(t, countResult.Metric.Type, "workload.googleapis.com/mysummary_count")
+	assert.Equal(t, countResult.Metric.Labels, map[string]string{})
+	assert.Nil(t, countResult.Metadata)
+	assert.Len(t, countResult.Points, 1)
+	assert.Equal(t, countResult.Points[0].Interval, &monitoringpb.TimeInterval{
+		StartTime: timestamppb.New(start),
+		EndTime:   timestamppb.New(end2),
+	})
+	assert.Equal(t, countResult.Points[0].Value.GetDoubleValue(), float64(20))
+	// Test quantile mapping
+	assert.Equal(t, quantileResult.MetricKind, metricpb.MetricDescriptor_GAUGE)
+	assert.Equal(t, quantileResult.ValueType, metricpb.MetricDescriptor_DOUBLE)
+	assert.Equal(t, quantileResult.Unit, "1")
+	assert.Same(t, quantileResult.Resource, mr)
+	assert.Equal(t, quantileResult.Metric.Type, "workload.googleapis.com/mysummary")
+	assert.Equal(t, quantileResult.Metric.Labels, map[string]string{
+		"quantile": "1",
+	})
+	assert.Nil(t, quantileResult.Metadata)
+	assert.Len(t, quantileResult.Points, 1)
+	assert.Equal(t, quantileResult.Points[0].Interval, &monitoringpb.TimeInterval{
+		EndTime: timestamppb.New(end2),
+	})
+	assert.Equal(t, quantileResult.Points[0].Value.GetDoubleValue(), 1.0)
+}
+
 func TestMetricNameToType(t *testing.T) {
 	mapper, shutdown := newTestMetricMapper()
 	defer shutdown()
diff --git a/exporter/trace/trace_proto_test.go b/exporter/trace/trace_proto_test.go
index a624f26..61e86de 100644
--- a/exporter/trace/trace_proto_test.go
+++ b/exporter/trace/trace_proto_test.go
@@ -17,10 +17,15 @@ package trace
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 	"go.opentelemetry.io/otel/trace"
 	tracepb "google.golang.org/genproto/googleapis/devtools/cloudtrace/v2"
 )
@@ -40,6 +45,69 @@ func genSpanContext() trace.SpanContext {
 	return span.SpanContext()
 }
 
+func TestTraceProto_attributesFromSpans(t *testing.T) {
+	t.Run("Should include instrumentation library", func(t *testing.T) {
+		e := testExporter()
+		startTime := time.Unix(1585674086, 1234)
+		endTime := startTime.Add(10 * time.Second)
+		traceState, _ := trace.ParseTraceState("key1=val1,key2=val2")
+		rawSpan := tracetest.SpanStub{
+			SpanContext: trace.NewSpanContext(trace.SpanContextConfig{
+				TraceID:    trace.TraceID{0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F},
+				SpanID:     trace.SpanID{0xFF, 0xFE, 0xFD, 0xFC, 0xFB, 0xFA, 0xF9, 0xF8},
+				TraceState: traceState,
+			}),
+			SpanKind:  trace.SpanKindServer,
+			Name:      "span data to span data",
+			StartTime: startTime,
+			EndTime:   endTime,
+			Status: sdktrace.Status{
+				Code:        codes.Error,
+				Description: "utterly unrecognized",
+			},
+			Attributes: []attribute.KeyValue{
+				attribute.Int64("timeout_ns", 12e9),
+				attribute.Bool("conflict", true),
+			},
+			Resource: resource.NewWithAttributes(
+				"http://example.com/custom-resource-schema",
+				attribute.String("rk1", "rv1"),
+				attribute.Int64("rk2", 5),
+				attribute.StringSlice("rk3", []string{"sv1", "sv2"}),
+				attribute.Bool("conflict", false),
+			),
+			InstrumentationLibrary: instrumentation.Library{
+				Name:    "lib-name",
+				Version: "v0.0.1",
+			},
+		}
+		span := e.ConvertSpan(context.Background(), rawSpan.Snapshot())
+		assert.NotNil(t, span)
+
+		// Ensure agent key is created.
+		assert.Contains(t, span.Attributes.AttributeMap, "g.co/agent")
+
+		// Ensure resource keys are copied.
+		assert.Contains(t, span.Attributes.AttributeMap, "rk1")
+		assert.Contains(t, span.Attributes.AttributeMap, "rk2")
+		// TODO - Support JSON rendering of list-attributes.
+		// assert.Contains(t, span.Attributes.AttributeMap, "rk3")
+
+		// Ensure instrumentation library values are copied.
+		assert.Contains(t, span.Attributes.AttributeMap, "otel.scope.name")
+		assert.Equal(t, span.Attributes.AttributeMap["otel.scope.name"].GetStringValue().Value, "lib-name")
+		assert.Contains(t, span.Attributes.AttributeMap, "otel.scope.version")
+
+		// Ensure monitored resource labels are created.
+		assert.Contains(t, span.Attributes.AttributeMap, "g.co/r/generic_node/location")
+		assert.Equal(t, span.Attributes.AttributeMap["g.co/r/generic_node/location"].GetStringValue().Value, "global")
+
+		// Ensure span attribute "wins" over resource attribute.
+		assert.Contains(t, span.Attributes.AttributeMap, "conflict")
+		assert.Equal(t, span.Attributes.AttributeMap["conflict"].GetBoolValue(), true)
+	})
+}
+
 func TestTraceProto_linksProtoFromLinks(t *testing.T) {
 	t.Run("Should be nil when no links", func(t *testing.T) {
 		e := testExporter()