Slight refactoring and adding processing for logs

processor/datadogsemanticsprocessor/README.md

@@ -2,7 +2,7 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [development]: traces   |
+| Stability     | [development]: traces, logs   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aprocessor%2Fdatadogsemantics%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aprocessor%2Fdatadogsemantics) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aprocessor%2Fdatadogsemantics%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aprocessor%2Fdatadogsemantics) |
 | Code coverage | [![codecov](https://codecov.io/github/open-telemetry/opentelemetry-collector-contrib/graph/main/badge.svg?component=processor_datadogsemantics)](https://app.codecov.io/gh/open-telemetry/opentelemetry-collector-contrib/tree/main/?components%5B0%5D=processor_datadogsemantics&displayType=list) |

processor/datadogsemanticsprocessor/factory.go

@@ -23,6 +23,7 @@ func NewFactory() processor.Factory {
 		metadata.Type,
 		createDefaultConfig,
 		processor.WithTraces(createTracesProcessor, metadata.TracesStability),
+		processor.WithLogs(createLogsProcessor, metadata.LogsStability),
 	)
 }
 
@@ -71,3 +72,43 @@ func createTracesProcessor(
 		processorhelper.WithCapabilities(consumerCapabilities),
 	)
 }
+
+type logsProcessor struct {
+	overrideIncomingDatadogFields bool
+	attrsTranslator               *attributes.Translator
+}
+
+func newLogsProcessor(
+	set processor.Settings,
+	cfg *Config,
+) (*logsProcessor, error) {
+	attrsTranslator, err := attributes.NewTranslator(set.TelemetrySettings)
+	if err != nil {
+		return nil, err
+	}
+	return &logsProcessor{
+		overrideIncomingDatadogFields: cfg.OverrideIncomingDatadogFields,
+		attrsTranslator:               attrsTranslator,
+	}, nil
+}
+
+func createLogsProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	next consumer.Logs,
+) (processor.Logs, error) {
+	oCfg := cfg.(*Config)
+	lp, err := newLogsProcessor(set, oCfg)
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewLogs(
+		ctx,
+		set,
+		cfg,
+		next,
+		lp.processLogs,
+		processorhelper.WithCapabilities(consumerCapabilities),
+	)
+}

processor/datadogsemanticsprocessor/internal/attributes/attributes.go

