[exporter/azuremonitor] Add support for custom events in Azure Monitor exporter (#38609)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Add support for custom events in Azure Monitor exporter. Added a new
config `custom_events_enabled ` (default = `false`): Enables export log
record to custom events when there's attribute
`microsoft.custom_event.name` or
`APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE`. Since the
`custom_events_enabled ` is false by default, it'll not break current
behavior.

The logic is align with current azure-sdk-for-js.

https://github.com/Azure/azure-sdk-for-js/blob/main/sdk/monitor/monitor-opentelemetry-exporter/src/utils/logUtils.ts#L78C3-L78C67

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

<!--Describe what testing was performed and which tests were added.-->
#### Testing

- added new tests for custom event data
- validated using multiple azure application insights, with enable or
not enable `custom_events_enabled `.

<!--Describe the documentation added.-->
#### Documentation

added new optional configuration

```
- `custom_events_enabled ` (default = `false`): Enables export log record to custom events when there's attribute `microsoft.custom_event.name` or `APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE`.
```

added document for custom events
```
#### Custom Events

When `custom_events_enabled ` = `true`, azure monitor exporter will export log record to custom events when there's attribute `microsoft.custom_event.name` or `APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE`. 
```

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

Author: hgaol

.chloggen/37422.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: azuremonitorexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: support custom event for logs for azure monitor exporter
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37422]
+
+# (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]

exporter/azureblobexporter/README.md

@@ -30,11 +30,11 @@ The following settings can be optionally configured and have default values:
   - metrics (default `metrics`): container to store metrics. default value is `metrics`.
   - logs (default `logs`): container to store logs. default value is `logs`.
   - traces (default `traces`): container to store traces. default value is `traces`.
