Event Hubs

Microsoft Azure

Synopsis

Creates a collector that connects to Azure Event Hubs and consumes messages from specified event hubs. Supports multiple authentication methods, TLS encryption, and multiple workers for high-throughput scenarios.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: eventhubs
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    event_hub: <string>
    client_connection_string: <string>
    tenant_id: <string>
    client_id: <string>
    client_secret: <string>
    namespace: <string>
    consumer_group: <string>
    container_connection_string: <string>
    container_url: <string>
    container_name: <string>
    timeout: <numeric>
    batch_size: <numeric>
    reuse: <boolean>
    workers: <numeric>
    tls:
      status: <boolean>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      ca_name: <string>
      server_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>
      insecure_skip_verify: <boolean>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
`id`	Y		Unique identifier
`name`	Y		Device name
`description`	N	-	Optional description
`type`	Y		Must be `eventhubs`
`tags`	N	-	Optional tags
`pipelines`	N	-	Optional pre-processor pipelines
`status`	N	`true`	Enable/disable the device

Connection

Event Hubs supports three authentication methods:

Method 1: Connection String Authentication

Field	Required	Default	Description
`client_connection_string`	Y*		Event Hubs connection string (required for connection string authentication)
`event_hub`	Y		Event hub name to consume from

Method 2: Service Principal Authentication

Field	Required	Description
`tenant_id`	Y*	Azure tenant ID (required for service principal authentication)
`client_id`	Y*	Azure service principal client ID
`client_secret`	Y*	Azure service principal client secret
`namespace`	Y*	Event Hubs namespace (required if not using connection string)
`event_hub`	Y	Event hub name to consume from

Method 3: Managed Identity Authentication

With Managed identity switched on in the console, the director signs in with the Azure default credential, which is the managed identity of the Azure resource it runs on. Leave tenant_id, client_id, client_secret and client_connection_string unset.

Field	Required	Description
`namespace`	Y*	Fully qualified Event Hubs namespace. The director needs it whenever there is no connection string
`event_hub`	Y	Event hub name to consume from
`container_url`	Y*	Checkpoint storage container URL (see Storage Configuration)

The identity needs the same Azure RBAC roles as a service principal (see IAM Permissions).

Consumer Configuration

Field	Required	Default	Description
`consumer_group`	N	`"$Default"`	Consumer group name

Storage Configuration

EventHubs requires checkpoint storage, reached with the same sign-in as the event hub: Method 1 with connection string authentication, Method 2 with service principal or managed identity authentication.

Sign-in	Checkpoint storage fields
Connection string (`client_connection_string`)	`container_connection_string` and `container_name`
Service principal (`tenant_id`, `client_id`, `client_secret`, `namespace`)	`container_url`
Managed identity (`namespace`, no client credentials)	`container_url`

The director takes the connection string path only when both client_connection_string and container_connection_string are set. Any other combination builds the checkpoint container client from container_url with the device's Azure credential, so a mixed pair fails when the device starts. A service principal or managed identity with a container connection string has no container URL to reach. A connection string with a container URL has no namespace to consume from.

In the console, the Checkpoint storage group on the Advanced configuration step shows Container URL for a managed identity or a service principal, and Container name with Connection string for a connection string. A device saved earlier with a mixed pair opens asking for the missing field, and saving it removes the stored fields of the other method.

Method 1: Storage Account Connection String

Field	Required	Default	Description
`container_connection_string`	Y*		Azure Storage connection string
`container_name`	Y*		Blob container name for checkpoints

Method 2: Storage Account URL

Field	Required	Default	Description
`container_url`	Y*		Azure Storage container URL

* = Conditionally required (see authentication and storage methods above). The storage method follows the sign-in and is not a separate choice.

Performance

Field	Required	Default	Description
`timeout`	N	`15`	Seconds a single receive may wait for events before the cycle ends and a new one begins
`batch_size`	N	`10000`	Maximum events requested per receive
`reuse`	N	`true`	Caps `workers` at the number of CPU cores the host offers
`workers`	N	`4`	Number of consumers to start, whether `reuse` is on or off. With `reuse` on, fewer start when the host has fewer CPU cores. `1` runs a single consumer
`max_buffer_size`	N	`128MB`	Ingest buffer admission cap. Also sizes this listener's ingest buffers: 3/8 of this value at start, growing to at most 3x

note

max_buffer_size is the ingest-buffer admission cap, and is distinct from any buffer_size field above, which sizes the network read buffer. Left unset it is 128MB, whose buffers take 48 MiB of heap per listener worker at start and grow to at most 384 MiB as bursts need the room — and with reuse: true a device runs one worker per listener. Size strings are binary, and 32MB and 32MiB are exact synonyms.

See Listener Memory Sizing for the arithmetic, the sizing table, and what happens when the cap is exhausted.

TLS

Field	Required	Default	Description
`tls.status`	N	`false`	Enable TLS encryption
`tls.cert_name`	Y*	`cert.pem`	Client certificate for mutual TLS
`tls.key_name`	Y*	`key.pem`	Client private key for mutual TLS
`tls.passphrase`	N	-	Passphrase for an encrypted private key
`tls.ca_name`	N	-	CA bundle used to verify the server certificate. When unset, the host trust store is used; when set, it replaces the host trust store rather than adding to it.
`tls.server_name`	N	-	SNI hostname override for the TLS handshake
`tls.min_tls_version`	N	`tls1.2`	Minimum accepted TLS version (`tls1.0`, `tls1.1`, `tls1.2`, `tls1.3`)
`tls.max_tls_version`	N	-	Maximum accepted TLS version. When unset, the highest mutually supported version is negotiated.
`tls.insecure_skip_verify`	N	`false`	Skip server certificate verification. Use only for testing.

