Azure Blob via Storage queue
Reads each blob Event Grid announces on a Storage queue, splits it into records, and deletes the message once Smeltr delivers every record.
The source receives a storage account's Event Grid BlobCreated events from a Storage
queue and reads the blobs they name.
Example
Reads each announced blob one line per record, as the worker host's managed identity.
{
"id": "blob-logs",
"type": "azureBlobQueue",
"queueUrl": "https://acmelogs.queue.core.windows.net/blob-events",
"account": "acmelogs",
"framing": "lines"
}
Before you start
- Create a Storage queue in the storage account that holds the blobs.
- Subscribe to the account's events with the event type Blob Created and the queue as the endpoint. The Event Grid schema and CloudEvents 1.0 both work.
- Choose how it signs in, and give that identity permission to read the blobs and to work the
queue. A key, signature or client secret goes in a credential.
- the worker host's managed identity, or a service principal, given Storage Blob Data Reader on the blobs and Storage Queue Data Contributor on the account's queues (see assign a role for data access);
- an account key; or
- an account shared access signature for the blob and queue services, allowed to read blobs, create queues, and read, add, update and process messages.
Settings
It also takes the settings every source takes.
queueUrl string required
The URL of the Storage queue Event Grid sends BlobCreated events to. Its name is at most 56 characters, leaving room for its poison queue's -poison.
https://acmelogs.queue.core.windows.net/blob-eventshttp://127.0.0.1:10001/acmelogs/blob-eventsaccount string required
The storage account holding the blobs. A blob in another account is refused.
acmelogsaccessKey credential default none
The credential holding the account's access key. Set at most one of accessKey, sasToken, clientSecret and managedIdentityClientId; with none, the host's managed or workload identity is used.
AZURE_ACCESS_KEY{"secret": "azure-access-key","version": 1}sasToken credential default none
The credential holding an account shared access signature that covers both the blob and queue services.
AZURE_SAS_TOKEN{"secret": "azure-sas-token","version": 1}tenantId string default none
A service principal's tenant. Set tenantId, clientId and clientSecret together, or none of them.
00000000-0000-0000-0000-000000000000clientId string default none
A service principal's client id.
11111111-1111-1111-1111-111111111111clientSecret credential default none
The credential holding a service principal's client secret.
AZURE_CLIENT_SECRET{"secret": "azure-client-secret","version": 1}managedIdentityClientId string default the host's managed or workload identity
The client id of the user-assigned managed identity to sign in as. Set it when the worker's host has more than one managed identity. Left out, it signs in as the host's managed identity, or an AKS pod's workload identity.
22222222-2222-2222-2222-222222222222format string default each payload is text, in _raw
How each payload becomes the record.
avro- Reads Avro files. Fields you don't declare are kept.
fields list of fields default none
The Avro fields to read, each with a name and a type such as i64 or str. Name a nested field with dots, such as user.id. Top-level fields you don't list are kept. Only with format avro.
[{"name": "user_id","type": "i64"},{"name": "user.name","type": "str"}]framing string default lines
How an object becomes records.
lines- Gives one record per line, or per piece
lineBreakercuts. object- Gives the whole object as one record, for a JSON document such as CloudTrail's that a pipeline unrolls.
escapedLines- Gives one record per line, unescaped as a
bytesobject destination wrote it.
lineBreaker string default a record per line
With lines framing, and only then: where a record ends. Its first capture group is the break and is dropped: the bytes before it end a record, line ends and all, and the bytes after it start the next. What follows the group is kept, so the pattern can say how a record starts: ([\r\n]+)\d{4}- keeps a stack trace with the line that logged it, and </event>(\s*) cuts XML events with no line ends between them. It needs a capture group and must not match the empty string.
([\r\n]+)\d{4}-</event>(\s*)identity identity default none
The connection identity it connects under, for a private CA or a client certificate.
azure-proxycorp-proxyAdvanced
endpoint string default the account's, in Azure's public cloud
The blob service's address, such as Azurite's or a sovereign cloud's.
http://localhost:10000/acmelogshttps://acmelogs.blob.core.usgovcloudapi.netauthEndpoint string default its cloud's own
Where a client secret is exchanged for a token. Only with clientSecret.
https://login.microsoftonline.ushttps://login.chinacloudapi.cnencoding string default auto
How a record's bytes become _raw's text: auto, none, or a WHATWG encoding label other than the replacement ones, such as utf-8, utf-16le or shift_jis.
auto- Detects each record's encoding.
utf-8- Decodes every record as UTF-8, as any other label decodes as its own encoding. Bytes that do not decode become U+FFFD.
none- Decodes nothing: the record has no
_raw, and its payload is__source.bytesalone. A pipeline reads it as text withdecode(__source.bytes, encoding).
visibilitySecs integer default 60
Seconds a received message stays hidden from other readers. The source extends it while it reads the message's blobs. 1 to 43200.
3003043200maxDeliveries integer default 5
Times a message is received before it is moved to the queue named after this one with -poison added. 1 to 1000.
103100Fields it gives
__source.bucket holds the blob's container, and __source.key its name.
| Field | Type | Holds |
|---|---|---|
__source.name | str | The name of the source that read the record. |
__source.type | str | The kind of source that read it, its type, such as kafka, s3List, httpApi or hecReceiver. |
__source.bucket | str | The bucket, or Azure container, the record's object is in. |
__source.key | str | The key, or Azure blob name, of the object the record was read from. |
__source.offset | i64 | The record's offset in its Kafka or Event Hubs partition, or its record number in its object, from 0. |
__source.bytes | bytes | The record's payload as the source read it, before any decoding. The source decodes _raw from it. A bytes destination writes it unless the destination names another field. Text functions refuse it; decode(__source.bytes, encoding) reads it as text. |
How it works
A message may hold one event or an array of them, as plain text or base64. The source reads the
blob a BlobCreated event names, at its latest version. For a Data Lake file it
skips CreateFile and reads the file at FlushWithClose. The source
deletes a message that announces no blob without reading anything. A message that is not an
Event Grid event, or names a blob in another account, stays on the queue, like one whose blob
the source cannot read.
The source decompresses a gzip, zstd, framed snappy or framed lz4 blob first and reads any other as it is.
The source keeps a message hidden while it reads the message's blobs and Smeltr delivers them,
and deletes it once Smeltr commits every record of every blob it names. It never deletes a
message whose blob it cannot read, and moves a message received more than maxDeliveries times to the queue named after this one with -poison added, making that queue if it is
not there. A message on the poison queue stays until someone deletes it.
Verify
Upload a blob with a line you can recognize to the account, then find that line at the destination. The queue's message count returns to zero once Smeltr delivers it.
Troubleshooting
- Nothing arrives
- Check that the event subscription filters for Blob Created, sends to this queue, shows delivered events in its metrics, and has no subject filter that leaves out your blobs.
- Messages pile up on the poison queue
- The source cannot read their blobs: check the identity's role on the container, or that the
blob's account is the source's
account, and the worker log for the reason.