« GCP Pub/Sub input HTTP Endpoint input »

› › ›

Google Cloud Storage Input

Use the google cloud storage input to read content from files stored in buckets which reside on your Google Cloud. The input can be configured to work with and without polling, though if polling is disabled it will only perform a single collection of data, list the file contents and end the process.

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

When processing google cloud buckets, 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.

Config Option Removal Notice : The bucket_timeout config option has been removed from the google cloud storage input. The intention behind this removal is to simplify the configuration and to make it more user friendly. The bucket_timeout option was confusing and had the potential to let users malconfigure the input, which could lead to unexpected behavior. The input now uses a more robust and efficient way to handle the bucket timeout internally.

Currently only JSON and NDJSON are supported object/file formats. Objects/files may be also be gzip compressed. "JSON credential keys" and "credential files" are supported authentication types. If an array is present as the root object for an object/file, it is automatically split into individual objects and processed. If a download for a file/object fails or gets interrupted, the download is retried for 2 times. This is currently not user configurable.

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

filebeat.inputs:
- type: gcs
  id: my-gcs-id
  enabled: true
  project_id: my_project_id
  auth.credentials_file.path: {{file_path}}/{{creds_file_name}}.json
  parse_json: true
  buckets:
  - name: gcs-test-new
    max_workers: 3
    poll: true
    poll_interval: 15s
  - name: gcs-test-old
    max_workers: 3
    poll: true
    poll_interval: 10s

Explanation : This configuration given above describes a basic gcs config having two buckets named gcs-test-new and gcs-test-old. Each of these buckets 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 google cloud storage input to identify the files it needs to read and process, it will require the bucket names to be specified. We can have as many buckets 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 buckets 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 bucket level with different values, thus offering extensive flexibility and customization. Examples below show this behavior.

On receiving this config the google cloud storage input will connect to the service and retrieve a Storage Client using the given bucket_name and auth.credentials_file, then it will spawn two main go-routines, one for each bucket. 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 bucket, each having 3 workers. These workers will be responsible for 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-09-01T13:54:24.588Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "8.5.0",
    "_id": "gcs-test-new-data_3.json-worker-1"
  },
  "log": {
    "offset": 200,
    "file": {
      "path": "gs://gcs-test-new/data_3.json"
    }
  },
  "input": {
    "type": "gcs"
  },
  "message": "{\n    \"id\": 1,\n    \"title\": \"iPhone 9\",\n    \"description\": \"An apple mobile which is nothing like apple\",\n    \"price\": 549,\n    \"discountPercentage\": 12.96,\n    \"rating\": 4.69,\n    \"stock\": 94,\n    \"brand\": \"Apple\",\n    \"category\": \"smartphones\",\n    \"thumbnail\": \"https://dummyjson.com/image/i/products/1/thumbnail.jpg\",\n    \"images\": [\n        \"https://dummyjson.com/image/i/products/1/1.jpg\",\n        \"https://dummyjson.com/image/i/products/1/2.jpg\",\n        \"https://dummyjson.com/image/i/products/1/3.jpg\",\n        \"https://dummyjson.com/image/i/products/1/4.jpg\",\n        \"https://dummyjson.com/image/i/products/1/thumbnail.jpg\"\n    ]\n}\n",
  "cloud": {
    "provider": "goole cloud"
  },
  "gcs": {
    "storage": {
      "bucket": {
        "name": "gcs-test-new"
      },
      "object": {
        "name": "data_3.json",
        "content_type": "application/json",
        "json_data": [
          {
            "id": 1,
            "discountPercentage": 12.96,
            "rating": 4.69,
            "brand": "Apple",
            "price": 549,
            "category": "smartphones",
            "thumbnail": "https://dummyjson.com/image/i/products/1/thumbnail.jpg",
            "description": "An apple mobile which is nothing like apple",
            "title": "iPhone 9",
            "stock": 94,
            "images": [
              "https://dummyjson.com/image/i/products/1/1.jpg",
              "https://dummyjson.com/image/i/products/1/2.jpg",
              "https://dummyjson.com/image/i/products/1/3.jpg",
              "https://dummyjson.com/image/i/products/1/4.jpg",
              "https://dummyjson.com/image/i/products/1/thumbnail.jpg"
            ]
          }
        ]
      }
    }
  },
  "event": {
    "kind": "publish_data"
  }
}

As we can see from the response above, the message field contains the original stringified data while the gcs.storage.object.data contains the objectified data.

Some of the key attributes are as follows :-

message : Original stringified object data.
log.file.path : Path of the object in google cloud.
gcs.storage.bucket.name : Name of the bucket from which the file has been read.
gcs.storage.object.name : Name of the file/object which has been read.
gcs.storage.object.content_type : Content type of the file/object. You can find the supported content types here .
gcs.storage.object.json_data : Objectified json file data, representing the contents of the file.

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

Supported Attributes :-

Metric	Description
`url`	URL of the input resource.
`errors_total`	Total number of errors encountered by the input.
`decode_errors_total`	Total number of decode errors encountered by the input.
`gcs_objects_requested_total`	Total number of GCS objects downloaded.
`gcs_objects_published_total`	Total number of GCS objects processed that were published.
`gcs_objects_listed_total`	Total number of GCS objects returned by list operations.
`gcs_bytes_processed_total`	Total number of GCS bytes processed.
`gcs_events_created_total`	Total number of events created from processing GCS data.
`gcs_failed_jobs_total`	Total number of failed jobs.
`gcs_expired_failed_jobs_total`	Total number of expired failed jobs that could not be recovered.
`gcs_objects_tracked_gauge`	Number of objects currently tracked in the state registry (gauge).
`gcs_objects_inflight_gauge`	Number of GCS objects inflight (gauge).
`gcs_jobs_scheduled_after_validation`	Histogram of the number of jobs scheduled after validation.
`gcs_object_processing_time`	Histogram of the elapsed GCS object processing times in nanoseconds (start of download to completion of parsing).
`gcs_object_size_in_bytes`	Histogram of processed GCS object size in bytes.
`gcs_events_per_object`	Histogram of event count per GCS object.
`source_lag_time`	Histogram of the time between the source (Updated) timestamp and the time the object was read, in nanoseconds.

Google Cloud Storage Input

Google Cloud Storage Input

project_id

auth.credentials_json.account_key

auth.credentials_file.path

buckets

name

max_workers

poll

poll_interval

parse_json

encoding

decoding

the CSV codec

file_selectors

expand_event_list_from_field

timestamp_epoch

retry

Metrics

Common input options

Common options

enabled

tags

fields

fields_under_root

processors

pipeline

keep_null

index

publisher_pipeline.disable_host

`project_id`

`auth.credentials_json.account_key`

`auth.credentials_file.path`

`buckets`

`name`

`max_workers`

`poll`

`poll_interval`

`parse_json`

`encoding`

`decoding`

`the CSV codec`

`file_selectors`

`expand_event_list_from_field`

`timestamp_epoch`

`retry`

`enabled`

`tags`

`fields`

`fields_under_root`

`processors`

`pipeline`

`keep_null`

`index`

`publisher_pipeline.disable_host`