[exporter/datadog] Simplify and add tags to logs from Datadog Exporter (

#16080)

* Adding logic to translator

* Fix trace id transform and add tests

* Tags change and tests

* Address requested changes

* Add otel tag to logs exporter tests

* Fix tags issue by setting ddtags in sender opts

* Address PR changes

* Change tag name and add changelog

* Address PR comments

* Fix if else structure

* Remove logs

* Fix clean up

.chloggen/simplify-logs-from-datadog-exporter.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: exporter/datadog
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fixes bug to append tags in attributes instead of replacing them, simplifies filelog receiver setup, and adds `otel_source` tag.
+
+# One or more tracking issues related to the change
+issues: [15387]
+
+# (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:

exporter/datadogexporter/go.mod

@@ -53,6 +53,7 @@ require (
  github.com/Showmax/go-fqdn v1.0.0 // indirect
  github.com/antonmedv/expr v1.9.0 // indirect
  github.com/armon/go-metrics v0.3.10 // indirect
+ github.com/benbjohnson/clock v1.3.0 // indirect
  github.com/beorn7/perks v1.0.1 // indirect
  github.com/bmatcuk/doublestar/v3 v3.0.0 // indirect
  github.com/cenkalti/backoff v2.2.1+incompatible // indirect

exporter/datadogexporter/internal/logs/sender.go

@@ -16,6 +16,7 @@ package logs // import "github.com/open-telemetry/opentelemetry-collector-contri
 
 import (
  "context"
+ "fmt"
 
  "github.com/DataDog/datadog-api-client-go/v2/api/datadog"
  "github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
@@ -61,6 +62,15 @@ func NewSender(endpoint string, logger *zap.Logger, s exporterhelper.TimeoutSett
 // SubmitLogs submits the logs contained in payload to the Datadog intake
 func (s *Sender) SubmitLogs(ctx context.Context, payload []datadogV2.HTTPLogItem) error {
  s.logger.Debug("Submitting logs", zap.Any("payload", payload))
+
+ // Correctly sets apiSubmitLogRequest ddtags field based on tags from translator Transform method
+ if payload[0].HasDdtags() {
+ tags := datadog.PtrString(payload[0].GetDdtags())
+ if s.opts.Ddtags != nil {
+ tags = datadog.PtrString(fmt.Sprint(*s.opts.Ddtags, ",", payload[0].GetDdtags()))
+ }
+ s.opts.Ddtags = tags
+ }
  _, r, err := s.api.SubmitLog(ctx, payload, s.opts)
  if err != nil {
  b := make([]byte, 1024) // 1KB message max

exporter/datadogexporter/internal/logs/translator.go

@@ -16,6 +16,7 @@ package logs // import "github.com/open-telemetry/opentelemetry-collector-contri
 
 import (
  "encoding/binary"
+ "encoding/hex"
  "strconv"
  "strings"
  "time"
@@ -27,6 +28,7 @@ import (
  "go.opentelemetry.io/collector/pdata/pcommon"
  "go.opentelemetry.io/collector/pdata/plog"
  conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
+ "go.uber.org/zap"
 )
 
 const (
@@ -58,9 +60,12 @@ const (
  logLevelFatal = "fatal"
 )
 
+// otelTag specifies a tag to be added to all logs sent from the Datadog exporter
+const otelTag = "otel_source:datadog_exporter"
+
 // Transform converts the log record in lr, which came in with the resource in res to a Datadog log item.
 // the variable specifies if the log body should be sent as an attribute or as a plain message.
-func Transform(lr plog.LogRecord, res pcommon.Resource) datadogV2.HTTPLogItem {
+func Transform(lr plog.LogRecord, res pcommon.Resource, logger *zap.Logger) datadogV2.HTTPLogItem {
  host, service := extractHostNameAndServiceName(res.Attributes(), lr.Attributes())
 
  l := datadogV2.HTTPLogItem{
@@ -83,6 +88,34 @@ func Transform(lr plog.LogRecord, res pcommon.Resource) datadogV2.HTTPLogItem {
  l.Message = v.AsString()
  case "status", "severity", "level", "syslog.severity":
  status = v.AsString()
+ case "traceid", "contextmap.traceid", "oteltraceid":
+ traceID, err := decodeTraceID(v.AsString())
+ if err != nil {
+ logger.Warn("failed to decode trace id",
+ zap.String("trace_id", v.AsString()),
+ zap.Error(err))
+ break
+ }
+ if l.AdditionalProperties[ddTraceID] == "" {
+ l.AdditionalProperties[ddTraceID] = strconv.FormatUint(traceIDToUint64(traceID), 10)
+ l.AdditionalProperties[otelTraceID] = v.AsString()
+ }
+ case "spanid", "contextmap.spanid", "otelspanid":
+ spanID, err := decodeSpanID(v.AsString())
+ if err != nil {
+ logger.Warn("failed to decode span id",
+ zap.String("span_id", v.AsString()),
+ zap.Error(err))
+ break
+ }
+ if l.AdditionalProperties[ddSpanID] == "" {
+ l.AdditionalProperties[ddSpanID] = strconv.FormatUint(spanIDToUint64(spanID), 10)
+ l.AdditionalProperties[otelSpanID] = v.AsString()
+ }
+ case "ddtags":
+ var tags = append(attributes.TagsFromAttributes(res.Attributes()), v.AsString(), otelTag)
+ tagStr := strings.Join(tags, ",")
+ l.Ddtags = datadog.PtrString(tagStr)
  default:
  l.AdditionalProperties[k] = v.AsString()
  }
@@ -123,11 +156,12 @@ func Transform(lr plog.LogRecord, res pcommon.Resource) datadogV2.HTTPLogItem {
  l.Message = lr.Body().AsString()
  }
 
- var tags = attributes.TagsFromAttributes(res.Attributes())
- if len(tags) > 0 {
+ if !l.HasDdtags() {
+  var tags = append(attributes.TagsFromAttributes(res.Attributes()), otelTag)
  tagStr := strings.Join(tags, ",")
  l.Ddtags = datadog.PtrString(tagStr)
  }
+
  return l
 }
 
@@ -155,6 +189,18 @@ func extractHostNameAndServiceName(resourceAttrs pcommon.Map, logAttrs pcommon.M
  return host, service
 }
 
+func decodeTraceID(traceID string) ([16]byte, error) {
+ var ret [16]byte
+ _, err := hex.Decode(ret[:], []byte(traceID))
+ return ret, err
+}
+
+func decodeSpanID(spanID string) ([8]byte, error) {
+ var ret [8]byte
+ _, err := hex.Decode(ret[:], []byte(spanID))
+ return ret, err
+}
+
 // traceIDToUint64 converts 128bit traceId to 64 bit uint64
 func traceIDToUint64(b [16]byte) uint64 {
  return binary.BigEndian.Uint64(b[len(b)-8:])

exporter/datadogexporter/internal/logs/translator_test.go

@@ -24,9 +24,11 @@ import (
  "go.opentelemetry.io/collector/pdata/pcommon"
  "go.opentelemetry.io/collector/pdata/plog"
  conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
+ "go.uber.org/zap/zaptest"
 )
 
 func TestTransform(t *testing.T) {
+ testLogger := zaptest.NewLogger(t)
  traceID := [16]byte{0x08, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x0, 0x0, 0x0, 0x0, 0x0a}
  var spanID [8]byte
  copy(spanID[:], traceID[8:])
@@ -55,6 +57,7 @@ func TestTransform(t *testing.T) {
  res: pcommon.NewResource(),
  },
  want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
  Message: *datadog.PtrString(""),
  AdditionalProperties: map[string]string{
  "app": "test",
@@ -80,7 +83,35 @@ func TestTransform(t *testing.T) {
  }(),
  },
  want: datadogV2.HTTPLogItem{
- Ddtags: datadog.PtrString("service:otlp_col"),
+ Ddtags: datadog.PtrString("service:otlp_col,otel_source:datadog_exporter"),
+ Message: *datadog.PtrString(""),
+ Service: datadog.PtrString("otlp_col"),
+ AdditionalProperties: map[string]string{
+ "app": "test",
+ "status": "debug",
+ otelSeverityNumber: "5",
+ },
+ },
+ },
+ {
+ // appends tags in attributes instead of replacing them
+ name: "append tags",
+ args: args{
+ lr: func() plog.LogRecord {
+ l := plog.NewLogRecord()
+ l.Attributes().PutStr("app", "test")
+ l.Attributes().PutStr("ddtags", "foo:bar")
+ l.SetSeverityNumber(5)
+ return l
+ }(),
+ res: func() pcommon.Resource {
+ r := pcommon.NewResource()
+ r.Attributes().PutStr(conventions.AttributeServiceName, "otlp_col")
+ return r
+ }(),
+ },
+ want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("service:otlp_col,foo:bar,otel_source:datadog_exporter"),
  Message: *datadog.PtrString(""),
  Service: datadog.PtrString("otlp_col"),
  AdditionalProperties: map[string]string{
@@ -107,6 +138,7 @@ func TestTransform(t *testing.T) {
  }(),
  },
  want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
  Message: *datadog.PtrString(""),
  Service: datadog.PtrString("otlp_col"),
  AdditionalProperties: map[string]string{
@@ -135,6 +167,7 @@ func TestTransform(t *testing.T) {
  }(),
  },
  want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
  Message: *datadog.PtrString(""),
  Service: datadog.PtrString("otlp_col"),
  AdditionalProperties: map[string]string{
@@ -149,6 +182,70 @@ func TestTransform(t *testing.T) {
  },
  },
  },
+ {
+ name: "trace from attributes",
+ args: args{
+ lr: func() plog.LogRecord {
+ l := plog.NewLogRecord()
+ l.Attributes().PutStr("app", "test")
+ l.Attributes().PutStr("spanid", "2e26da881214cd7c")
+ l.Attributes().PutStr("traceid", "437ab4d83468c540bb0f3398a39faa59")
+ l.Attributes().PutStr(conventions.AttributeServiceName, "otlp_col")
+ l.SetSeverityNumber(5)
+ return l
+ }(),
+ res: func() pcommon.Resource {
+ r := pcommon.NewResource()
+ return r
+ }(),
+ },
+ want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
+ Message: *datadog.PtrString(""),
+ Service: datadog.PtrString("otlp_col"),
+ AdditionalProperties: map[string]string{
+ "app": "test",
+ "status": "debug",
+ otelSeverityNumber: "5",
+ otelSpanID: "2e26da881214cd7c",
+ otelTraceID: "437ab4d83468c540bb0f3398a39faa59",
+ ddSpanID: "3325585652813450620",
+ ddTraceID: "13479048940416379481",
+ "service.name": "otlp_col",
+ },
+ },
+ },
+ {
+ name: "trace from attributes decode error",
+ args: args{
+ lr: func() plog.LogRecord {
+ l := plog.NewLogRecord()
+ l.Attributes().PutStr("app", "test")
+ l.Attributes().PutStr("spanid", "2e26da881214cd7c")
+ l.Attributes().PutStr("traceid", "invalidtraceid")
+ l.Attributes().PutStr(conventions.AttributeServiceName, "otlp_col")
+ l.SetSeverityNumber(5)
+ return l
+ }(),
+ res: func() pcommon.Resource {
+ r := pcommon.NewResource()
+ return r
+ }(),
+ },
+ want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
+ Message: *datadog.PtrString(""),
+ Service: datadog.PtrString("otlp_col"),
+ AdditionalProperties: map[string]string{
+ "app": "test",
+ "status": "debug",
+ otelSeverityNumber: "5",
+ otelSpanID: "2e26da881214cd7c",
+ ddSpanID: "3325585652813450620",
+ "service.name": "otlp_col",
+ },
+ },
+ },
  {
  // here SeverityText should take precedence for log status
  name: "SeverityText",
@@ -169,6 +266,7 @@ func TestTransform(t *testing.T) {
  }(),
  },
  want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
  Message: *datadog.PtrString(""),
  Service: datadog.PtrString("otlp_col"),
  AdditionalProperties: map[string]string{
@@ -203,6 +301,7 @@ func TestTransform(t *testing.T) {
  }(),
  },
  want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
  Message: *datadog.PtrString(""),
  Service: datadog.PtrString("otlp_col"),
  AdditionalProperties: map[string]string{
@@ -237,6 +336,7 @@ func TestTransform(t *testing.T) {
  }(),
  },
  want: datadogV2.HTTPLogItem{
+ Ddtags: datadog.PtrString("otel_source:datadog_exporter"),
  Message: *datadog.PtrString(""),
  Service: datadog.PtrString("otlp_col"),
  AdditionalProperties: map[string]string{
@@ -254,7 +354,7 @@ func TestTransform(t *testing.T) {
  }
  for _, tt := range tests {
  t.Run(tt.name, func(t *testing.T) {
- got := Transform(tt.args.lr, tt.args.res)
+ got := Transform(tt.args.lr, tt.args.res, testLogger)
 
  gs, err := got.MarshalJSON()
  if err != nil {

exporter/datadogexporter/logs_exporter.go

@@ -94,7 +94,7 @@ func (exp *logsExporter) consumeLogs(ctx context.Context, ld plog.Logs) (err err
  // iterate over Logs
  for k := 0; k < lsl.Len(); k++ {
  log := lsl.At(k)
- payload = append(payload, logs.Transform(log, res))
+ payload = append(payload, logs.Transform(log, res, exp.params.Logger))
  }
  }
  }

exporter/datadogexporter/logs_exporter_test.go

@@ -58,6 +58,7 @@ func TestLogsExporter(t *testing.T) {
  "status": "Info",
  "dd.span_id": fmt.Sprintf("%d", spanIDToUint64(ld.SpanID())),
  "dd.trace_id": fmt.Sprintf("%d", traceIDToUint64(ld.TraceID())),
+ "ddtags": "otel_source:datadog_exporter",
  "otel.severity_text": "Info",
  "otel.severity_number": "9",
  "otel.span_id": ld.SpanID().HexString(),
@@ -86,6 +87,7 @@ func TestLogsExporter(t *testing.T) {
  "status": "Info",
  "dd.span_id": fmt.Sprintf("%d", spanIDToUint64(ld.SpanID())),
  "dd.trace_id": fmt.Sprintf("%d", traceIDToUint64(ld.TraceID())),
+ "ddtags": "otel_source:datadog_exporter",
  "otel.severity_text": "Info",
  "otel.severity_number": "9",
  "otel.span_id": ld.SpanID().HexString(),