[exporter/datadog] Add new metrics helper functions based on native APIs

exporter/datadogexporter/internal/metrics/consumer.go

@@ -0,0 +1,156 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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
+//
+//     http://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 metrics // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/metrics"
+
+import (
+	"context"
+
+	"github.com/DataDog/datadog-agent/pkg/otlp/model/translator"
+	"github.com/DataDog/datadog-agent/pkg/quantile"
+	"github.com/DataDog/datadog-agent/pkg/trace/pb"
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadog"
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
+	"go.opentelemetry.io/collector/component"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/metrics/sketches"
+)
+
+var _ translator.Consumer = (*Consumer)(nil)
+var _ translator.HostConsumer = (*Consumer)(nil)
+var _ translator.TagsConsumer = (*Consumer)(nil)
+var _ translator.APMStatsConsumer = (*Consumer)(nil)
+
+// Consumer implements translator.Consumer. It records consumed metrics, sketches and
+// APM stats payloads. It provides them to the caller using the All method.
+type Consumer struct {
+	ms        []datadogV2.MetricSeries
+	sl        sketches.SketchSeriesList
+	as        []pb.ClientStatsPayload
+	seenHosts map[string]struct{}
+	seenTags  map[string]struct{}
+}
+
+// NewConsumer creates a new Datadog consumer. It implements translator.Consumer.
+func NewConsumer() *Consumer {
+	return &Consumer{
+		seenHosts: make(map[string]struct{}),
+		seenTags:  make(map[string]struct{}),
+	}
+}
+
+// toDataType maps translator datatypes to DatadogV2's datatypes.
+func (c *Consumer) toDataType(dt translator.MetricDataType) (out datadogV2.MetricIntakeType) {
+	out = datadogV2.METRICINTAKETYPE_UNSPECIFIED
+
+	switch dt {
+	case translator.Count:
+		out = datadogV2.METRICINTAKETYPE_COUNT
+	case translator.Gauge:
+		out = datadogV2.METRICINTAKETYPE_GAUGE
+	}
+
+	return
+}
+
+// runningMetrics gets the running metrics for the exporter.
+func (c *Consumer) runningMetrics(timestamp uint64, buildInfo component.BuildInfo) (series []datadogV2.MetricSeries) {
+	for host := range c.seenHosts {
+		// Report the host as running
+		runningMetric := DefaultMetrics("metrics", host, timestamp, buildInfo)
+		series = append(series, runningMetric...)
+	}
+
+	for tag := range c.seenTags {
+		runningMetrics := DefaultMetrics("metrics", "", timestamp, buildInfo)
+		for i := range runningMetrics {
+			runningMetrics[i].Tags = append(runningMetrics[i].Tags, tag)
+		}
+		series = append(series, runningMetrics...)
+	}
+
+	return
+}
+
+// All gets all metrics (consumed metrics and running metrics).
+func (c *Consumer) All(timestamp uint64, buildInfo component.BuildInfo, tags []string) ([]datadogV2.MetricSeries, sketches.SketchSeriesList, []pb.ClientStatsPayload) {
+	series := c.ms
+	series = append(series, c.runningMetrics(timestamp, buildInfo)...)
+	if len(tags) == 0 {
+		return series, c.sl, c.as
+	}
+	for i := range series {
+		series[i].Tags = append(series[i].Tags, tags...)
+	}
+	for i := range c.sl {
+		c.sl[i].Tags = append(c.sl[i].Tags, tags...)
+	}
+	for i := range c.as {
+		c.as[i].Tags = append(c.as[i].Tags, tags...)
+	}
+	return series, c.sl, c.as
+}
+
+// ConsumeAPMStats implements translator.APMStatsConsumer.
+func (c *Consumer) ConsumeAPMStats(s pb.ClientStatsPayload) {
+	c.as = append(c.as, s)
+}
+
+// ConsumeTimeSeries implements the translator.Consumer interface.
+func (c *Consumer) ConsumeTimeSeries(
+	_ context.Context,
+	dims *translator.Dimensions,
+	typ translator.MetricDataType,
+	timestamp uint64,
+	value float64,
+) {
+	dt := c.toDataType(typ)
+	met := NewMetric(dims.Name(), dt, timestamp, value, dims.Tags())
+	met.SetResources([]datadogV2.MetricResource{
+		{
+			Name: datadog.PtrString(dims.Host()),
+			Type: datadog.PtrString("host"),
+		},
+	})
+	c.ms = append(c.ms, met)
+}
+
+// ConsumeSketch implements the translator.Consumer interface.
+func (c *Consumer) ConsumeSketch(
+	_ context.Context,
+	dims *translator.Dimensions,
+	timestamp uint64,
+	sketch *quantile.Sketch,
+) {
+	c.sl = append(c.sl, sketches.SketchSeries{
+		Name:     dims.Name(),
+		Tags:     dims.Tags(),
+		Host:     dims.Host(),
+		Interval: 1,
+		Points: []sketches.SketchPoint{{
+			Ts:     int64(timestamp / 1e9),
+			Sketch: sketch,
+		}},
+	})
+}
+
+// ConsumeHost implements the translator.HostConsumer interface.
+func (c *Consumer) ConsumeHost(host string) {
+	c.seenHosts[host] = struct{}{}
+}
+
+// ConsumeTag implements the translator.TagsConsumer interface.
+func (c *Consumer) ConsumeTag(tag string) {
+	c.seenTags[tag] = struct{}{}
+}

