map Azure logs into OpenTelemetry fields/attributes

.chloggen/azehrec-parse.yaml

@@ -0,0 +1,16 @@
+# 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: adds alternate log formatter that maps Azure log fields into OpenTelemetry attributes
+
+# One or more tracking issues related to the change
+issues: [16283]
+
+# (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:

cmd/configschema/go.mod

@@ -551,6 +551,7 @@ require (
 	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/prometheus/prometheus v0.39.1 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/relvacode/iso8601 v1.1.0 // indirect
 	github.com/rs/cors v1.8.2 // indirect
 	github.com/scaleway/scaleway-sdk-go v1.0.0-beta.9 // indirect
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect

go.mod

@@ -142,7 +142,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/riakreceiver v0.66.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/saphanareceiver v0.66.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/sapmreceiver v0.66.0
-	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/signalfxreceiver v0.66.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/signalfxreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/simpleprometheusreceiver v0.66.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver v0.66.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/snmpreceiver v0.66.0
@@ -560,6 +560,7 @@ require (
 	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/prometheus/statsd_exporter v0.22.7 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/relvacode/iso8601 v1.1.0 // indirect
 	github.com/rs/cors v1.8.2 // indirect
 	github.com/scaleway/scaleway-sdk-go v1.0.0-beta.9 // indirect
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect

receiver/azureeventhubreceiver/README.md

@@ -7,7 +7,10 @@
 | Distributions            | [contrib] |
 
 ## Overview
-The Azure Event Hub receiver listens to logs emitted by Azure Event hubs.
+Azure resources and services can be
+[configured](https://learn.microsoft.com/en-us/azure/azure-monitor/essentials/diagnostic-settings)
+to send their logs to an Azure Event Hub. The Azure Event Hub receiver pulls logs from an Azure
+Event Hub, transforms them, and pushes them through the collector pipeline.
 
 ## Configuration
 
@@ -24,18 +27,65 @@ The offset at which to start watching the event hub. If empty, it starts with th
 
 Default: ""
 
-Example:
+### format (Optional)
+Determines how to transform the Event Hub messages into OpenTelemetry logs. See the "Format"
+section below for details.
+
+Default: "raw"
+
+### Example Configuration
 
 ```yaml
 receivers:
   azureeventhub:
     connection: Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=superSecret1234=;EntityPath=hubName
     partition: foo
     offset: "1234-5566"
+    format: "azure"
 ```
 
 This component can persist its state using the [storage extension].
 
+## Format
+
+### raw
+
+The "raw" format maps the AMQP properties and data into the
+attributes and body of an OpenTelemetry LogRecord, respectively.
+The body is represented as a raw byte array.
+
+### azure
+
+The "azure" format extracts the Azure log records from the AMQP
+message data, parses them, and maps the fields to OpenTelemetry
+attributes. The table below summarizes the mapping between the 
+[Azure common log format](https://learn.microsoft.com/en-us/azure/azure-monitor/essentials/resource-logs-schema)
+and the OpenTelemetry attributes.
+
+
+| Azure                            | OpenTelemetry                          | 
+|----------------------------------|----------------------------------------|
+| callerIpAddress (optional)       | net.sock.peer.addr (attribute)         | 
+| correlationId (optional)         | azure.correlation.id (attribute)       | 
+| category (optional)              | azure.category (attribute)             | 
+| durationMs (optional)            | azure.duration (attribute)             | 
+| Level (optional)                 | severity_number, severity_text (field) | 
+| location (optional)              | cloud.region (attribute)               | 
+| —                                | cloud.provider (attribute)             | 
+| operationName (required)         | azure.operation.name (attribute)       |
+| operationVersion (optional)      | azure.operation.version (attribute)    | 
+| properties (optional)            | azure.properties (attribute, nested)   | 
+| resourceId (required)            | azure.resource.id (resource attribute) | 
+| resultDescription (optional)     | azure.result.description (attribute)   | 
+| resultSignature (optional)       | azure.result.signature (attribute)     | 
+| resultType (optional)            | azure.result.type (attribute)          | 
+| tenantId (required, tenant logs) | azure.tenant.id (attribute)            | 
+| time (required)                  | time_unix_nano (field)                 | 
+| identity (optional)              | azure.identity (attribute, nested)     |
+
+Note: JSON does not distinguish between fixed and floating point numbers. All
+JSON numbers are encoded as doubles.
+
 [alpha]: https://github.com/open-telemetry/opentelemetry-collector#alpha
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 [storage extension]: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/storage

receiver/azureeventhubreceiver/azure_formatter.go

@@ -0,0 +1,34 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+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/plog"
+)
+
+type azureLogFormatConverter struct {
+	buildInfo component.BuildInfo
+}
+
+func newAzureLogFormatConverter(settings component.ReceiverCreateSettings) *azureLogFormatConverter {
+	return &azureLogFormatConverter{buildInfo: settings.BuildInfo}
+}
+
+func (c *azureLogFormatConverter) ToLogs(event *eventhub.Event) (*plog.Logs, error) {
+	logs, err := transform(c.buildInfo, event.Data)
+	return logs, err
+}

receiver/azureeventhubreceiver/azure_formatter_impl.go

@@ -0,0 +1,204 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package azureeventhubreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/azureeventhubreceiver"
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"strconv"
+
+	"github.com/relvacode/iso8601"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+)
+
+const azureCategory = "azure.category"
+const azureCorrelationID = "azure.correlation.id"
+const azureDuration = "azure.duration"
+const azureIdentity = "azure.identity"
+const azureOperationName = "azure.operation.name"
+const azureOperationVersion = "azure.operation.version"
+const azureProperties = "azure.properties"
+const azureResourceID = "azure.resource.id"
+const azureResultType = "azure.result.type"
+const azureResultSignature = "azure.result.signature"
+const azureResultDescription = "azure.result.description"
+const azureTenantID = "azure.tenant.id"
+
+const cloudProvider = "cloud.provider"
+const cloudRegion = "cloud.region"
+
+const netSockPeerAddr = "net.sock.peer.addr"
+
+// azureRecords represents an array of Azure log records
+// as exported via an Azure Event Hub
+type azureRecords struct {
+	Records []azureLogRecord `json:"records"`
+}
+
+// azureLogRecord represents a single Azure log following
+// the common schema:
+// https://learn.microsoft.com/en-us/azure/azure-monitor/essentials/resource-logs-schema
+type azureLogRecord struct {
+	Time              string       `json:"time"`
+	ResourceID        string       `json:"resourceId"`
+	TenantID          *string      `json:"tenantId"`
+	OperationName     string       `json:"operationName"`
+	OperationVersion  *string      `json:"operationVersion"`
+	Category          string       `json:"category"`
+	ResultType        *string      `json:"resultType"`
+	ResultSignature   *string      `json:"resultSignature"`
+	ResultDescription *string      `json:"resultDescription"`
+	DurationMs        *string      `json:"durationMs"`
+	CallerIPAddress   *string      `json:"callerIpAddress"`
+	CorrelationID     *string      `json:"correlationId"`
+	Identity          *interface{} `json:"identity"`
+	Level             *string      `json:"Level"`
+	Location          *string      `json:"location"`
+	Properties        *interface{} `json:"properties"`
+}
+
+// asTimestamp will parse an ISO8601 string into an OpenTelemetry
+// nanosecond timestamp. If the string cannot be parsed, it will
+// return zero and the error.
+func asTimestamp(s string) (pcommon.Timestamp, error) {
+	t, err := iso8601.ParseString(s)
+	if err != nil {
+		return 0, err
+	}
+	return pcommon.Timestamp(t.UnixNano()), nil
+}
+
+// asSeverity converts the Azure log level to equivalent
+// OpenTelemetry severity numbers. If the log level is not
+// valid, then the 'Unspecified' value is returned.
+func asSeverity(s string) plog.SeverityNumber {
+	switch s {
+	case "Informational":
+		return plog.SeverityNumberInfo
+	case "Warning":
+		return plog.SeverityNumberWarn
+	case "Error":
+		return plog.SeverityNumberError
+	case "Critical":
+		return plog.SeverityNumberFatal
+	default:
+		return plog.SeverityNumberUnspecified
+	}
+}
+
+// setIf will modify the given raw map by setting
+// the key and value iff the value is not null and
+// not the empty string.
+func setIf(attrs map[string]interface{}, key string, value *string) {
+	if value != nil && *value != "" {
+		attrs[key] = *value
+	}
+}
+
+// extractRawAttributes creates a raw attribute map and
+// inserts attributes from the Azure log record. Optional
+// attributes are only inserted if they are defined. The
+// azureDuration value is only set if the value in the
+// Azure log record can be parsed as an integer.
+func extractRawAttributes(log azureLogRecord) map[string]interface{} {
+	var attrs = map[string]interface{}{}
+
+	attrs[azureCategory] = log.Category
+	setIf(attrs, azureCorrelationID, log.CorrelationID)
+	if log.DurationMs != nil {
+		duration, err := strconv.ParseInt(*log.DurationMs, 10, 64)
+		if err == nil {
+			attrs[azureDuration] = duration
+		}
+	}
+	if log.Identity != nil {
+		attrs[azureIdentity] = *log.Identity
+	}
+	attrs[azureOperationName] = log.OperationName
+	setIf(attrs, azureOperationVersion, log.OperationVersion)
+	if log.Properties != nil {
+		attrs[azureProperties] = *log.Properties
+	}
+	setIf(attrs, azureResultDescription, log.ResultDescription)
+	setIf(attrs, azureResultSignature, log.ResultSignature)
+	setIf(attrs, azureResultType, log.ResultType)
+	setIf(attrs, azureTenantID, log.TenantID)
+
+	setIf(attrs, cloudRegion, log.Location)
+	attrs[cloudProvider] = "azure"
+
+	setIf(attrs, netSockPeerAddr, log.CallerIPAddress)
+
+	return attrs
+}
+
+// transform takes a byte array containing a JSON-encoded
+// payload with Azure log records and transforms it into
+// an OpenTelemetry plog.Logs object. The data in the Azure
+// log record appears as fields and attributes in the
+// OpenTelemetry representation; the bodies of the
+// OpenTelemetry log records are empty.
+func transform(buildInfo component.BuildInfo, data []byte) (*plog.Logs, error) {
+
+	l := plog.NewLogs()
+
+	var azureLogs azureRecords
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	if err := decoder.Decode(&azureLogs); err != nil {
+		return &l, err
+	}
+
+	resourceLogs := l.ResourceLogs().AppendEmpty()
+	scopeLogs := resourceLogs.ScopeLogs().AppendEmpty()
+	scopeLogs.Scope().SetName(fmt.Sprintf("otelcol/%s", typeStr))
+	scopeLogs.Scope().SetVersion(buildInfo.Version)
+	logRecords := scopeLogs.LogRecords()
+
+	resourceID := ""
+	for _, azureLog := range azureLogs.Records {
+		resourceID = azureLog.ResourceID
+		nanos, err := asTimestamp(azureLog.Time)
+		if err != nil {
+			continue
+		}
+
+		lr := logRecords.AppendEmpty()
+
+		lr.SetTimestamp(nanos)
+
+		if azureLog.Level != nil {
+			severity := asSeverity(*azureLog.Level)
+			lr.SetSeverityNumber(severity)
+			lr.SetSeverityText(*azureLog.Level)
+		}
+
+		//
+		if err := lr.Attributes().FromRaw(extractRawAttributes(azureLog)); err != nil {
+			return &l, err
+		}
+
+		// The Azure resource ID will be pulled into a common resource attribute.
+		// This implementation assumes that a single log message from Azure will
+		// contain ONLY logs from a single resource.
+		if resourceID != "" {
+			resourceLogs.Resource().Attributes().PutStr(azureResourceID, resourceID)
+		}
+	}
+
+	return &l, nil
+}