« Azure eventhub input Benchmark input »

› › ›

Azure Blob Storage Input

Use the azure blob storage input to read content from files stored in containers which reside on your Azure Cloud. The input can be configured to work with and without polling, though currently, if polling is disabled it will only perform a one time passthrough, list the file contents and end the process. Polling is generally recommented for most cases even though it can get expensive with dealing with a very large number of files.

To mitigate errors and ensure a stable processing environment, this input employs the following features :

When processing azure blob containers, if suddenly there is any outage, the process will be able to resume post the last file it processed and was successfully able to save the state for.
If any errors occur for certain files, they will be logged appropriately, but the rest of the files will continue to be processed normally.
If any major error occurs which stops the main thread, the logs will be appropriately generated, describing said error.

NOTE: Currently only JSON and NDJSON are supported blob/file formats. Blobs/files may be also be gzip compressed. As for authentication types, we currently have support for shared access keys and connection strings.

A sample configuration with detailed explanation for each field is given below :-

filebeat.inputs:
- type: azure-blob-storage
  id: my-azureblobstorage-id
  enabled: true
  account_name: some_account
  auth.shared_credentials.account_key: some_key
  containers:
  - name: container_1
    max_workers: 3
    poll: true
    poll_interval: 10s
  - name: container_2
    max_workers: 3
    poll: true
    poll_interval: 10s

Explanation : This configuration given above describes a basic blob storage config having two containers named container_1 and container_2. Each of these containers have their own attributes such as name, max_workers, poll and poll_interval. These attributes have detailed explanations given below. For now lets try to understand how this config works.

For azure blob storage input to identify the files it needs to read and process, it will require the container names to be specified. We can have as many containers as we deem fit. We are also able to configure the attributes max_workers, poll and poll_interval at the root level, which will then be applied to all containers which do not specify any of these attributes explicitly.

If the attributes max_workers, poll and poll_interval are specified at the root level, these can still be overridden at the container level with different values, thus offering extensive flexibility and customization. Examples below show this behaviour.

On receiving this config the azure blob storage input will connect to the service and retrieve a ServiceClient using the given account_name and auth.shared_credentials.account_key, then it will spawn two main go-routines, one for each container. After this each of these routines (threads) will initialize a scheduler which will in turn use the max_workers value to initialize an in-memory worker pool (thread pool) with 3 workers available. Basically that equates to two instances of a worker pool, one per container, each having 3 workers. These workers will be responsible if performing jobs that process a file (in this case read and output the contents of a file).

The scheduler is responsible for scheduling jobs, and uses the maximum available workers in the pool, at each iteration, to decide the number of files to retrieve and process. This keeps work distribution efficient. The scheduler uses poll_interval attribute value to decide how long to wait after each iteration. Each iteration consists of processing a certain number of files, decided by the maximum available workers value.

A Sample Response :-

 {
    "@timestamp": "2022-07-25T07:00:18.544Z",
    "@metadata": {
        "beat": "filebeat",
        "type": "_doc",
        "version": "8.4.0",
        "_id": "beatscontainer-data_3.json-worker-1"
    },
    "message": "{\n    \"id\": 3,\n    \"title\": \"Samsung Universe 9\",\n    \"description\": \"Samsung's new variant which goes beyond Galaxy to the Universe\",\n    \"price\": 1249,\n    \"discountPercentage\": 15.46,\n    \"rating\": 4.09,\n    \"stock\": 36,\n    \"brand\": \"Samsung\",\n    \"category\": \"smartphones\",\n    \"thumbnail\": \"https://dummyjson.com/image/i/products/3/thumbnail.jpg\",\n    \"images\": [\n        \"https://dummyjson.com/image/i/products/3/1.jpg\"\n    ]\n}",
    "cloud": {
        "provider": "azure"
    },
    "input": {
        "type": "azure-blob-storage"
    },
    "log": {
        "offset": 200,
        "file": {
            "path": "https://beatsblobstorage1.blob.core.windows.net/beatscontainer/data_3.json"
        }
    },
    "azure": {
        "storage": {
            "container": {
                "name": "beatscontainer"
            },
            "blob": {
                "content_type": "application/json",
                "name": "data_3.json"
            }
        }
    },
    "event": {
        "kind": "publish_data"
    }
}

As we can see from the response above, the message field contains the original stringified data.

Some of the key attributes are as follows :-

message : Original stringified blob data.
log.file.path : Path of the blob in azure cloud.
azure.storage.blob.container.name : Name of the container from which the file has been read.
azure.storage.blob.object.name : Name of the file/blob which has been read.
azure.storage.blob.object.content_type : Content type of the file/blob. You can find the supported content types here.

Now let’s explore the configuration attributes a bit more elaborately.

Supported Attributes :-

Azure Blob Storage Input

Azure Blob Storage Input

account_name

auth.oauth2

auth.shared_credentials.account_key

auth.connection_string.uri

storage_url

containers

name

max_workers

poll

poll_interval

encoding

decoding

the CSV codec

file_selectors

expand_event_list_from_field

timestamp_epoch

`account_name`

`auth.oauth2`

`auth.shared_credentials.account_key`

`auth.connection_string.uri`

`storage_url`

`containers`

`name`

`max_workers`

`poll`

`poll_interval`

`encoding`

`decoding`

`the CSV codec`

`file_selectors`

`expand_event_list_from_field`

`timestamp_epoch`