[receiver/azureeventhubreceiver]Implemented receiver to support traces

.chloggen/azureeventhub-trace-receiver.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. filelogreceiver)
+component: azureeventhubreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Added traces support in azureeventhubreceiver
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33583]
+
+# (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]

pkg/translator/azure/resources_to_traces.go

@@ -0,0 +1,184 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package azure // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/azure"
+
+import (
+	"bytes"
+
+	"encoding/hex"
+	"net/url"
+
+	jsoniter "github.com/json-iterator/go"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	conventions "go.opentelemetry.io/collector/semconv/v1.13.0"
+	"go.uber.org/zap"
+)
+
+const (
+	// Constants for OpenTelemetry Specs
+	traceAzureResourceID = "azure.resource.id"
+	traceScopeName       = "otelcol/azureresourcetraces"
+)
+
+type azureTracesRecords struct {
+	Records []azureTracesRecord `json:"records"`
+}
+
+// Azure Trace Records based on Azure AppRequests & AppDependencies table data
+// the common record schema reference:
+// https://learn.microsoft.com/en-us/azure/azure-monitor/reference/tables/apprequests
+// https://learn.microsoft.com/en-us/azure/azure-monitor/reference/tables/appdependencies
+type azureTracesRecord struct {
+	Time                  string             `json:"time"`
+	ResourceID            string             `json:"resourceId"`
+	ResourceGUID          string             `json:"ResourceGUID"`
+	Type                  string             `json:"Type"`
+	AppRoleInstance       string             `json:"AppRoleInstance"`
+	AppRoleName           string             `json:"AppRoleName"`
+	AppVersion            string             `json:"AppVersion"`
+	ClientCity            string             `json:"ClientCity"`
+	ClientCountryOrRegion string             `json:"ClientCountryOrRegion"`
+	ClientIP              string             `json:"ClientIP"`
+	ClientStateOrProvince string             `json:"ClientStateOrProvince"`
+	ClientType            string             `json:"ClientType"`
+	IKey                  string             `json:"IKey"`
+	OperationName         string             `json:"OperationName"`
+	OperationId           string             `json:"OperationId"`
+	ParentId              string             `json:"ParentId"`
+	SDKVersion            string             `json:"SDKVersion"`
+	Properties            map[string]string  `json:"Properties"`
+	Measurements          map[string]float64 `json:"Measurements"`
+	SpanId                string             `json:"Id"`
+	Name                  string             `json:"Name"`
+	Url                   string             `json:"Url"`
+	Source                string             `json:"Source"`
+	Success               bool               `json:"Success"`
+	ResultCode            string             `json:"ResultCode"`
+	DurationMs            float64            `json:"DurationMs"`
+	PerformanceBucket     string             `json:"PerformanceBucket"`
+	ItemCount             float64            `json:"ItemCount"`
+}
+
+var _ ptrace.Unmarshaler = (*TracesUnmarshaler)(nil)
+
+type TracesUnmarshaler struct {
+	Version string
+	Logger  *zap.Logger
+}
+
+func (r TracesUnmarshaler) UnmarshalTraces(buf []byte) (ptrace.Traces, error) {
+	t := ptrace.NewTraces()
+
+	var azureTraces azureTracesRecords
+	decoder := jsoniter.NewDecoder(bytes.NewReader(buf))
+	err := decoder.Decode(&azureTraces)
+	if err != nil {
+		return t, err
+	}
+
+	resourceTraces := t.ResourceSpans().AppendEmpty()
+	resource := resourceTraces.Resource()
+	resource.Attributes().PutStr(conventions.AttributeTelemetrySDKName, traceScopeName)
+	resource.Attributes().PutStr(conventions.AttributeTelemetrySDKLanguage, conventions.AttributeTelemetrySDKLanguageGo)
+	resource.Attributes().PutStr(conventions.AttributeTelemetrySDKVersion, r.Version)
+	resource.Attributes().PutStr(conventions.AttributeCloudProvider, conventions.AttributeCloudProviderAzure)
+
+	scopeSpans := resourceTraces.ScopeSpans().AppendEmpty()
+
+	spans := scopeSpans.Spans()
+
+	resourceID := ""
+	for _, azureTrace := range azureTraces.Records {
+		if resourceID == "" && azureTrace.ResourceID != "" {
+			resourceID = azureTrace.ResourceID
+		}
+
+		resource.Attributes().PutStr("service.name", azureTrace.AppRoleName)
+
+		nanos, err := asTimestamp(azureTrace.Time)
+		if err != nil {
+			r.Logger.Warn("Invalid Timestamp", zap.String("time", azureTrace.Time))
+			continue
+		}
+
+		var trace_id, trace_err = TraceIDFromHex(azureTrace.OperationId)
+		if trace_err != nil {
+			r.Logger.Warn("Invalid TraceID", zap.String("trace_id", azureTrace.OperationId))
+			return t, err
+		}
+		var span_id, span_err = SpanIDFromHex(azureTrace.SpanId)
+		if span_err != nil {
+			r.Logger.Warn("Invalid SpanID", zap.String("span_id", azureTrace.SpanId))
+			return t, err
+		}
+		var parent_id, parent_err = SpanIDFromHex(azureTrace.ParentId)
+		if parent_err != nil {
+			r.Logger.Warn("Invalid ParentID", zap.String("parent_id", azureTrace.ParentId))
+			return t, err
+		}
+
+		span := spans.AppendEmpty()
+		span.SetTraceID(trace_id)
+		span.SetSpanID(span_id)
+		span.SetParentSpanID(parent_id)
+
+		span.Attributes().PutStr("OperationName", azureTrace.OperationName)
+		span.Attributes().PutStr("AppRoleName", azureTrace.AppRoleName)
+		span.Attributes().PutStr("AppRoleInstance", azureTrace.AppRoleInstance)
+		span.Attributes().PutStr("Type", azureTrace.Type)
+
+		span.Attributes().PutStr("http.url", azureTrace.Url)
+
+		urlObj, _ := url.Parse(azureTrace.Url)
+		hostname := urlObj.Host
+		hostpath := urlObj.Path
+		scheme := urlObj.Scheme
+
+		span.Attributes().PutStr("http.host", hostname)
+		span.Attributes().PutStr("http.path", hostpath)
+		span.Attributes().PutStr("http.response.status_code", azureTrace.ResultCode)
+		span.Attributes().PutStr("http.client_ip", azureTrace.ClientIP)
+		span.Attributes().PutStr("http.client_city", azureTrace.ClientCity)
+		span.Attributes().PutStr("http.client_type", azureTrace.ClientType)
+		span.Attributes().PutStr("http.client_state", azureTrace.ClientStateOrProvince)
+		span.Attributes().PutStr("http.client_type", azureTrace.ClientType)
+		span.Attributes().PutStr("http.client_country", azureTrace.ClientCountryOrRegion)
+		span.Attributes().PutStr("http.scheme", scheme)
+		span.Attributes().PutStr("http.method", azureTrace.Properties["HTTP Method"])
+
+		span.SetKind(ptrace.SpanKindServer)
+		span.SetName(azureTrace.Name)
+		span.SetStartTimestamp(nanos)
+		span.SetEndTimestamp(nanos + pcommon.Timestamp(azureTrace.DurationMs*1e6))
+	}
+
+	if resourceID != "" {
+		resourceTraces.Resource().Attributes().PutStr(traceAzureResourceID, resourceID)
+	} else {
+		r.Logger.Warn("No ResourceID Set on Traces!")
+	}
+
+	return t, nil
+}
+
+func TraceIDFromHex(hexStr string) (pcommon.TraceID, error) {
+	bytes, err := hex.DecodeString(hexStr)
+	if err != nil {
+		return pcommon.TraceID{}, err
+	}
+	var id pcommon.TraceID
+	copy(id[:], bytes)
+	return id, nil
+}
+
+func SpanIDFromHex(hexStr string) (pcommon.SpanID, error) {
+	bytes, err := hex.DecodeString(hexStr)
+	if err != nil {
+		return pcommon.SpanID{}, err
+	}
+	var id pcommon.SpanID
+	copy(id[:], bytes)
+	return id, nil
+}

