Initial DogStatsD implementation (#15)

Initial metrics exporter through DogStatsD with support for all metric types but summary and distribution

Author: mx-psi

exporter/datadogexporter/config.go

@@ -24,6 +24,7 @@ import (
 
 var (
 	errUnsetAPIKey = errors.New("the Datadog API key is unset")
+	errConflict    = errors.New("'site' and 'api_key' must not be set with agent mode")
 )
 
 // Config defines configuration for the Datadog exporter.
@@ -38,25 +39,46 @@ type Config struct {
 	// The default value is "datadoghq.com".
 	Site string `mapstructure:"site"`
 
-	// MetricsURL is the host of the Datadog intake server to send metrics to.
-	// If not set, the value is obtained from the Site.
+	// MetricsURL is the host of the Datadog intake server or Dogstatsd server to send metrics to.
+	// If not set, the value is obtained from the Site and the sending method.
 	MetricsURL string `mapstructure:"metrics_url"`
+
+	// Tags is the list of default tags to add to every metric or trace
+	Tags []string `mapstructure:"tags"`
+
+	// Mode states the mode for sending metrics and traces.
+	// The possible values are "api" and "agent".
+	// The default value is "agent".
+	Mode string `mapstructure:"sending_method"`
 }
 
 // Sanitize tries to sanitize a given configuration
 func (c *Config) Sanitize() error {
+	if c.Mode == AgentMode {
 
-	// Check API key is set
-	if c.APIKey == "" {
-		return errUnsetAPIKey
-	}
+		if c.APIKey != "" || c.Site != DefaultSite {
+			return errConflict
+		}
+
+		if c.MetricsURL == "" {
+			c.MetricsURL = "127.0.0.1:8125"
+		}
+
+	} else if c.Mode == APIMode {
+
+		if c.APIKey == "" {
+			return errUnsetAPIKey
+		}
+
+		// Sanitize API key
+		c.APIKey = strings.TrimSpace(c.APIKey)
 
-	// Sanitize API key
-	c.APIKey = strings.TrimSpace(c.APIKey)
+		if c.MetricsURL == "" {
+			c.MetricsURL = fmt.Sprintf("https://api.%s", c.Site)
+		}
 
-	// Set Endpoint based on site if unset
-	if c.MetricsURL == "" {
-		c.MetricsURL = fmt.Sprintf("https://api.%s", c.Site)
+	} else {
+		return fmt.Errorf("Selected mode '%s' is invalid", c.Mode)
 	}
 
 	return nil

exporter/datadogexporter/config_test.go

@@ -46,9 +46,10 @@ func TestLoadConfig(t *testing.T) {
 			NameVal: "datadog",
 			TypeVal: "datadog",
 		},
-		APIKey:     "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
 		Site:       DefaultSite,
-		MetricsURL: "https://api.datadoghq.com",
+		MetricsURL: "127.0.0.1:8125",
+		Tags:       []string{"tool:opentelemetry", "version:0.1.0"},
+		Mode:       AgentMode,
 	})
 
 	e1 := cfg.Exporters["datadog/2"].(*Config)
@@ -61,10 +62,17 @@ func TestLoadConfig(t *testing.T) {
 				NameVal: "datadog/2",
 				TypeVal: "datadog",
 			},
-			APIKey:     "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+			APIKey:     "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
 			Site:       "datadoghq.eu",
 			MetricsURL: "https://api.datadoghq.eu",
+			Tags:       DefaultTags,
+			Mode:       APIMode,
 		})
+
+	e3 := cfg.Exporters["datadog/invalid"].(*Config)
+	err = e3.Sanitize()
+	require.Error(t, err)
+
 }
 
 // TestOverrideMetricsURL tests that the metrics URL is overridden
@@ -77,6 +85,7 @@ func TestOverrideMetricsURL(t *testing.T) {
 		APIKey:     "notnull",
 		Site:       DefaultSite,
 		MetricsURL: DebugEndpoint,
+		Mode:       APIMode,
 	}
 
 	err := cfg.Sanitize()

exporter/datadogexporter/dogstatsd.go

@@ -0,0 +1,87 @@
+// 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 datadogexporter
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/DataDog/datadog-go/statsd"
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.uber.org/zap"
+)
+
+type dogStatsDExporter struct {
+	logger *zap.Logger
+	cfg    *Config
+	client *statsd.Client
+}
+
+func newDogStatsDExporter(logger *zap.Logger, cfg *Config) (*dogStatsDExporter, error) {
+
+	client, err := statsd.New(
+		cfg.MetricsURL,
+		statsd.WithNamespace("opentelemetry."),
+		statsd.WithTags(cfg.Tags),
+	)
+
+	if err != nil {
+		return nil, fmt.Errorf("Failed to initialize DogStatsD client: %s", err)
+	}
+
+	return &dogStatsDExporter{logger, cfg, client}, nil
+}
+
+func (exp *dogStatsDExporter) PushMetricsData(_ context.Context, md pdata.Metrics) (int, error) {
+	metrics, droppedTimeSeries, err := MapMetrics(exp, md)
+
+	if err != nil {
+		return droppedTimeSeries, err
+	}
+
+	for name, data := range metrics {
+		for _, metric := range data {
+			switch metric.GetType() {
+			case Count:
+				err = exp.client.Count(name, metric.GetValue().(int64), metric.GetTags(), metric.GetRate())
+			case Gauge:
+				err = exp.client.Gauge(name, metric.GetValue().(float64), metric.GetTags(), metric.GetRate())
+			}
+
+			if err != nil {
+				return droppedTimeSeries, err
+			}
+		}
+	}
+
+	return droppedTimeSeries, nil
+}
+
+func (exp *dogStatsDExporter) GetLogger() *zap.Logger {
+	return exp.logger
+}
+
+func (exp *dogStatsDExporter) GetConfig() *Config {
+	return exp.cfg
+}
+
+func (exp *dogStatsDExporter) GetQueueSettings() exporterhelper.QueueSettings {
+	return exporterhelper.CreateDefaultQueueSettings()
+}
+
+func (exp *dogStatsDExporter) GetRetrySettings() exporterhelper.RetrySettings {
+	return exporterhelper.CreateDefaultRetrySettings()
+}