-- blob_name_format:
-  - metrics_format (default `2006/01/02/metrics_15_04_05_{{.SerialNum}}.{{.FileExtension}}`): blob name format. The date format follows constants in Golang, refer [here](https://go.dev/src/time/format.go).
-  - logs_format (default `2006/01/02/logs_15_04_05_{{.SerialNum}}.{{.FileExtension}}`): blob name format.
-  - traces_format (default `2006/01/02/traces_15_04_05_{{.SerialNum}}.{{.FileExtension}}`): blob name format.
-  - serial_num_range (default `10000`): a range of random number for SerialNum.
+- blob_name_format: the final blob name will be blob_name + 
+  - metrics_format (default `2006/01/02/metrics_15_04_05.json`): blob name format. The date format follows constants in Golang, refer [here](https://go.dev/src/time/format.go).
+  - logs_format (default `2006/01/02/logs_15_04_05.json`): blob name format.
+  - traces_format (default `2006/01/02/traces_15_04_05.json`): blob name format.
+  - serial_num_range (default `10000`): a range of random number to be appended after blob_name. e.g. `blob_name_{serial_num}`.
 - format (default `json`): `json` or `proto`. which present otel json or otel protobuf format, the file extension will be `json` or `pb`.
 - encodings (default using encoding specified in `format`, which is `json`): if specified, uses the encoding extension to encode telemetry data. Overrides format.
   - logs (default `nil`): encoding component id.

exporter/azuremonitorexporter/README.md

@@ -43,6 +43,7 @@ The following settings can be optionally configured:
   - `queue_size` (default = 1000): Maximum number of batches kept in memory before data; ignored if `enabled` is `false`
   - `storage` (default = `none`): When set, enables persistence and uses the component specified as a storage extension for the persistent queue
 - `shutdown_timeout` (default = 1s): Timeout to wait for graceful shutdown. Once exceeded, the component will shut down forcibly, dropping any element in queue.
+- `custom_events_enabled` (default = `false`): Enables export log record to custom events when there's attribute `microsoft.custom_event.name` or `APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE`.
 
 Example:
 
@@ -118,6 +119,10 @@ Exception events are saved to the Application Insights `exception` table.
 This exporter saves log records to Application Insights `traces` table.
 [TraceId](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/logs/data-model.md#field-traceid) is mapped to `operation_id` column and [SpanId](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/logs/data-model.md#field-spanid) is mapped to `operation_parentId` column.
 
+#### Custom Events
+
+When `custom_events_enabled` = `true`, azure monitor exporter will export log record to custom events when there's attribute `microsoft.custom_event.name` or `APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE`. 
+
 ### Metrics
 
 This exporter saves metrics to Application Insights `customMetrics` table.

exporter/azuremonitorexporter/azuremonitor_exporter.go

@@ -53,7 +53,7 @@ func (exporter *azureMonitorExporter) Shutdown(_ context.Context) (err error) {
 
 func (exporter *azureMonitorExporter) consumeLogs(_ context.Context, logData plog.Logs) error {
 	resourceLogs := logData.ResourceLogs()
-	logPacker := newLogPacker(exporter.logger)
+	logPacker := newLogPacker(exporter.logger, exporter.config)
 
 	for i := 0; i < resourceLogs.Len(); i++ {
 		scopeLogs := resourceLogs.At(i).ScopeLogs()

exporter/azuremonitorexporter/config.go

@@ -12,12 +12,13 @@ import (
 
 // Config defines configuration for Azure Monitor
 type Config struct {
-	QueueSettings      exporterhelper.QueueConfig `mapstructure:"sending_queue"`
-	Endpoint           string                     `mapstructure:"endpoint"`
-	ConnectionString   configopaque.String        `mapstructure:"connection_string"`
-	InstrumentationKey configopaque.String        `mapstructure:"instrumentation_key"`
-	MaxBatchSize       int                        `mapstructure:"maxbatchsize"`
-	MaxBatchInterval   time.Duration              `mapstructure:"maxbatchinterval"`
-	SpanEventsEnabled  bool                       `mapstructure:"spaneventsenabled"`
-	ShutdownTimeout    time.Duration              `mapstructure:"shutdown_timeout"`
+	QueueSettings       exporterhelper.QueueConfig `mapstructure:"sending_queue"`
+	Endpoint            string                     `mapstructure:"endpoint"`
+	ConnectionString    configopaque.String        `mapstructure:"connection_string"`
+	InstrumentationKey  configopaque.String        `mapstructure:"instrumentation_key"`
+	MaxBatchSize        int                        `mapstructure:"maxbatchsize"`
+	MaxBatchInterval    time.Duration              `mapstructure:"maxbatchinterval"`
+	SpanEventsEnabled   bool                       `mapstructure:"spaneventsenabled"`
+	ShutdownTimeout     time.Duration              `mapstructure:"shutdown_timeout"`
+	CustomEventsEnabled bool                       `mapstructure:"custom_events_enabled"`
 }

exporter/azuremonitorexporter/conventions.go

@@ -16,8 +16,10 @@ import (
 
 const (
 	// TODO replace with convention.* values once/if available
-	attributeOtelStatusCode        string = "otel.status_code"
-	attributeOtelStatusDescription string = "otel.status_description"
+	attributeOtelStatusCode                          string = "otel.status_code"
+	attributeOtelStatusDescription                   string = "otel.status_description"
+	attributeMicrosoftCustomEventName                string = "microsoft.custom_event.name"
+	attributeApplicationInsightsEventMarkerAttribute string = "APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE"
 )
 
 // NetworkAttributes is the set of known network attributes

exporter/azuremonitorexporter/factory.go

@@ -46,11 +46,12 @@ type factory struct {
 
 func createDefaultConfig() component.Config {
 	return &Config{
-		MaxBatchSize:      1024,
-		MaxBatchInterval:  10 * time.Second,
-		SpanEventsEnabled: false,
-		QueueSettings:     exporterhelper.NewDefaultQueueConfig(),
-		ShutdownTimeout:   1 * time.Second,
+		MaxBatchSize:        1024,
+		MaxBatchInterval:    10 * time.Second,
+		SpanEventsEnabled:   false,
+		QueueSettings:       exporterhelper.NewDefaultQueueConfig(),
+		ShutdownTimeout:     1 * time.Second,
+		CustomEventsEnabled: false,
 	}
 }
 

exporter/azuremonitorexporter/log_to_envelope.go

@@ -16,29 +16,48 @@ import (
 
 type logPacker struct {
 	logger *zap.Logger
+	config *Config
 }
 
-func (packer *logPacker) LogRecordToEnvelope(logRecord plog.LogRecord, resource pcommon.Resource, instrumentationScope pcommon.InstrumentationScope) *contracts.Envelope {
+func (packer *logPacker) initEnvelope(logRecord plog.LogRecord) (*contracts.Envelope, *contracts.Data) {
 	envelope := contracts.NewEnvelope()
 	envelope.Tags = make(map[string]string)
 	envelope.Time = toTime(timestampFromLogRecord(logRecord)).Format(time.RFC3339Nano)
+	return envelope, contracts.NewData()
+}
+
+func (packer *logPacker) handleEventData(envelope *contracts.Envelope, data *contracts.Data, logRecord plog.LogRecord) {
+	attributes := logRecord.Attributes()
+	eventData := contracts.NewEventData()
+	if val, ok := attributes.Get(attributeMicrosoftCustomEventName); ok {
+		eventData.Name = val.AsString()
+	} else if val, ok := attributes.Get(attributeApplicationInsightsEventMarkerAttribute); ok {
+		eventData.Name = val.AsString()
+	}
 
-	data := contracts.NewData()
+	eventData.Properties = make(map[string]string)
+	setAttributesAsProperties(attributes, eventData.Properties)
 
+	data.BaseData = eventData
+	data.BaseType = eventData.BaseType()
+	envelope.Name = eventData.EnvelopeName("")
+	envelope.Data = data
+
+	packer.sanitizeAll(envelope, eventData)
+}
+
+func (packer *logPacker) handleMessageData(envelope *contracts.Envelope, data *contracts.Data, logRecord plog.LogRecord, resource pcommon.Resource, instrumentationScope pcommon.InstrumentationScope) {
 	messageData := contracts.NewMessageData()
 	messageData.Properties = make(map[string]string)
-
 	messageData.SeverityLevel = packer.toAiSeverityLevel(logRecord.SeverityNumber())
-
 	messageData.Message = logRecord.Body().AsString()
 
 	envelope.Tags[contracts.OperationId] = traceutil.TraceIDToHexOrEmptyString(logRecord.TraceID())
 	envelope.Tags[contracts.OperationParentId] = traceutil.SpanIDToHexOrEmptyString(logRecord.SpanID())
 
-	envelope.Name = messageData.EnvelopeName("")
-
 	data.BaseData = messageData
 	data.BaseType = messageData.BaseType()
+	envelope.Name = messageData.EnvelopeName("")
 	envelope.Data = data
 
 	resourceAttributes := resource.Attributes()
@@ -49,9 +68,26 @@ func (packer *logPacker) LogRecordToEnvelope(logRecord plog.LogRecord, resource
 
 	setAttributesAsProperties(logRecord.Attributes(), messageData.Properties)
 
-	packer.sanitize(func() []string { return messageData.Sanitize() })
-	packer.sanitize(func() []string { return envelope.Sanitize() })
+	packer.sanitizeAll(envelope, messageData)
+}
+
+func (packer *logPacker) sanitizeAll(envelope *contracts.Envelope, data any) {
+	if sanitizer, ok := data.(interface{ Sanitize() []string }); ok {
+		packer.sanitize(sanitizer.Sanitize)
+	}
+	packer.sanitize(envelope.Sanitize)
 	packer.sanitize(func() []string { return contracts.SanitizeTags(envelope.Tags) })
+}
+
+func (packer *logPacker) LogRecordToEnvelope(logRecord plog.LogRecord, resource pcommon.Resource, instrumentationScope pcommon.InstrumentationScope) *contracts.Envelope {
+	envelope, data := packer.initEnvelope(logRecord)
+	attributes := logRecord.Attributes()
+
+	if packer.config.CustomEventsEnabled && isEventData(attributes) {
+		packer.handleEventData(envelope, data, logRecord)
+	} else {
+		packer.handleMessageData(envelope, data, logRecord, resource, instrumentationScope)
+	}
 
 	return envelope
 }
@@ -79,9 +115,10 @@ func (packer *logPacker) toAiSeverityLevel(sn plog.SeverityNumber) contracts.Sev
 	}
 }
 
-func newLogPacker(logger *zap.Logger) *logPacker {
+func newLogPacker(logger *zap.Logger, config *Config) *logPacker {
 	packer := &logPacker{
 		logger: logger,
+		config: config,
 	}
 	return packer
 }
@@ -97,3 +134,17 @@ func timestampFromLogRecord(lr plog.LogRecord) pcommon.Timestamp {
 
 	return pcommon.NewTimestampFromTime(timeNow())
 }
+
+func hasOneOfKeys(attrMap pcommon.Map, keys ...string) bool {
+	for _, key := range keys {
+		_, exists := attrMap.Get(key)
+		if exists {
+			return true
+		}
+	}
+	return false
+}
+
+func isEventData(attrMap pcommon.Map) bool {
+	return hasOneOfKeys(attrMap, attributeMicrosoftCustomEventName, attributeApplicationInsightsEventMarkerAttribute)
+}

exporter/azuremonitorexporter/logexporter_test.go

@@ -24,7 +24,8 @@ import (
 
 const (
 	defaultEnvelopeName = "Microsoft.ApplicationInsights.Message"
-	defaultdBaseType    = "MessageData"
+	defaultBaseType     = "MessageData"
+	eventBaseType       = "EventData"
 )
 
 var (
@@ -54,8 +55,9 @@ func TestLogRecordToEnvelope(t *testing.T) {
 	}
 
 	tests := []struct {
-		name  string
-		index int
+		name     string
+		baseType string
+		index    int
 	}{
 		{
 			name:  "timestamp is correct",
@@ -86,7 +88,7 @@ func TestLogRecordToEnvelope(t *testing.T) {
 			assert.Equal(t, toTime(timestampFromLogRecord(logRecord)).Format(time.RFC3339Nano), envelope.Time)
 			require.NotNil(t, envelope.Data)
 			envelopeData := envelope.Data.(*contracts.Data)
-			assert.Equal(t, defaultdBaseType, envelopeData.BaseType)
+			assert.Equal(t, defaultBaseType, envelopeData.BaseType)
 
 			require.NotNil(t, envelopeData.BaseData)
 
@@ -189,7 +191,7 @@ func getLogsExporter(config *Config, transportChannel appinsights.TelemetryChann
 }
 
 func getLogPacker() *logPacker {
-	return newLogPacker(zap.NewNop())
+	return newLogPacker(zap.NewNop(), defaultConfig)
 }
 
 func getTestLogs() plog.Logs {
@@ -261,3 +263,87 @@ func getTestLogRecord(index int) (pcommon.Resource, pcommon.InstrumentationScope
 
 	return resource, scope, logRecord
 }
+
+func TestHandleEventData(t *testing.T) {
+	logger := zap.NewNop()
+	config := &Config{}
+	packer := newLogPacker(logger, config)
+
+	tests := []struct {
+		name              string
+		logRecord         func() plog.LogRecord
+		expectedEventName string
+		expectedProperty  map[string]string
+	}{
+		{
+			name: "Event name from attributeMicrosoftCustomEventName",
+			logRecord: func() plog.LogRecord {
+				lr := plog.NewLogRecord()
+				lr.Attributes().PutStr(attributeMicrosoftCustomEventName, "CustomEvent")
+				lr.Attributes().PutStr("test_attribute", "test_value")
+				return lr
+			},
+			expectedEventName: "CustomEvent",
+			expectedProperty: map[string]string{
+				attributeMicrosoftCustomEventName: "CustomEvent",
+				"test_attribute":                  "test_value",
+			},
+		},
+		{
+			name: "Event name from attributeApplicationInsightsEventMarkerAttribute",
+			logRecord: func() plog.LogRecord {
+				lr := plog.NewLogRecord()
+				lr.Attributes().PutStr(attributeApplicationInsightsEventMarkerAttribute, "MarkerEvent")
+				lr.Attributes().PutStr("test_attribute", "test_value")
+				return lr
+			},
+			expectedEventName: "MarkerEvent",
+			expectedProperty: map[string]string{
+				attributeApplicationInsightsEventMarkerAttribute: "MarkerEvent",
+				"test_attribute": "test_value",
+			},
+		},
+		{
+			name: "No event name attributes",
+			logRecord: func() plog.LogRecord {
+				lr := plog.NewLogRecord()
+				lr.Attributes().PutStr("test_attribute", "test_value")
+				return lr
+			},
+			expectedEventName: "",
+			expectedProperty: map[string]string{
+				"test_attribute": "test_value",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			envelope := contracts.NewEnvelope()
+			data := contracts.NewData()
+			logRecord := tt.logRecord()
+
+			packer.handleEventData(envelope, data, logRecord)
+
+			eventData := data.BaseData.(*contracts.EventData)
+			assert.Equal(t, tt.expectedEventName, eventData.Name)
+			assert.Equal(t, tt.expectedProperty, eventData.Properties)
+		})
+	}
+}
+
+func TestSetAttributesAsProperties(t *testing.T) {
+	properties := make(map[string]string)
+	attributes := pcommon.NewMap()
+	attributes.PutStr("string_key", "string_value")
+	attributes.PutInt("int_key", 123)
+	attributes.PutDouble("double_key", 4.56)
+	attributes.PutBool("bool_key", true)
+
+	setAttributesAsProperties(attributes, properties)
+
+	assert.Equal(t, "string_value", properties["string_key"])
+	assert.Equal(t, "123", properties["int_key"])
+	assert.Equal(t, "4.56", properties["double_key"])
+	assert.Equal(t, "true", properties["bool_key"])
+}