azureeventhubreceiver

README

Azure Event Hub Receiver

Status
Stability	beta: metrics, logs, traces
Distributions	contrib
Issues
Code coverage
Code Owners	@atoulme, @cparkins, @dyl10s | Seeking more code owners!
Emeritus	@djaglowski

Overview

Azure resources and services can be configured to send their telemetry to an Azure Event Hub. The Azure Event Hub receiver pulls telemetry from an Azure Event Hub, transforms them, and pushes them through the collector pipeline.

Read further

• Diagnostic settings in Azure Monitor
• Stream Azure monitoring data to an event hub and external partner

Configuration

Connection Settings

Either connection or auth must be specified.

Field	Type	Required	Default	Description
connection	string	Yes*		Connection string from the Azure portal. Ignored if auth is set.
auth	string	Yes*		ID of an authentication extension (AAD, managed identity, or service principal).
event_hub.name	string	**		Event Hub name. Required when using auth.
event_hub.namespace	string	**		Fully qualified namespace (e.g., namespace.servicebus.windows.net). Required when using auth.

* One of connection or auth is required.
** Required when using auth.

Consumer Settings

Field	Type	Required	Default	Description
group	string	No	$Default	Consumer group to read from.
partition	string	No	""	Specific partition to watch. If empty, watches all partitions.
offset	string	No	""	Starting offset. Only applicable when partition is set.
storage	string	No		ID of a storage extension for checkpoint persistence.

Distributed Consumption Settings

When blob_checkpoint_store is set, the receiver uses the Azure SDK Processor to coordinate partition ownership across multiple collector instances via Azure Blob Storage. This is mutually exclusive with partition, offset, and storage.

