[exporter/datadogexporter] Add logs support

cmd/configschema/go.mod

@@ -53,6 +53,7 @@ require (
 	github.com/DataDog/datadog-agent/pkg/util/cgroups v0.40.0-rc.2 // indirect
 	github.com/DataDog/datadog-agent/pkg/util/log v0.40.0-rc.2 // indirect
 	github.com/DataDog/datadog-agent/pkg/util/scrubber v0.40.0-rc.2 // indirect
+	github.com/DataDog/datadog-api-client-go/v2 v2.2.0 // indirect
 	github.com/DataDog/datadog-go/v5 v5.1.1 // indirect
 	github.com/DataDog/gohai v0.0.0-20220718130825-1776f9beb9cc // indirect
 	github.com/DataDog/sketches-go v1.4.1 // indirect

exporter/datadogexporter/README.md

@@ -2,11 +2,14 @@
 
 | Status                   |                  |
 | ------------------------ |------------------|
-| Stability                | [beta]           |
-| Supported pipeline types | traces, metrics  |
+| Stability                | traces [beta]    |
+|                          | metrics [beta]   |
+|                          | logs [alpha]     |
+| Supported pipeline types | traces, metrics, logs|
 | Distributions            | [contrib], [AWS] |
 
-This exporter sends metric and trace data to [Datadog](https://datadoghq.com). For environment specific setup instructions visit the [Datadog Documentation](https://docs.datadoghq.com/tracing/setup_overview/open_standards/otel_collector_datadog_exporter/).
+
+This exporter sends metric, traces and logs to [Datadog](https://datadoghq.com). For environment specific setup instructions visit the [Datadog Documentation](https://docs.datadoghq.com/tracing/setup_overview/open_standards/otel_collector_datadog_exporter/).
 
 > Please review the Collector's [security
 > documentation](https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/security.md),
@@ -53,5 +56,6 @@ Retry settings will only affect metrics.
 *Please Note:* Currently [Span Events](https://github.com/open-telemetry/opentelemetry-specification/blob/11cc73939a32e3a2e6f11bdeab843c61cf8594e9/specification/trace/api.md#add-events) are extracted and added to Spans as Json on the Datadog Span Tag `events`.
 
 [beta]:https://github.com/open-telemetry/opentelemetry-collector#beta
+[alpha]:https://github.com/open-telemetry/opentelemetry-collector#alpha
 [contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 [AWS]:https://aws-otel.github.io/docs/partners/datadog

exporter/datadogexporter/config.go

@@ -242,6 +242,19 @@ type TracesConfig struct {
 	flushInterval float64
 }
 
+// LogsConfig defines logs exporter specific configuration
+type LogsConfig struct {
+	// TCPAddr.Endpoint is the host of the Datadog intake server to send logs to.
+	// If unset, the value is obtained from the Site.
+	confignet.TCPAddr `mapstructure:",squash"`
+
+	// SendLogRecordBody when enabled would send string representation of body field of log record as message to Datadog
+	// By default this is enabled ,when we parse the body string into attributes we may want to disable
+	// Set this to true, if we are unable to parse body or we want use datadog backend for parsing
+	// It is essential for Log Trace correlation to work, that we parse the body before reaching exporter.
+	SendLogRecordBody bool `mapstructure:"send_log_record_body"`
+}
+
 // TagsConfig defines the tag-related configuration
 // It is embedded in the configuration
 type TagsConfig struct {
@@ -339,6 +352,9 @@ type Config struct {
 	// Traces defines the Traces exporter specific configuration
 	Traces TracesConfig `mapstructure:"traces"`
 
+	// Logs defines the Logs exporter specific configuration
+	Logs LogsConfig `mapstructure:"logs"`
+
 	// HostMetadata defines the host metadata specific configuration
 	HostMetadata HostMetadataConfig `mapstructure:"host_metadata"`
 
@@ -483,5 +499,9 @@ func (c *Config) Unmarshal(configMap *confmap.Conf) error {
 	if !configMap.IsSet("traces::endpoint") {
 		c.Traces.TCPAddr.Endpoint = fmt.Sprintf("https://trace.agent.%s", c.API.Site)
 	}
+	if !configMap.IsSet("logs::endpoint") {
+		// this is v2 endpoint
+		c.Logs.TCPAddr.Endpoint = fmt.Sprintf("https://http-intake.logs.%s", c.API.Site)
+	}
 	return nil
 }

exporter/datadogexporter/factory.go

@@ -28,6 +28,7 @@ import (
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/featuregate"
 	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 
@@ -39,7 +40,8 @@ const (
 	// typeStr is the type of the exporter
 	typeStr = "datadog"
 	// The stability level of the exporter.
-	stability = component.StabilityLevelBeta
+	stability     = component.StabilityLevelBeta
+	logsStability = component.StabilityLevelAlpha
 )
 
 type factory struct {
@@ -66,6 +68,7 @@ func newFactoryWithRegistry(registry *featuregate.Registry) component.ExporterFa
 		f.createDefaultConfig,
 		component.WithMetricsExporter(f.createMetricsExporter, stability),
 		component.WithTracesExporter(f.createTracesExporter, stability),
+		component.WithLogsExporter(f.createLogsExporter, logsStability),
 	)
 }
 
@@ -125,6 +128,13 @@ func (f *factory) createDefaultConfig() config.Exporter {
 			IgnoreResources: []string{},
 		},
 
+		Logs: LogsConfig{
+			TCPAddr: confignet.TCPAddr{
+				Endpoint: "https://http-intake.logs.datadoghq.com",
+			},
+			SendLogRecordBody: true,
+		},
+
 		HostMetadata: HostMetadataConfig{
 			Enabled:        true,
 			HostnameSource: hostnameSource,
@@ -264,3 +274,51 @@ func (f *factory) createTracesExporter(
 		exporterhelper.WithShutdown(stop),
 	)
 }
+
+// createLogsExporter creates a trace exporter based on this config.
+func (f *factory) createLogsExporter(
+	ctx context.Context,
+	set component.ExporterCreateSettings,
+	c config.Exporter,
+) (component.LogsExporter, error) {
+	cfg := checkAndCastConfig(c)
+
+	var pusher consumer.ConsumeLogsFunc
+	hostProvider, err := f.SourceProvider(set.TelemetrySettings, cfg.Hostname)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build hostname provider: %w", err)
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	// cancel() runs on shutdown
+	if cfg.OnlyMetadata {
+		// only host metadata needs to be sent, once.
+		pusher = func(_ context.Context, td plog.Logs) error {
+			f.onceMetadata.Do(func() {
+				attrs := pcommon.NewMap()
+				go metadata.Pusher(ctx, set, newMetadataConfigfromConfig(cfg), hostProvider, attrs)
+			})
+			return nil
+		}
+	} else {
+		exp, err := newLogsExporter(ctx, set, cfg, &f.onceMetadata, hostProvider)
+		if err != nil {
+			cancel()
+			return nil, err
+		}
+		pusher = exp.consumeLogs
+	}
+
+	return exporterhelper.NewLogsExporter(
+		ctx,
+		set,
+		cfg,
+		pusher,
+		exporterhelper.WithTimeout(exporterhelper.TimeoutSettings{Timeout: 0 * time.Second}),
+		exporterhelper.WithRetry(cfg.RetrySettings),
+		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithShutdown(func(context.Context) error {
+			cancel()
+			return nil
+		}),
+	)
+}