exporter/datadogexporter/internal/metrics/consumer_deprecated_test.go

@@ -19,8 +19,6 @@ import (
 	"testing"
 
 	"github.com/DataDog/datadog-agent/pkg/otlp/model/attributes"
-	"github.com/DataDog/datadog-agent/pkg/otlp/model/source"
-	"github.com/DataDog/datadog-agent/pkg/otlp/model/translator"
 	"github.com/DataDog/datadog-agent/pkg/trace/pb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -32,23 +30,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/testutil"
 )
 
-type testProvider string
-
-func (t testProvider) Source(context.Context) (source.Source, error) {
-	return source.Source{Kind: source.HostnameKind, Identifier: string(t)}, nil
-}
-
-func newTranslator(t *testing.T, logger *zap.Logger) *translator.Translator {
-	tr, err := translator.New(logger,
-		translator.WithHistogramMode(translator.HistogramModeDistributions),
-		translator.WithNumberMode(translator.NumberModeCumulativeToDelta),
-		translator.WithFallbackSourceProvider(testProvider("fallbackHostname")),
-	)
-	require.NoError(t, err)
-	return tr
-}
-
-func TestRunningMetrics(t *testing.T) {
+func TestZorkianRunningMetrics(t *testing.T) {
 	ms := pmetric.NewMetrics()
 	rms := ms.ResourceMetrics()
 

exporter/datadogexporter/internal/metrics/consumer_test.go

@@ -0,0 +1,150 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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
+//
+//     http://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 metrics
+
+import (
+	"context"
+	"testing"
+
+	"github.com/DataDog/datadog-agent/pkg/otlp/model/attributes"
+	"github.com/DataDog/datadog-agent/pkg/otlp/model/source"
+	"github.com/DataDog/datadog-agent/pkg/otlp/model/translator"
+	"github.com/DataDog/datadog-agent/pkg/trace/pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/testutil"
+)
+
+type testProvider string
+
+func (t testProvider) Source(context.Context) (source.Source, error) {
+	return source.Source{Kind: source.HostnameKind, Identifier: string(t)}, nil
+}
+
+func newTranslator(t *testing.T, logger *zap.Logger) *translator.Translator {
+	tr, err := translator.New(logger,
+		translator.WithHistogramMode(translator.HistogramModeDistributions),
+		translator.WithNumberMode(translator.NumberModeCumulativeToDelta),
+		translator.WithFallbackSourceProvider(testProvider("fallbackHostname")),
+	)
+	require.NoError(t, err)
+	return tr
+}
+
+func TestRunningMetrics(t *testing.T) {
+	ms := pmetric.NewMetrics()
+	rms := ms.ResourceMetrics()
+
+	rm := rms.AppendEmpty()
+	resAttrs := rm.Resource().Attributes()
+	resAttrs.PutStr(attributes.AttributeDatadogHostname, "resource-hostname-1")
+
+	rm = rms.AppendEmpty()
+	resAttrs = rm.Resource().Attributes()
+	resAttrs.PutStr(attributes.AttributeDatadogHostname, "resource-hostname-1")
+
+	rm = rms.AppendEmpty()
+	resAttrs = rm.Resource().Attributes()
+	resAttrs.PutStr(attributes.AttributeDatadogHostname, "resource-hostname-2")
+
+	rms.AppendEmpty()
+
+	logger, _ := zap.NewProduction()
+	tr := newTranslator(t, logger)
+
+	ctx := context.Background()
+	consumer := NewConsumer()
+	assert.NoError(t, tr.MapMetrics(ctx, ms, consumer))
+
+	var runningHostnames []string
+	for _, metric := range consumer.runningMetrics(0, component.BuildInfo{}) {
+		for _, res := range metric.Resources {
+			runningHostnames = append(runningHostnames, *res.Name)
+		}
+	}
+
+	assert.ElementsMatch(t,
+		runningHostnames,
+		[]string{"fallbackHostname", "resource-hostname-1", "resource-hostname-2"},
+	)
+}
+
+func TestTagsMetrics(t *testing.T) {
+	ms := pmetric.NewMetrics()
+	rms := ms.ResourceMetrics()
+
+	rm := rms.AppendEmpty()
+	baseAttrs := testutil.NewAttributeMap(map[string]string{
+		conventions.AttributeCloudProvider:      conventions.AttributeCloudProviderAWS,
+		conventions.AttributeCloudPlatform:      conventions.AttributeCloudPlatformAWSECS,
+		conventions.AttributeAWSECSTaskFamily:   "example-task-family",
+		conventions.AttributeAWSECSTaskRevision: "example-task-revision",
+		conventions.AttributeAWSECSLaunchtype:   conventions.AttributeAWSECSLaunchtypeFargate,
+	})
+	baseAttrs.CopyTo(rm.Resource().Attributes())
+	rm.Resource().Attributes().PutStr(conventions.AttributeAWSECSTaskARN, "task-arn-1")
+
+	rm = rms.AppendEmpty()
+	baseAttrs.CopyTo(rm.Resource().Attributes())
+	rm.Resource().Attributes().PutStr(conventions.AttributeAWSECSTaskARN, "task-arn-2")
+
+	rm = rms.AppendEmpty()
+	baseAttrs.CopyTo(rm.Resource().Attributes())
+	rm.Resource().Attributes().PutStr(conventions.AttributeAWSECSTaskARN, "task-arn-3")
+
+	logger, _ := zap.NewProduction()
+	tr := newTranslator(t, logger)
+
+	ctx := context.Background()
+	consumer := NewConsumer()
+	assert.NoError(t, tr.MapMetrics(ctx, ms, consumer))
+
+	runningMetrics := consumer.runningMetrics(0, component.BuildInfo{})
+	var runningTags []string
+	var runningHostnames []string
+	for _, metric := range runningMetrics {
+		runningTags = append(runningTags, metric.Tags...)
+		for _, res := range metric.Resources {
+			runningHostnames = append(runningHostnames, *res.Name)
+		}
+	}
+
+	assert.ElementsMatch(t, runningHostnames, []string{"", "", ""})
+	assert.Len(t, runningMetrics, 3)
+	assert.ElementsMatch(t, runningTags, []string{"task_arn:task-arn-1", "task_arn:task-arn-2", "task_arn:task-arn-3"})
+}
+
+func TestConsumeAPMStats(t *testing.T) {
+	c := NewConsumer()
+	for _, sp := range testutil.StatsPayloads {
+		c.ConsumeAPMStats(sp)
+	}
+	require.Len(t, c.as, len(testutil.StatsPayloads))
+	require.ElementsMatch(t, c.as, testutil.StatsPayloads)
+	_, _, out := c.All(0, component.BuildInfo{}, []string{})
+	require.ElementsMatch(t, out, testutil.StatsPayloads)
+	_, _, out = c.All(0, component.BuildInfo{}, []string{"extra:key"})
+	var copies []pb.ClientStatsPayload
+	for _, sp := range testutil.StatsPayloads {
+		sp.Tags = append(sp.Tags, "extra:key")
+		copies = append(copies, sp)
+	}
+	require.ElementsMatch(t, out, copies)
+}