[exporter/tanzuobservability] Don't roll resource tags into metrics

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### 💡 Enhancements 💡
 
+- `tanzuobservabilityexporter`: Use resourcetotelemetry helper (#8338)
 - `cmd/mdatagen`: Add resource attributes definition to metadata.yaml and move `pdata.Metrics` creation to the
   generated code (#5270) 
 - Add `make crosslink` target to ensure replace statements are included in `go.mod` for all transitive dependencies within repository (#8822)

exporter/tanzuobservabilityexporter/config.go

@@ -22,6 +22,8 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry"
 )
 
 type TracesConfig struct {
@@ -30,6 +32,7 @@ type TracesConfig struct {
 
 type MetricsConfig struct {
 	confighttp.HTTPClientSettings `mapstructure:",squash"`
+	ResourceAttributes            resourcetotelemetry.Settings `mapstructure:"resource_attributes"`
 }
 
 // Config defines configuration options for the exporter.

exporter/tanzuobservabilityexporter/factory.go

@@ -16,11 +16,14 @@ package tanzuobservabilityexporter // import "github.com/open-telemetry/opentele
 
 import (
 	"context"
+	"fmt"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry"
 )
 
 const exporterType = "tanzuobservability"
@@ -87,12 +90,23 @@ func createMetricsExporter(
 
 	tobsCfg := cfg.(*Config)
 
-	return exporterhelper.NewMetricsExporter(
+	exporter, err := exporterhelper.NewMetricsExporter(
 		cfg,
 		set,
 		exp.pushMetricsData,
 		exporterhelper.WithQueue(tobsCfg.QueueSettings),
 		exporterhelper.WithRetry(tobsCfg.RetrySettings),
 		exporterhelper.WithShutdown(exp.shutdown),
 	)
+	if err != nil {
+		return nil, err
+	}
+	ourConfig, ok := cfg.(*Config)
+	if !ok {
+		return nil, fmt.Errorf("invalid config: %#v", cfg)
+	}
+	return resourcetotelemetry.WrapMetricsExporter(
+		ourConfig.Metrics.ResourceAttributes,
+		exporter,
+	), nil
 }

exporter/tanzuobservabilityexporter/factory_test.go

@@ -28,6 +28,8 @@ import (
 	"go.opentelemetry.io/collector/config/configtest"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/service/servicetest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry"
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
@@ -61,6 +63,7 @@ func TestLoadConfig(t *testing.T) {
 		},
 		Metrics: MetricsConfig{
 			HTTPClientSettings: confighttp.HTTPClientSettings{Endpoint: "http://localhost:2916"},
+			ResourceAttributes: resourcetotelemetry.Settings{Enabled: true},
 		},
 		QueueSettings: exporterhelper.QueueSettings{
 			Enabled:      true,

exporter/tanzuobservabilityexporter/go.mod

@@ -5,6 +5,7 @@ go 1.17
 require (
 	github.com/google/uuid v1.3.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.48.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry v0.48.0
 	github.com/stretchr/testify v1.7.1
 	github.com/wavefronthq/wavefront-sdk-go v0.9.10
 	go.opentelemetry.io/collector v0.48.0

exporter/tanzuobservabilityexporter/metrics.go

@@ -65,8 +65,8 @@ type metricsConsumer struct {
 
 type metricInfo struct {
 	pdata.Metric
-	Source string
-	Tags   map[string]string
+	Source    string
+	SourceKey string
 }
 
 // newMetricsConsumer returns a new metricsConsumer. consumers are the
@@ -103,13 +103,13 @@ func (c *metricsConsumer) Consume(ctx context.Context, md pdata.Metrics) error {
 	rms := md.ResourceMetrics()
 	for i := 0; i < rms.Len(); i++ {
 		rm := rms.At(i).Resource().Attributes()
-		source, resourceTags := getSourceAndResourceTags(rm)
+		source, sourceKey := getSourceAndKey(rm)
 		ilms := rms.At(i).ScopeMetrics()
 		for j := 0; j < ilms.Len(); j++ {
 			ms := ilms.At(j).Metrics()
 			for k := 0; k < ms.Len(); k++ {
 				m := ms.At(k)
-				mi := metricInfo{m, source, resourceTags}
+				mi := metricInfo{Metric: m, Source: source, SourceKey: sourceKey}
 				select {
 				case <-ctx.Done():
 					return multierr.Combine(append(errs, errors.New("context canceled"))...)
@@ -241,7 +241,7 @@ func pushGaugeNumberDataPoint(
 	settings component.TelemetrySettings,
 	missingValues *counter,
 ) {
-	tags := attributesToTags(mi.Tags, numberDataPoint.Attributes())
+	tags := attributesToTagsForMetrics(numberDataPoint.Attributes(), mi.SourceKey)
 	ts := numberDataPoint.Timestamp().AsTime().Unix()
 	value, err := getValue(numberDataPoint)
 	if err != nil {
@@ -339,7 +339,7 @@ func (s *sumConsumer) PushInternalMetrics(errs *[]error) {
 }
 
 func (s *sumConsumer) pushNumberDataPoint(mi metricInfo, numberDataPoint pdata.NumberDataPoint, errs *[]error) {
-	tags := attributesToTags(mi.Tags, numberDataPoint.Attributes())
+	tags := attributesToTagsForMetrics(numberDataPoint.Attributes(), mi.SourceKey)
 	value, err := getValue(numberDataPoint)
 	if err != nil {
 		logMissingValue(mi.Metric, s.settings, &s.missingValues)
@@ -484,7 +484,7 @@ func (c *cumulativeHistogramDataPointConsumer) Consume(
 	reporting *histogramReporting,
 ) {
 	name := mi.Name()
-	tags := attributesToTags(mi.Tags, histogram.Attributes())
+	tags := attributesToTagsForMetrics(histogram.Attributes(), mi.SourceKey)
 	ts := histogram.Timestamp().AsTime().Unix()
 	explicitBounds := histogram.ExplicitBounds()
 	bucketCounts := histogram.BucketCounts()
@@ -530,7 +530,7 @@ func (d *deltaHistogramDataPointConsumer) Consume(
 	errs *[]error,
 	reporting *histogramReporting) {
 	name := mi.Name()
-	tags := attributesToTags(mi.Tags, his.Attributes())
+	tags := attributesToTagsForMetrics(his.Attributes(), mi.SourceKey)
 	ts := his.Timestamp().AsTime().Unix()
 	explicitBounds := his.ExplicitBounds()
 	bucketCounts := his.BucketCounts()
@@ -656,7 +656,7 @@ func (s *summaryConsumer) sendSummaryDataPoint(
 ) {
 	name := mi.Name()
 	ts := summaryDataPoint.Timestamp().AsTime().Unix()
-	tags := attributesToTags(mi.Tags, summaryDataPoint.Attributes())
+	tags := attributesToTagsForMetrics(summaryDataPoint.Attributes(), mi.SourceKey)
 	count := summaryDataPoint.Count()
 	sum := summaryDataPoint.Sum()
 
@@ -687,6 +687,13 @@ func (s *summaryConsumer) sendMetric(
 	}
 }
 
+func attributesToTagsForMetrics(attributes pdata.Map, sourceKey string) map[string]string {
+	tags := attributesToTags(attributes)
+	delete(tags, sourceKey)
+	replaceSource(tags)
+	return tags
+}
+
 func quantileTagValue(quantile float64) string {
 	return strconv.FormatFloat(quantile, 'f', -1, 64)
 }