[receiver/datadogreceiver] Adding logs support to the datadog receiver (open-telemetry#43841)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
Adding a feature to the datadog receiver to support logs, in addition to
the currently supported traces and metrics. This change adds support for
the `api/v2/logs` endpoint.

Note: There's purposefully no datadog semantics or conforming in this
changeset because I plan to make those changes next in the
[ddsemanticsprocessor](https://github.com/Sawmills/opentelemetry-collector-contrib/tree/e70d1b2ba66405e10dfc02afc170f8e2462b4a36/processor/datadogsemanticsprocessor).
I didn't think they belonged here.

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes

<!--Describe what testing was performed and which tests were added.-->
#### Testing
- Built a custom collector locally with these changes and verified with
the debug exporter that otel logs are being exported properly
- Added tests for the log translation layer
- Added e2e tests for the receiver for single and multiple logs

<!--Describe the documentation added.-->
#### Documentation
Updated the `receiver/datadogreceiver/README.md` to now show support for
logs.

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Co-authored-by: Robin Law <[email protected]>

Author: bvigar

.chloggen/ddreceiver-logs-support.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: receiver/datadog
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Adding log telemetry functionality to the existing datadog receiver component."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [43841]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/datadogreceiver/README.md

@@ -3,7 +3,7 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [alpha]: traces, metrics   |
+| Stability     | [alpha]: traces, metrics, logs   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fdatadog%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fdatadog) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fdatadog%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fdatadog) |
 | Code coverage | [![codecov](https://codecov.io/github/open-telemetry/opentelemetry-collector-contrib/graph/main/badge.svg?component=receiver_datadog)](https://app.codecov.io/gh/open-telemetry/opentelemetry-collector-contrib/tree/main/?components%5B0%5D=receiver_datadog&displayType=list) |
@@ -16,7 +16,7 @@
 ## Overview
 
 The Datadog receiver enables translation between Datadog and OpenTelemetry-compatible backends.
-It currently has support for Datadog's APM traces and Datadog metrics.
+It currently has support for Datadog's APM traces, metrics, and logs.
 
 ## Configuration
 
@@ -120,6 +120,12 @@ Format example can be found [here](./internal/translator/traces_translator_test.
 | /api/v1/distribution_points | Development |                            |
 | /intake                     | Development | Support for proxying calls |
 
+**Logs**
+
+| Datadog API Endpoint | Status      | Notes                        |
+|----------------------|-------------|------------------------------|
+| /api/v2/logs         | Development | Support for msgpack and json |
+
 ### Temporality considerations
 
 Some backends use a different [timestamp temporality](https://opentelemetry.io/docs/specs/otel/metrics/data-model/#temporality) than Datadog uses. Both delta and cumulative temporalities are allowed in the spec.

receiver/datadogreceiver/factory.go

@@ -23,7 +23,9 @@ func NewFactory() receiver.Factory {
 		metadata.Type,
 		createDefaultConfig,
 		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
-		receiver.WithTraces(createTracesReceiver, metadata.TracesStability))
+		receiver.WithTraces(createTracesReceiver, metadata.TracesStability),
+		receiver.WithLogs(createLogsReceiver, metadata.LogsStability),
+	)
 }
 
 func createDefaultConfig() component.Config {
@@ -74,4 +76,19 @@ func createMetricsReceiver(ctx context.Context, params receiver.Settings, cfg co
 	return r, nil
 }
 
+func createLogsReceiver(ctx context.Context, params receiver.Settings, cfg component.Config, consumer consumer.Logs) (receiver.Logs, error) {
+	var err error
+	rcfg := cfg.(*Config)
+	r := receivers.GetOrAdd(cfg, func() (dd component.Component) {
+		dd, err = newDataDogReceiver(ctx, rcfg, params)
+		return dd
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	r.Unwrap().(*datadogReceiver).nextLogsConsumer = consumer
+	return r, nil
+}
+
 var receivers = sharedcomponent.NewSharedComponents()

receiver/datadogreceiver/generated_component_test.go

@@ -15,6 +15,8 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.139.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.139.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/datadog v0.139.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.139.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.139.0
 	github.com/stretchr/testify v1.11.1
 	github.com/tinylib/msgp v1.5.0
 	github.com/vmihailenco/msgpack/v5 v5.4.1

receiver/datadogreceiver/go.mod

@@ -0,0 +1,42 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package translator // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/datadogreceiver/internal/translator"
+import (
+	"go.opentelemetry.io/collector/pdata/plog"
+)
+
+// I think the struct we should be unmarshaling into (below) should work, but doesn't include the status field for some reason...
+// https://github.com/DataDog/datadog-api-client-go/blob/151a146e7ae57e5e8ed620a9ae6f958ed7db1ca7/api/datadogV2/model_http_log_item.go#L14-L34
+type DatadogLogPayload struct {
+	Message   string `json:"message"`
+	Status    string `json:"status"`
+	Timestamp int64  `json:"timestamp"`
+	Hostname  string `json:"hostname"`
+	Service   string `json:"service"`
+	Source    string `json:"ddsource"`
+	Tags      string `json:"ddtags"`
+}
+
+func ToPlog(incomingLogs []*DatadogLogPayload) plog.Logs {
+	plogPayload := plog.NewLogs()
+	if len(incomingLogs) == 0 {
+		return plogPayload
+	}
+
+	resourceLogs := plogPayload.ResourceLogs().AppendEmpty()
+	scopeLogs := resourceLogs.ScopeLogs().AppendEmpty()
+	scopeLogs.LogRecords().EnsureCapacity(len(incomingLogs))
+	for _, incomingLog := range incomingLogs {
+		logRecord := scopeLogs.LogRecords().AppendEmpty()
+		logRecord.Body().SetStr(incomingLog.Message)
+		logRecord.Attributes().PutStr("status", incomingLog.Status)
+		logRecord.Attributes().PutInt("timestamp", incomingLog.Timestamp)
+		logRecord.Attributes().PutStr("hostname", incomingLog.Hostname)
+		logRecord.Attributes().PutStr("service", incomingLog.Service)
+		logRecord.Attributes().PutStr("ddsource", incomingLog.Source)
+		logRecord.Attributes().PutStr("ddtags", incomingLog.Tags)
+	}
+
+	return plogPayload
+}