exporter/datadogexporter/example/config.yaml

@@ -25,6 +25,11 @@ exporters:
     #
     # metrics_url: https://api.datadoghq.com
 
+    ## @param tags - list of strings - optional - default: []
+    ## The list of default tags to add to every metric or trace
+    #
+    # tags: []
+
 service:
   pipelines:
     traces:

exporter/datadogexporter/factory.go

@@ -28,6 +28,16 @@ const (
 
 	// DefaultSite is the default site of the Datadog intake to send data to
 	DefaultSite = "datadoghq.com"
+
+	// List the different sending methods
+	AgentMode   = "agent"
+	APIMode     = "api"
+	DefaultMode = APIMode
+)
+
+var (
+	// DefaultTags is the default set of tags to add to every metric or trace
+	DefaultTags = []string{}
 )
 
 // NewFactory creates a Datadog exporter factory
@@ -44,6 +54,8 @@ func NewFactory() component.ExporterFactory {
 func createDefaultConfig() configmodels.Exporter {
 	return &Config{
 		Site: DefaultSite,
+		Tags: DefaultTags,
+		Mode: DefaultMode,
 	}
 }
 
@@ -61,8 +73,17 @@ func CreateMetricsExporter(
 		return nil, err
 	}
 
-	//Metrics are not yet supported
-	return nil, configerror.ErrDataTypeIsNotSupported
+	exp, err := newMetricsExporter(params.Logger, cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return exporterhelper.NewMetricsExporter(
+		cfg,
+		exp.PushMetricsData,
+		exporterhelper.WithQueue(exp.GetQueueSettings()),
+		exporterhelper.WithRetry(exp.GetRetrySettings()),
+	)
 }
 
 // CreateTracesExporter creates a traces exporter based on this config.

exporter/datadogexporter/factory_test.go

@@ -0,0 +1,116 @@
+// 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 datadogexporter
+
+import (
+	"context"
+	"path"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configcheck"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/config/configtest"
+	"go.uber.org/zap"
+)
+
+// Test that the factory creates the default configuration
+func TestCreateDefaultConfig(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+
+	assert.Equal(t, cfg, &Config{
+		Site: DefaultSite,
+		Tags: DefaultTags,
+		Mode: DefaultMode,
+	}, "failed to create default config")
+
+	assert.NoError(t, configcheck.ValidateConfig(cfg))
+}
+
+func TestCreateAgentMetricsExporter(t *testing.T) {
+	logger := zap.NewNop()
+
+	factories, err := componenttest.ExampleComponents()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[configmodels.Type(typeStr)] = factory
+	cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	ctx := context.Background()
+	exp, err := factory.CreateMetricsExporter(
+		ctx,
+		component.ExporterCreateParams{Logger: logger},
+		cfg.Exporters["datadog"],
+	)
+	assert.Nil(t, err)
+	assert.NotNil(t, exp)
+}
+
+func TestCreateAPIMetricsExporter(t *testing.T) {
+	logger := zap.NewNop()
+
+	factories, err := componenttest.ExampleComponents()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[configmodels.Type(typeStr)] = factory
+	cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	ctx := context.Background()
+	exp, err := factory.CreateMetricsExporter(
+		ctx,
+		component.ExporterCreateParams{Logger: logger},
+		cfg.Exporters["datadog/2"],
+	)
+
+	// Not implemented
+	assert.NotNil(t, err)
+	assert.Nil(t, exp)
+}
+
+func TestCreateAPITraceExporter(t *testing.T) {
+	logger := zap.NewNop()
+
+	factories, err := componenttest.ExampleComponents()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[configmodels.Type(typeStr)] = factory
+	cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	ctx := context.Background()
+	exp, err := factory.CreateTraceExporter(
+		ctx,
+		component.ExporterCreateParams{Logger: logger},
+		cfg.Exporters["datadog/2"],
+	)
+
+	// Not implemented
+	assert.NotNil(t, err)
+	assert.Nil(t, exp)
+}

exporter/datadogexporter/go.mod

@@ -3,6 +3,9 @@ module github.com/DataDog/opentelemetry-collector-contrib/exporter/datadogexport
 go 1.15
 
 require (
+	github.com/DataDog/datadog-go v3.7.2+incompatible
+	github.com/census-instrumentation/opencensus-proto v0.3.0
 	github.com/stretchr/testify v1.6.1
 	go.opentelemetry.io/collector v0.8.1-0.20200820203435-961c48b75778
+	go.uber.org/zap v1.15.0
 )