Field	Type	Required	Default	Description
blob_checkpoint_store.connection	string	*		Azure Blob Storage connection string. Required when not using auth.
blob_checkpoint_store.storage_account_url	string	*		Blob service URL (e.g., https://myaccount.blob.core.windows.net). Required when using auth.
blob_checkpoint_store.container_name	string	Yes		Blob container for checkpoint data. Must already exist before starting the collector.

* One of connection or storage_account_url is required.

Polling Settings

Field	Type	Required	Default	Description
max_poll_events	int	No	100	Maximum events to retrieve per poll.
poll_rate	int	No	5	Maximum seconds to wait before returning fewer than max_poll_events.
prefetch_count	int	No	0	Size of the SDK's per-partition prefetch buffer. The SDK fills this buffer asynchronously so ReceiveEvents calls return from a local cache instead of waiting on the network. 0 uses the SDK default (300). A negative value disables prefetch. Increase this when poll batches are draining the buffer faster than the SDK refills it.

Data Transformation Settings

Field	Type	Required	Default	Description
format	string	No	azure	Message format: azure, raw, or "". Mutually exclusive with encoding. See Format section.
encoding	string	No		ID of an encoding extension used to unmarshal the message body. Mutually exclusive with format. See Encoding section.
apply_semantic_conventions	bool	No	false	Translate Azure Resource Logs using OpenTelemetry semantic convention attribute names.
time_formats.logs	[]string	No		Custom time formats for logs. Uses Go time layout. Falls back to ISO8601.
time_formats.metrics	[]string	No		Custom time formats for metrics.
time_formats.traces	[]string	No		Custom time formats for traces.
metric_aggregation	string	No		Set to average to aggregate datapoints as sum/count. By default, creates separate metrics with suffixes (_TOTAL, _MIN, etc.).

Example Configuration

The simplest configuration uses a connection string from the Azure portal:

receivers:
  azure_event_hub:
    connection: Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=<key-name>;SharedAccessKey=<key>;EntityPath=<hub-name>
    group: $Default  # optional, defaults to $Default

The full list of settings exposed for this receiver are documented above with detailed sample configurations in testdata/config.yaml.

Advanced Configuration

Azure Auth Extension with Service Principal

receivers:
  azure_event_hub:
    event_hub:
      name: hubName
      namespace: namespace.servicebus.windows.net
    auth: azure_auth

extensions:
  azure_auth:
    service_principal:
      client_id: ${env:AZURE_CLIENT_ID}
      client_secret: ${env:AZURE_CLIENT_SECRET}
      tenant_id: ${env:AZURE_TENANT_ID}

Checkpoint Persistence with Storage Extension

This component can persist its state (checkpoint offsets) using the storage extension. This is strongly recommended for production deployments to prevent message reprocessing after collector restarts.

Without a storage extension configured, the receiver will start from the latest offset on each restart, potentially missing messages that arrived while the collector was down.

receivers:
  azure_event_hub:
    connection: Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=superSecret1234=;EntityPath=hubName
    storage: file_storage

extensions:
  file_storage:
    directory: /var/lib/otelcol/eventhub

Distributed Consumption with Blob Checkpoint Store

When blob_checkpoint_store is configured, the receiver automatically:

• Coordinates partition ownership across collector instances via blob leases
• Checkpoints progress to Azure Blob Storage
• Rebalances partitions when instances are added or removed

Prerequisites:

• The blob container must already exist before starting the collector
• All collector instances must use the same consumer group and container

# With connection strings
receivers:
  azure_event_hub:
    connection: Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=superSecret1234=;EntityPath=hubName
    blob_checkpoint_store:
      connection: DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey=mykey;EndpointSuffix=core.windows.net
      container_name: eventhub-checkpoints

# With auth extension
receivers:
  azure_event_hub:
    event_hub:
      name: hubName
      namespace: namespace.servicebus.windows.net
    auth: azureauth
    blob_checkpoint_store:
      storage_account_url: https://myaccount.blob.core.windows.net
      container_name: eventhub-checkpoints

Custom Time Formats and Partitioning

receivers:
  azure_event_hub:
    connection: Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=superSecret1234=;EntityPath=hubName
    partition: "0"
    group: my-consumer-group
    offset: "1234-5566"
    format: "azure"
    time_formats:
      logs: ["01/02/2006 15:04:05", "2006-01-02 15:04:05"]
      metrics: ["01/02/2006 15:04:05"]

Known Limitations

Before using this receiver, be aware of the following limitations:

Checkpoint Persistence

• Without a configured storage extension, checkpoint state is stored only in memory. If the collector restarts, the receiver will resume from the latest offset, potentially missing messages or reprocessing already-consumed messages.

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.

[!WARNING] This format is only supported for Logs. Using raw format with Metrics or Traces pipelines will result in an error.

azure

Logs

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 and the OpenTelemetry attributes.

Azure	OpenTelemetry
callerIpAddress (optional)	network.peer.address (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 or timeStamp (required)	time_unix_nano (time takes precedence)
identity (optional)	azure.identity (attribute, nested)

Notes:

• JSON does not distinguish between fixed and floating point numbers. All JSON numbers are encoded as doubles.

Metrics

For Metrics the Azure Metric Records are an array of "records" with the following fields, by type of metric.

From this data a Metric of type Gauge is created with a Data Points that represents the values for the Metric including: Total, Minimum, Maximum, Average and Count.

Platform metric (from Azure resources)

Azure	Open Telemetry
time	time_unix_nano (field)
resourceId	azure.resource.id (resource attribute)
metricName
timeGrain	start_time_unix_nano (field)
total	mapped to datapoint metricName + "_TOTAL"
count	mapped to datapoint metricName + "_COUNT"
minimum	mapped to datapoint metricName + "_MINIMUM"
maximum	mapped to datapoint metricName + "_MAXIMUM"
average	mapped to datapoint metricName + "_AVERAGE"

Application metrics (from Application Insights)

See: https://learn.microsoft.com/en-us/azure/azure-monitor/reference/tables/appmetrics

Azure	Open Telemetry
time	time_unix_nano (field)
resourceId	azure.resource.id (resource attribute)
Name	(metric name)
AppRoleInstance	service.instance.id (resource attribute)
AppRoleName	service.name (resource attribute)
AppVersion	service.version (resource attribute)
SDKVersion	telemetry.sdk.version (resource attribute)
ClientCountryOrRegion	cloud.region (resource attribute)
ClientOS	os.name (resource attribute)
Properties (key/value map)	mapped to resource attributes
Sum	mapped to datapoint metricName + "_TOTAL"
ItemCount	mapped to datapoint metricName + "_COUNT"
Min	mapped to datapoint metricName + "_MINIMUM"
Max	mapped to datapoint metricName + "_MAXIMUM"

Traces

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

Encoding

As an alternative to the built-in format, the encoding option delegates unmarshaling of the message body to an encoding extension. This is mutually exclusive with format.

extensions:
  azure_encoding:

receivers:
  azure_event_hub:
    connection: Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=<key-name>;SharedAccessKey=<key>;EntityPath=<hub-name>
    encoding: azure_encoding

[!NOTE] The encoding extension only receives the message body. AMQP properties and enqueued time (which the raw format maps onto the LogRecord) are not applied on the encoding path. Use format: raw if you need those.

Documentation

Overview

Package azureeventhubreceiver listens to logs emitted by Azure Event hubs.

Index

• func NewFactory() receiver.Factory
• type BlobCheckpointStoreConfig
• type Config
  • func (config *Config) Validate() error
• type EventHubConfig
• type TimeFormat

Functions

func NewFactory

func NewFactory() receiver.Factory

NewFactory creates a factory for the Azure Event Hub receiver.

Types

type BlobCheckpointStoreConfig

type BlobCheckpointStoreConfig struct {
	// Connection is the connection string for Azure Blob Storage.
	// Required when the parent config does not use auth.
	Connection string `mapstructure:"connection"`
	// StorageAccountURL is the blob service URL (e.g., https://myaccount.blob.core.windows.net).
	// Required when the parent config uses auth.
	StorageAccountURL string `mapstructure:"storage_account_url"`
	// ContainerName is the blob container used for checkpoint data. Required.
	ContainerName string `mapstructure:"container_name"`
}

BlobCheckpointStoreConfig defines the configuration for Azure Blob Storage based checkpoint coordination used in distributed consumption mode.

type Config

type Config struct {
	Connection               string         `mapstructure:"connection"`
	EventHub                 EventHubConfig `mapstructure:"event_hub"`
	Partition                string         `mapstructure:"partition"`
	Offset                   string         `mapstructure:"offset"`
	StorageID                *component.ID  `mapstructure:"storage"`
	Auth                     *component.ID  `mapstructure:"auth"`
	Encoding                 *component.ID  `mapstructure:"encoding"`
	Format                   string         `mapstructure:"format"`
	ConsumerGroup            string         `mapstructure:"group"`
	ApplySemanticConventions bool           `mapstructure:"apply_semantic_conventions"`
	TimeFormats              TimeFormat     `mapstructure:"time_formats"`
	MetricAggregation        string         `mapstructure:"metric_aggregation"`

	// BlobCheckpointStore enables distributed consumption using Azure Blob Storage
	// for checkpoint coordination. When configured, the receiver uses the Azure SDK
	// Processor for dynamic partition assignment across multiple collector instances.
	BlobCheckpointStore *BlobCheckpointStoreConfig `mapstructure:"blob_checkpoint_store"`

	// azeventhub lib specific
	PollRate      int `mapstructure:"poll_rate"`
	MaxPollEvents int `mapstructure:"max_poll_events"`

	// PrefetchCount controls the size of the SDK's internal prefetch buffer per
	// partition. The SDK uses this value to maintain an asynchronous cache of
	// events so that ReceiveEvents calls return from a local buffer rather than
	// waiting on the network.
	//   0  - use the SDK default (300)
	//  <0  - disable prefetch
	//  >0  - use the explicit value
	PrefetchCount int32 `mapstructure:"prefetch_count"`
}

func (*Config) Validate

func (config *Config) Validate() error

Validate config

type EventHubConfig

type EventHubConfig struct {
	// Name is the name of the Event Hub.
	Name string `mapstructure:"name"`
	// Namespace is the fully qualified namespace of the Event Hub.
	Namespace string `mapstructure:"namespace"`
}

EventHubConfig defines the configuration for an Azure Event Hub when using authentication.

type TimeFormat

type TimeFormat struct {
	Logs    []string `mapstructure:"logs"`
	Metrics []string `mapstructure:"metrics"`
	Traces  []string `mapstructure:"traces"`
}