receiver/azureeventhubreceiver/README.md

@@ -3,7 +3,7 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [alpha]: metrics, logs   |
+| Stability     | [alpha]: metrics, logs, traces   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fazureeventhub%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fazureeventhub) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fazureeventhub%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fazureeventhub) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@atoulme](https://www.github.com/atoulme), [@cparkins](https://www.github.com/cparkins) \| Seeking more code owners! |
@@ -120,4 +120,16 @@ with a Data Points that represents the values
 for the Metric including: Total, Minimum, Maximum,
 Average and Count.
 
+Traces based on Azure Application Insights array of records from `AppRequests` & `AppDependencies` with the following fields.
+
+| Azure       | Open Telemetry                                        |
+|-------------|-------------------------------------------------------|
+| Time        | start_time(time_unix_nano(time))                      |
+|             | end_time(start_time + time_unix_nano(durationMs))     |
+| Name        | span.name                                             |
+| OperationId | trace.id                                              |
+| ParentId    | span.parentId                                         |
+| Id          | span.id                                               |
+| AppRoleName | service.name                                          |
+
 [storage extension]: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/storage

receiver/azureeventhubreceiver/azureresourcetraces_unmarshaler.go

@@ -0,0 +1,36 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package azureeventhubreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/azureeventhubreceiver"
+
+import (
+	eventhub "github.com/Azure/azure-event-hubs-go/v3"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/azure"
+)
+
+type azureTracesEventUnmarshaler struct {
+	unmarshaler *azure.TracesUnmarshaler
+}
+
+func newAzureTracesUnmarshaler(buildInfo component.BuildInfo, logger *zap.Logger) eventTracesUnmarshaler {
+	return azureTracesEventUnmarshaler{
+		unmarshaler: &azure.TracesUnmarshaler{
+			Version: buildInfo.Version,
+			Logger:  logger,
+		},
+	}
+}
+
+// UnmarshalTraces takes a byte array containing a JSON-encoded
+// payload with Azure records and transforms it into
+// an OpenTelemetry ptraces.traces object. The data in the Azure
+// record appears as fields and attributes in the
+// OpenTelemetry representation; the bodies of the
+// OpenTelemetry trace records are empty.
+func (r azureTracesEventUnmarshaler) UnmarshalTraces(event *eventhub.Event) (ptrace.Traces, error) {
+	return r.unmarshaler.UnmarshalTraces(event.Data)
+}

receiver/azureeventhubreceiver/eventhubhandler_test.go

@@ -57,15 +57,20 @@ func (m mockListenerHandleWrapper) Err() error {
 }
 
 type mockDataConsumer struct {
-	logsUnmarshaler  eventLogsUnmarshaler
-	nextLogsConsumer consumer.Logs
-	obsrecv          *receiverhelper.ObsReport
+	logsUnmarshaler    eventLogsUnmarshaler
+	nextLogsConsumer   consumer.Logs
+	nextTracesConsumer consumer.Traces
+	obsrecv            *receiverhelper.ObsReport
 }
 
 func (m *mockDataConsumer) setNextLogsConsumer(nextLogsConsumer consumer.Logs) {
 	m.nextLogsConsumer = nextLogsConsumer
 }
 
+func (m *mockDataConsumer) setNextTracesConsumer(nextTracesConsumer consumer.Traces) {
+	m.nextTracesConsumer = nextTracesConsumer
+}
+
 func (m *mockDataConsumer) setNextMetricsConsumer(_ consumer.Metrics) {}
 
 func (m *mockDataConsumer) consume(ctx context.Context, event *eventhub.Event) error {

receiver/azureeventhubreceiver/factory.go

@@ -38,7 +38,9 @@ func NewFactory() receiver.Factory {
 		metadata.Type,
 		createDefaultConfig,
 		receiver.WithLogs(f.createLogsReceiver, metadata.LogsStability),
-		receiver.WithMetrics(f.createMetricsReceiver, metadata.MetricsStability))
+		receiver.WithMetrics(f.createMetricsReceiver, metadata.MetricsStability),
+		receiver.WithTraces(f.createTracesReceiver, metadata.TracesStability),
+	)
 }
 
 func createDefaultConfig() component.Config {
@@ -79,6 +81,23 @@ func (f *eventhubReceiverFactory) createMetricsReceiver(
 	return receiver, nil
 }
 
+func (f *eventhubReceiverFactory) createTracesReceiver(
+	_ context.Context,
+	settings receiver.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Traces,
+) (receiver.Traces, error) {
+
+	receiver, err := f.getReceiver(component.DataTypeTraces, cfg, settings)
+	if err != nil {
+		return nil, err
+	}
+
+	receiver.(dataConsumer).setNextTracesConsumer(nextConsumer)
+
+	return receiver, nil
+}
+
 func (f *eventhubReceiverFactory) getReceiver(
 	receiverType component.Type,
 	cfg component.Config,
@@ -95,6 +114,7 @@ func (f *eventhubReceiverFactory) getReceiver(
 
 		var logsUnmarshaler eventLogsUnmarshaler
 		var metricsUnmarshaler eventMetricsUnmarshaler
+		var tracesUnmarshaler eventTracesUnmarshaler
 		switch receiverType {
 		case component.DataTypeLogs:
 			if logFormat(receiverConfig.Format) == rawLogFormat {
@@ -110,7 +130,12 @@ func (f *eventhubReceiverFactory) getReceiver(
 				metricsUnmarshaler = newAzureResourceMetricsUnmarshaler(settings.BuildInfo, settings.Logger)
 			}
 		case component.DataTypeTraces:
-			err = errors.New("unsupported traces data")
+			if logFormat(receiverConfig.Format) == rawLogFormat {
+				tracesUnmarshaler = nil
+				err = errors.New("raw format not supported for Traces")
+			} else {
+				tracesUnmarshaler = newAzureTracesUnmarshaler(settings.BuildInfo, settings.Logger)
+			}
 		}
 
 		if err != nil {
@@ -120,7 +145,7 @@ func (f *eventhubReceiverFactory) getReceiver(
 		eventHandler := newEventhubHandler(receiverConfig, settings)
 
 		var rcvr component.Component
-		rcvr, err = newReceiver(receiverType, logsUnmarshaler, metricsUnmarshaler, eventHandler, settings)
+		rcvr, err = newReceiver(receiverType, logsUnmarshaler, metricsUnmarshaler, tracesUnmarshaler, eventHandler, settings)
 		return rcvr
 	})
 

receiver/azureeventhubreceiver/metadata.yaml

@@ -4,7 +4,7 @@ scope_name: otelcol/azureeventhubreceiver
 status:
   class: receiver
   stability:
-    alpha: [metrics, logs]
+    alpha: [metrics, logs, traces]
   distributions: [contrib]
   codeowners:
     active: [atoulme, cparkins]