* = Required when tls.status is true.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
Inline PEM content — used verbatim when the value contains -----BEGIN.
Environment variable — ${ENV_VAR}.
Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

Details

IAM Permissions

When using service principal or managed identity authentication, the following Azure RBAC roles are required:

Azure Role	Scope	Purpose
`Azure Event Hubs Data Receiver`	Event Hubs Namespace or Event Hub	Consume events and read hub properties
`Storage Blob Data Contributor`	Storage Account or Container	Read, write, and list checkpoint blobs

The checkpoint storage requires Contributor (not just Reader) because the device writes checkpoint state and manages ownership blobs for partition load balancing.

When using connection string authentication, Azure RBAC roles are not needed for Event Hubs access. The Shared Access Policy embedded in the connection string governs access (typically Listen claim for consumers). Checkpoint storage is then reached with the storage account connection string in container_connection_string.

Multiple Workers

The collector starts workers workers whether reuse is on or off. Each worker maintains its own Event Hubs consumer and processes messages independently, automatically balancing message volumes. reuse only bounds the count: on, it caps workers at the host's CPU cores, so fewer consumers can run than workers names. Set workers: 1 to run a single consumer.

In the console, a new device starts with Multi-worker mode enabled and Workers (Optional) at 4, the same settings a device that sets neither key runs with. A saved device without reuse opens with Multi-worker mode enabled, as Director runs it, and one saved with reuse: false keeps that value.

Messages

The collector supports automatic checkpoint management, consumer group load balancing, multiple Event Hub subscriptions, TLS-encrypted connections, connection string, service principal and managed identity authentication, and custom message-processing pipelines.

Examples

Basic with Connection String

Creating a simple EventHubs consumer with connection string...

- id: 1
  name: basic_eventhubs
  type: eventhubs
  properties:
    client_connection_string: "Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=mykey;SharedAccessKey=myvalue"
    event_hub: "logs"
    container_connection_string: "DefaultEndpointsProtocol=https;AccountName=mystorage;AccountKey=mykey"
    container_name: "checkpoints"

Service Principal Authentication

Connecting with service principal authentication...

- id: 2
  name: sp_eventhubs
  type: eventhubs
  properties:
    tenant_id: "12345678-1234-1234-1234-123456789012"
    client_id: "87654321-4321-4321-4321-210987654321"
    client_secret: "${AZURE_CLIENT_SECRET}"
    namespace: "mynamespace"
    event_hub: "security-logs"
    consumer_group: "datastream-group"
    container_url: "https://mystorage.blob.core.windows.net/checkpoints"

High-Volume Processing

Optimizing for throughput with multiple workers...

- id: 3
  name: performant_eventhubs
  type: eventhubs
  properties:
    client_connection_string: "${EVENTHUBS_CONNECTION_STRING}"
    event_hub: "high-volume-logs"
    consumer_group: "processing-group"
    container_connection_string: "${STORAGE_CONNECTION_STRING}"
    container_name: "checkpoints"
    reuse: true
    workers: 8

Secure Connection

Secure EventHubs connection with TLS...

- id: 4
  name: secure_eventhubs
  type: eventhubs
  properties:
    tenant_id: "${AZURE_TENANT_ID}"
    client_id: "${AZURE_CLIENT_ID}"
    client_secret: "${AZURE_CLIENT_SECRET}"
    namespace: "secure-namespace"
    event_hub: "secure-logs"
    consumer_group: "secure-group"
    container_url: "${STORAGE_CONTAINER_URL}"
    tls:
      status: true
      cert_name: "eventhubs.crt"
      key_name: "eventhubs.key"

Pipeline Processing

Applying custom processing to EventHubs messages...

- id: 5
  name: pipeline_eventhubs
  type: eventhubs
  pipelines:
    - json_parser
    - field_extractor
    - normalize_timestamps
  properties:
    client_connection_string: "${EVENTHUBS_CONNECTION_STRING}"
    event_hub: "application-logs"
    consumer_group: "processing-group"
    container_connection_string: "${STORAGE_CONNECTION_STRING}"
    container_name: "checkpoints"

Multiple Consumer Groups

Configuring consumer groups for load distribution...

- id: 6
  name: distributed_eventhubs
  type: eventhubs
  properties:
    client_connection_string: "${EVENTHUBS_CONNECTION_STRING}"
    event_hub: "distributed-logs"
    consumer_group: "instance-1"
    container_connection_string: "${STORAGE_CONNECTION_STRING}"
    container_name: "checkpoints-instance1"
    reuse: true
    workers: 4

Synopsis​

Schema​

Configuration​

Device​

Connection​

Consumer Configuration​

Storage Configuration​

Performance​

TLS​

Details​

IAM Permissions​

Multiple Workers​

Messages​

Examples​

Basic with Connection String​

Service Principal Authentication​

High-Volume Processing​

Secure Connection​

Pipeline Processing​

Multiple Consumer Groups​