tanzuobservabilityexporter: use uber/atomic instead of sync/atomic

exporter/tanzuobservabilityexporter/go.mod

@@ -11,6 +11,7 @@ require (
 	go.opentelemetry.io/collector v0.50.1-0.20220429151328-041f39835df7
 	go.opentelemetry.io/collector/pdata v0.50.1-0.20220429151328-041f39835df7
 	go.opentelemetry.io/collector/semconv v0.50.1-0.20220429151328-041f39835df7
+	go.uber.org/atomic v1.9.0
 	go.uber.org/multierr v1.8.0
 	go.uber.org/zap v1.21.0
 )
@@ -38,7 +39,6 @@ require (
 	go.opentelemetry.io/otel v1.7.0 // indirect
 	go.opentelemetry.io/otel/metric v0.30.0 // indirect
 	go.opentelemetry.io/otel/trace v1.7.0 // indirect
-	go.uber.org/atomic v1.9.0 // indirect
 	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
 	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
 	golang.org/x/text v0.3.7 // indirect

exporter/tanzuobservabilityexporter/metrics.go

@@ -20,13 +20,13 @@ import (
 	"fmt"
 	"math"
 	"strconv"
-	"sync/atomic"
 
 	"github.com/wavefronthq/wavefront-sdk-go/histogram"
 	"github.com/wavefronthq/wavefront-sdk-go/senders"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/atomic"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 )
@@ -180,37 +180,20 @@ type flushCloser interface {
 	Close()
 }
 
-// counter represents an internal counter metric. The zero value is ready to use
-type counter struct {
-	count int64
-}
-
-// Report reports this counter to tanzu observability. name is the name of
+// report the counter to tanzu observability. name is the name of
 // the metric to be reported. tags is the tags for the metric. sender is what
 // sends the metric to tanzu observability. Any errors get added to errs.
-func (c *counter) Report(
-	name string, tags map[string]string, sender gaugeSender, errs *[]error,
-) {
-	err := sender.SendMetric(name, float64(c.Get()), 0, "", tags)
+func report(count *atomic.Int64, name string, tags map[string]string, sender gaugeSender, errs *[]error) {
+	err := sender.SendMetric(name, float64(count.Load()), 0, "", tags)
 	if err != nil {
 		*errs = append(*errs, err)
 	}
 }
 
-// Inc increments this counter by one.
-func (c *counter) Inc() {
-	atomic.AddInt64(&c.count, 1)
-}
-
-// Get gets the value of this counter.
-func (c *counter) Get() int64 {
-	return atomic.LoadInt64(&c.count)
-}
-
 // logMissingValue keeps track of metrics with missing values. metric is the
 // metric with the missing value. settings logs the missing value. count counts
 // metrics with missing values.
-func logMissingValue(metric pmetric.Metric, settings component.TelemetrySettings, count *counter) {
+func logMissingValue(metric pmetric.Metric, settings component.TelemetrySettings, count *atomic.Int64) {
 	namef := zap.String(metricNameString, metric.Name())
 	typef := zap.String(metricTypeString, metric.DataType().String())
 	settings.Logger.Debug("Metric missing value", namef, typef)
@@ -240,7 +223,7 @@ func pushGaugeNumberDataPoint(
 	errs *[]error,
 	sender gaugeSender,
 	settings component.TelemetrySettings,
-	missingValues *counter,
+	missingValues *atomic.Int64,
 ) {
 	tags := attributesToTagsForMetrics(numberDataPoint.Attributes(), mi.SourceKey)
 	ts := numberDataPoint.Timestamp().AsTime().Unix()
@@ -263,16 +246,17 @@ type gaugeSender interface {
 type gaugeConsumer struct {
 	sender        gaugeSender
 	settings      component.TelemetrySettings
-	missingValues counter
+	missingValues *atomic.Int64
 }
 
 // newGaugeConsumer returns a typedMetricConsumer that consumes gauge metrics
 // by sending them to tanzu observability.
 func newGaugeConsumer(
 	sender gaugeSender, settings component.TelemetrySettings) typedMetricConsumer {
 	return &gaugeConsumer{
-		sender:   sender,
-		settings: settings,
+		sender:        sender,
+		settings:      settings,
+		missingValues: atomic.NewInt64(0),
 	}
 }
 
@@ -290,27 +274,28 @@ func (g *gaugeConsumer) Consume(mi metricInfo, errs *[]error) {
 			errs,
 			g.sender,
 			g.settings,
-			&g.missingValues)
+			g.missingValues)
 	}
 }
 
 func (g *gaugeConsumer) PushInternalMetrics(errs *[]error) {
-	g.missingValues.Report(missingValueMetricName, typeIsGaugeTags, g.sender, errs)
+	report(g.missingValues, missingValueMetricName, typeIsGaugeTags, g.sender, errs)
 }
 
 type sumConsumer struct {
 	sender        senders.MetricSender
 	settings      component.TelemetrySettings
-	missingValues counter
+	missingValues *atomic.Int64
 }
 
 // newSumConsumer returns a typedMetricConsumer that consumes sum metrics
 // by sending them to tanzu observability.
 func newSumConsumer(
 	sender senders.MetricSender, settings component.TelemetrySettings) typedMetricConsumer {
 	return &sumConsumer{
-		sender:   sender,
-		settings: settings,
+		sender:        sender,
+		settings:      settings,
+		missingValues: atomic.NewInt64(0),
 	}
 }
 
@@ -330,20 +315,20 @@ func (s *sumConsumer) Consume(mi metricInfo, errs *[]error) {
 			s.pushNumberDataPoint(mi, numberDataPoints.At(i), errs)
 		} else {
 			pushGaugeNumberDataPoint(
-				mi, numberDataPoints.At(i), errs, s.sender, s.settings, &s.missingValues)
+				mi, numberDataPoints.At(i), errs, s.sender, s.settings, s.missingValues)
 		}
 	}
 }
 
 func (s *sumConsumer) PushInternalMetrics(errs *[]error) {
-	s.missingValues.Report(missingValueMetricName, typeIsSumTags, s.sender, errs)
+	report(s.missingValues, missingValueMetricName, typeIsSumTags, s.sender, errs)
 }
 
 func (s *sumConsumer) pushNumberDataPoint(mi metricInfo, numberDataPoint pmetric.NumberDataPoint, errs *[]error) {
 	tags := attributesToTagsForMetrics(numberDataPoint.Attributes(), mi.SourceKey)
 	value, err := getValue(numberDataPoint)
 	if err != nil {
-		logMissingValue(mi.Metric, s.settings, &s.missingValues)
+		logMissingValue(mi.Metric, s.settings, s.missingValues)
 		return
 	}
 	err = s.sender.SendDeltaCounter(mi.Name(), value, mi.Source, tags)
@@ -355,24 +340,28 @@ func (s *sumConsumer) pushNumberDataPoint(mi metricInfo, numberDataPoint pmetric
 // histogramReporting takes care of logging and internal metrics for histograms
 type histogramReporting struct {
 	settings                 component.TelemetrySettings
-	malformedHistograms      counter
-	noAggregationTemporality counter
+	malformedHistograms      *atomic.Int64
+	noAggregationTemporality *atomic.Int64
 }
 
 // newHistogramReporting returns a new histogramReporting instance.
 func newHistogramReporting(settings component.TelemetrySettings) *histogramReporting {
-	return &histogramReporting{settings: settings}
+	return &histogramReporting{
+		settings:                 settings,
+		malformedHistograms:      atomic.NewInt64(0),
+		noAggregationTemporality: atomic.NewInt64(0),
+	}
 }
 
 // Malformed returns the number of malformed histogram data points.
 func (r *histogramReporting) Malformed() int64 {
-	return r.malformedHistograms.Get()
+	return r.malformedHistograms.Load()
 }
 
 // NoAggregationTemporality returns the number of histogram metrics that have no
 // aggregation temporality.
 func (r *histogramReporting) NoAggregationTemporality() int64 {
-	return r.noAggregationTemporality.Get()
+	return r.noAggregationTemporality.Load()
 }
 
 // LogMalformed logs seeing one malformed data point.
@@ -392,12 +381,8 @@ func (r *histogramReporting) LogNoAggregationTemporality(metric pmetric.Metric)
 // Report sends the counts in this instance to wavefront.
 // sender is what sends to wavefront. Any errors sending get added to errs.
 func (r *histogramReporting) Report(sender gaugeSender, errs *[]error) {
-	r.malformedHistograms.Report(malformedHistogramMetricName, nil, sender, errs)
-	r.noAggregationTemporality.Report(
-		noAggregationTemporalityMetricName,
-		typeIsHistogramTags,
-		sender,
-		errs)
+	report(r.malformedHistograms, malformedHistogramMetricName, nil, sender, errs)
+	report(r.noAggregationTemporality, noAggregationTemporalityMetricName, typeIsHistogramTags, sender, errs)
 }
 
 type histogramConsumer struct {