add metrics client for trace agent

exporter/datadogexporter/traces_exporter.go

@@ -13,6 +13,7 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/trace/agent"
 	traceconfig "github.com/DataDog/datadog-agent/pkg/trace/config"
 	tracelog "github.com/DataDog/datadog-agent/pkg/trace/log"
+	tracemetrics "github.com/DataDog/datadog-agent/pkg/trace/metrics"
 	"github.com/DataDog/datadog-agent/pkg/trace/telemetry"
 	"github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
 	"github.com/DataDog/opentelemetry-mapping-go/pkg/inframetadata"
@@ -21,13 +22,15 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+
 	"go.uber.org/zap"
 	zorkian "gopkg.in/zorkian/go-datadog-api.v2"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/clientutil"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/hostmetadata"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/metrics"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/datadogexporter/internal/scrub"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/datadog"
 )
 
 type traceExporter struct {
@@ -42,6 +45,7 @@ type traceExporter struct {
 	sourceProvider   source.Provider         // is able to source the origin of a trace (hostname, container, etc)
 	metadataReporter *inframetadata.Reporter // reports host metadata from resource attributes and metrics
 	retrier          *clientutil.Retrier     // retrier handles retries on requests
+	metricsClient    tracemetrics.StatsClient
 }
 
 func newTracesExporter(
@@ -64,6 +68,7 @@ func newTracesExporter(
 		sourceProvider:   sourceProvider,
 		retrier:          clientutil.NewRetrier(params.Logger, cfg.BackOffConfig, scrubber),
 		metadataReporter: metadataReporter,
+		metricsClient:    datadog.NewMetricClient(params.MeterProvider.Meter("datadogexporter")),
 	}
 	// client to send running metric to the backend & perform API key validation
 	errchan := make(chan error)
@@ -210,5 +215,6 @@ func newTraceAgent(ctx context.Context, params exporter.CreateSettings, cfg *Con
 		acfg.Endpoints[0].Host = addr
 	}
 	tracelog.SetLogger(&zaplogger{params.Logger})
+
 	return agent.NewAgent(ctx, acfg, telemetry.NewNoopCollector()), nil
 }

internal/datadog/metrics_client.go

@@ -0,0 +1,97 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package datadog // import "github.com/open-telemetry/opentelemetry-collector-contrib/internal/datadog"
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/DataDog/datadog-agent/pkg/trace/metrics"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+type metricsClient struct {
+	meter  metric.Meter
+	guages map[string]float64
+	mutex  sync.Mutex
+}
+
+// NewMetricClient returns a new metrics client
+func NewMetricClient(meter metric.Meter) metrics.StatsClient {
+	// We need to lock the mutex here to avoid
+	// duplicate metrics being created by both datadogconnector and exporter
+	var mutex sync.Mutex
+	mutex.Lock()
+	defer mutex.Unlock()
+	m := &metricsClient{
+		meter:  meter,
+		guages: make(map[string]float64),
+	}
+	metrics.Client = m
+	return m
+}
+
+func (m *metricsClient) Gauge(name string, value float64, tags []string, _ float64) error {
+	m.mutex.Lock()
+	if _, ok := m.guages[name]; ok {
+		m.guages[name] = value
+		return nil
+	}
+	m.guages[name] = value
+	m.mutex.Unlock()
+	_, err := m.meter.Float64ObservableGauge(name, metric.WithFloat64Callback(func(_ context.Context, f metric.Float64Observer) error {
+		attr := attributeFromTags(tags)
+		if v, ok := m.guages[name]; ok {
+			f.Observe(v, metric.WithAttributes(attr...))
+		}
+		return nil
+	}))
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *metricsClient) Count(name string, value int64, tags []string, _ float64) error {
+	counter, err := m.meter.Int64Counter(name)
+	if err != nil {
+		return err
+	}
+	attr := attributeFromTags(tags)
+	counter.Add(context.Background(), value, metric.WithAttributes(attr...))
+	return nil
+}
+
+func attributeFromTags(tags []string) []attribute.KeyValue {
+	attr := make([]attribute.KeyValue, 0, len(tags))
+	for _, t := range tags {
+		kv := strings.Split(t, ":")
+		attr = append(attr, attribute.KeyValue{
+			Key:   attribute.Key(kv[0]),
+			Value: attribute.StringValue(kv[1]),
+		})
+	}
+	return attr
+}
+
+func (m *metricsClient) Histogram(name string, value float64, tags []string, _ float64) error {
+	hist, err := m.meter.Float64Histogram(name)
+	if err != nil {
+		return err
+	}
+	attr := attributeFromTags(tags)
+	hist.Record(context.Background(), value, metric.WithAttributes(attr...))
+	return nil
+}
+
+func (m *metricsClient) Timing(name string, value time.Duration, tags []string, rate float64) error {
+	return m.Histogram(name, float64(value.Milliseconds()), tags, rate)
+}
+
+func (m *metricsClient) Flush() error {
+	return nil
+}