@@ -0,0 +1,25 @@
+package attributes
+
+import (
+	"errors"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+var (
+	SignalTypeLogs = attribute.NewSet(attribute.String("signal", "logs"))
+)
+
+func InsertAttrIfMissingOrShouldOverride(overrideIncoming bool, sattr pcommon.Map, key string, value any) (err error) {
+	if _, ok := sattr.Get(key); overrideIncoming || !ok {
+		switch v := value.(type) {
+		case string:
+			sattr.PutStr(key, v)
+		case int64:
+			sattr.PutInt(key, v)
+		default:
+			err = errors.New("unsupported value type")
+		}
+	}
+	return err
+}

processor/datadogsemanticsprocessor/logs_processor.go

@@ -0,0 +1,130 @@
+package datadogsemanticsprocessor
+
+import (
+	"context"
+	"fmt"
+	"github.com/DataDog/datadog-agent/pkg/opentelemetry-mapping-go/otlp/attributes/source"
+	"github.com/DataDog/datadog-agent/pkg/trace/traceutil"
+	"github.com/DataDog/datadog-agent/pkg/trace/transform"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/datadogsemanticsprocessor/internal/attributes"
+	"go.opentelemetry.io/collector/pdata/plog"
+	semconv "go.opentelemetry.io/otel/semconv/v1.32.0"
+)
+
+func (lp *logsProcessor) processLogs(ctx context.Context, logs plog.Logs) (output plog.Logs, err error) {
+	resourceLogs := logs.ResourceLogs()
+	for i := 0; i < resourceLogs.Len(); i++ {
+		resourceLog := resourceLogs.At(i)
+		otelResource := resourceLog.Resource()
+		resourceAttrs := otelResource.Attributes()
+		for j := 0; j < resourceLog.ScopeLogs().Len(); j++ {
+			serviceVersion := ""
+			if serviceVersionAttr, ok := resourceAttrs.Get(string(semconv.ServiceVersionKey)); ok {
+				serviceVersion = serviceVersionAttr.AsString()
+			}
+			err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, resourceAttrs, "datadog.version", serviceVersion)
+			if err != nil {
+				return plog.Logs{}, err
+			}
+
+			if lp.overrideIncomingDatadogFields {
+				resourceAttrs.Remove("datadog.host.name")
+			}
+
+			datadogHostName := ""
+			if src, ok := lp.attrsTranslator.ResourceToSource(ctx, otelResource, attributes.SignalTypeLogs, nil); ok && src.Kind == source.HostnameKind {
+				datadogHostName = src.Identifier
+			}
+
+			err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, resourceAttrs, "datadog.host.name", datadogHostName)
+			if err != nil {
+				return plog.Logs{}, err
+			}
+
+			// TODO: maybe this stuff?
+			scopeLog := resourceLog.ScopeLogs().At(j)
+			for k := 0; k < scopeLog.LogRecords().Len(); k++ {
+				otelLog := scopeLog.LogRecords().At(k)
+				attrs := otelLog.Attributes()
+
+				// Note: default value from GetOTelService is "otlpresourcenoservicename"
+				err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, resourceAttrs, "datadog.service", attrs.)
+				if err != nil {
+					return plog.Logs{}, err
+				}
+
+				env := "default"
+				if envFromAttr := transform.GetOTelEnv(otelspan, otelres, false); envFromAttr != "" {
+					env = envFromAttr
+				}
+				err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, rattr, "datadog.env", env)
+				if err != nil {
+					return plog.Logs{}, err
+				}
+
+				err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.name", traceutil.GetOTelOperationNameV2(otelspan, otelres))
+				if err != nil {
+					return plog.Logs{}, err
+				}
+				err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.resource", traceutil.GetOTelResourceV2(otelspan, otelres))
+				if err != nil {
+					return plog.Logs{}, err
+				}
+				err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.type", traceutil.GetOTelSpanType(otelspan, otelres))
+				if err != nil {
+					return plog.Logs{}, err
+				}
+				spanKind := otelspan.Kind()
+				err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.span.kind", traceutil.OTelSpanKindName(spanKind))
+				if err != nil {
+					return plog.Logs{}, err
+				}
+
+				// Map VCS (version control system) attributes for source code integration
+				if vcsRevision, ok := sattr.Get(string(semconv.VCSRefHeadRevisionKey)); ok {
+					err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "git.commit.sha", vcsRevision.AsString())
+					if err != nil {
+						return plog.Logs{}, err
+					}
+				}
+				if vcsRepoURL, ok := sattr.Get(string(semconv.VCSRepositoryURLFullKey)); ok {
+					// Strip protocol from repository URL as required by Datadog
+					cleanURL := stripProtocolFromURL(vcsRepoURL.AsString())
+					err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "git.repository_url", cleanURL)
+					if err != nil {
+						return plog.Logs{}, err
+					}
+				}
+
+				metaMap := make(map[string]string)
+				code := transform.GetOTelStatusCode(otelspan, otelres, false)
+				if code != 0 {
+					err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.http_status_code", fmt.Sprintf("%d", code))
+					if err != nil {
+						return plog.Logs{}, err
+					}
+				}
+				ddError := int64(status2Error(otelspan.Status(), otelspan.Events(), metaMap))
+				err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.error", ddError)
+				if err != nil {
+					return plog.Logs{}, err
+				}
+				if ddError == 1 {
+					err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.error.msg", metaMap["error.msg"])
+					if err != nil {
+						return plog.Logs{}, err
+					}
+					err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.error.type", metaMap["error.type"])
+					if err != nil {
+						return plog.Logs{}, err
+					}
+					err = attributes.InsertAttrIfMissingOrShouldOverride(lp.overrideIncomingDatadogFields, sattr, "datadog.error.stack", metaMap["error.stack"])
+					if err != nil {
+						return plog.Logs{}, err
+					}
+				}
+			}
+		}
+	}
+	return logs, nil
+}

processor/datadogsemanticsprocessor/metadata.yaml

@@ -3,7 +3,7 @@ type: datadogsemantics
 status:
   class: processor
   stability:
-    development: [traces]
+    development: [traces, logs]
   distributions: [contrib]
   codeowners:
     active: [songy23, IbraheemA, mx-psi, dineshg13, ankitpatel96, jade-guiton-dd, jackgopack4]