Google Cloud Storage

Synopsis

Creates a collector that subscribes to a Google Cloud Pub/Sub subscription to receive OBJECT_FINALIZE notifications, then downloads and ingests the corresponding objects from Google Cloud Storage.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: gcs
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    project_id: <string>
    subscription_id: <string>
    credentials_json: <string>
    bucket: <string>
    object_size: <numeric>
    file_name_filter: <string>
    max_files_in_archive: <numeric>
    max_size_archive_bytes: <numeric>
    timeout: <numeric>
    mode: <enum>
    prefix: <string>
    workers: <numeric>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
`id`	Y	-	Unique numeric identifier
`name`	Y	-	Device name
`description`	N	-	Optional description
`type`	Y	-	Must be `gcs`
`tags`	N	-	Array of labels for categorization
`pipelines`	N	-	Array of preprocessing pipeline references
`status`	N	`true`	Enable/disable the device

Connection

Field	Required	Default	Description
`project_id`	Y	-	Google Cloud project ID that owns the Pub/Sub subscription
`subscription_id`	Y	-	Pub/Sub subscription ID that receives GCS object notifications
`credentials_json`	N	-	Service account credentials JSON. If omitted, Application Default Credentials are used
`bucket`	N	-	Restrict ingestion to objects from this bucket name. If omitted, all buckets in the subscription are processed. `bucket_name` is accepted as an alias, read only when `bucket` is unset

Object Processing

Field	Required	Default	Description
`object_size`	N	`100000000`	Maximum object size in bytes to download. Objects exceeding this limit are skipped. Set to `0` for no limit
`file_name_filter`	N	`".*"`	Regular expression applied to the object name before download. Objects not matching the pattern are skipped
`mode`	N	`queue`	Ingestion mode: `queue` or `poll`. See Ingestion Modes
`prefix`	N	-	Restrict ingestion to keys under this prefix. In `poll` mode it may carry date tokens
`workers`	N	`max(NumCPU/4, 1)`	Concurrent download and decode workers, 1-64
`timeout`	N	`10`	Config poll interval in seconds. Accepts values between `1` and `10`

Archive Extraction

Field	Required	Default	Description
`max_files_in_archive`	N	`100`	Maximum number of files extracted from a single archive. Set to `0` for no limit
`max_size_archive_bytes`	N	`104857600`	Maximum total extracted size in bytes across all files in an archive. Set to `0` for no limit

Details

Ingestion Modes

This device offers two ingestion modes, selected with mode:

`mode`	Behaviour
`queue`	Default. The device consumes notifications from the bucket's event queue and fetches each object as it is announced.
`poll`	The device lists the bucket on a schedule and ingests what it has not seen before. No queue is needed.

Any other value fails with mode field not valid.

Poll mode suits a bucket with no notification wiring, and a historical backfill, since a listing can be bounded to a date range. It costs a list operation per cycle, so it is the wrong choice for a high-rate bucket that already has a queue.

Poll Mode Properties

These are read only when mode: poll is set.

Field	Required	Default	Description
`interval`	N	`60`	Seconds between list cycles. Ignored when `cron` is set
`cron`	N	-	Cron expression for the list cycle, taking precedence over `interval`
`workers`	N	`max(NumCPU/4, 1)`	Concurrent download and decode workers, 1–64
`max_retries`	N	`0`	Consecutive cycles an object may fail transiently before the poller gives up on it and moves the cursor past it. `0` means never give up
`start_date`	N	-	Lower bound of the listing window. Accepts Unix seconds, RFC 3339, a timestamp without an offset (read in `timezone`), or a relative expression. Unset or negative means the whole bucket
`end_date`	N	-	Upper bound, exclusive. Unset means unbounded; a negative value is an error, and a value at or before `start_date` is rejected
`timezone`	N	`UTC`	Zone used to interpret an offset-less `start_date` or `end_date`

The Cursor

Poll mode keeps a since-cursor so a restart or a cluster failover does not re-ingest what it already has. The cursor is scoped to the listing target — the bucket and prefix together — so changing either starts a fresh one.

A stored cursor takes precedence over start_date, but only while the window is unchanged: editing start_date, end_date or prefix discards the cursor and re-evaluates the new window. A bounded window that has been fully drained stops listing and idles until the window changes.

Date Tokens in `prefix`

In poll mode prefix may carry date tokens, so a listing walks only the partitions that fall inside its window: %Y, %y, %m, %d, %j, %H, and %% for a literal percent. The finest token present sets the partition step.

A token-bearing prefix needs a bounded start_date — there is no way to walk partitions from the beginning of a bucket — and a window spanning too many partitions is rejected rather than expanded.

warning

Date tokens are rejected in queue mode. A queue subscription filters on a fixed key prefix, which cannot be re-expanded per cycle, so a prefix containing %Y and the like fails at apply time. Write %% where a literal percent is wanted.

The device uses two Google Cloud clients concurrently. A Pub/Sub subscriber listens on the configured subscription and receives notifications for each newly written GCS object. Only OBJECT_FINALIZE events are processed; all other event types are acknowledged and discarded. When a matching notification arrives, the GCS client downloads the object content for ingestion.

If bucket is set, notifications referencing a different bucket are acknowledged and skipped without downloading any data. This allows a single Pub/Sub subscription to cover multiple buckets while the device ingests from one specific bucket.

Authentication uses the credentials_json field when provided. If the field is empty, the client falls back to Application Default Credentials, which includes Workload Identity, environment variables, and the gcloud CLI credential chain.

The following file formats are supported directly: .json, .jsonl, .parquet (also .parq, .pq). Files with .log or .txt extensions are auto-detected. Archive formats .gz, .zip, .bz2, .tar, .tgz, .tar.gz, and .tar.bz2 are extracted before ingestion. Files with any other extension are acknowledged and discarded.

Object size is validated before download using the GCS object metadata. If object_size is set to a positive value, any object whose stored size exceeds the limit is skipped. For archives, object_size limits the compressed download size while max_size_archive_bytes limits the total extracted content size.

The file_name_filter regex is applied to the object name (the key path within the bucket) before download. If the pattern does not compile, the device falls back to the default .* pattern and logs a warning. The pattern is re-evaluated on each config poll interval, so it can be changed without restarting the device.

Transient processing errors cause the Pub/Sub message to be nacked so Pub/Sub retries delivery. Permanent errors (unsupported format, size limit exceeded, archive constraint violations, file name filter mismatch) cause the message to be acknowledged to prevent infinite redelivery.

Replay

The gcsreplay device type is this collector permanently in poll mode. It lists the bucket for a bounded time window and ingests the objects it finds, without a Pub/Sub notification subscription.

Every property behaves as documented under Poll Mode above, including start_date, end_date, timezone, the date tokens in prefix, and the cursor. The only configuration difference is type: mode may be omitted, since a replay can only poll, and setting it to queue is rejected.

What the replay type adds is a reported run — live progress, a run history, and the pause, resume, cancel and run-now controls. See Replay Devices.

Examples

Basic

Creating a basic GCS collector with ADC authentication...

- id: 1
  name: gcs_basic
  type: gcs
  properties:
    project_id: "my-gcp-project"
    subscription_id: "gcs-notifications-sub"

Service Account

Authenticating with a service account credentials file...

- id: 2
  name: gcs_service_account
  type: gcs
  properties:
    project_id: "my-gcp-project"
    subscription_id: "gcs-notifications-sub"
    credentials_json: '{"type":"service_account","project_id":"my-gcp-project","private_key_id":"...","private_key":"-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----\n","client_email":"datastream@my-gcp-project.iam.gserviceaccount.com","client_id":"...","auth_uri":"https://accounts.google.com/o/oauth2/auth","token_uri":"https://oauth2.googleapis.com/token"}'

Bucket Filter

Filtering to a single bucket from a shared subscription...

- id: 3
  name: gcs_bucket_filter
  type: gcs
  properties:
    project_id: "my-gcp-project"
    subscription_id: "gcs-all-buckets-sub"
    bucket: "security-logs-bucket"

File Name Filter

Ingesting only JSON log files from a specific prefix...

- id: 4
  name: gcs_filtered
  type: gcs
  properties:
    project_id: "my-gcp-project"
    subscription_id: "gcs-notifications-sub"
    bucket: "app-logs-bucket"
    file_name_filter: "^logs/firewall/.*\\.json$"
    object_size: 52428800

Archive Processing

Configuring archive extraction with size and file count limits...

- id: 5
  name: gcs_archives
  type: gcs
  pipelines:
    - json_parser
  properties:
    project_id: "my-gcp-project"
    subscription_id: "gcs-notifications-sub"
    bucket: "archive-logs-bucket"
    max_files_in_archive: 50
    max_size_archive_bytes: 52428800
    object_size: 10485760

Synopsis​

Schema​

Configuration​

Device​

Connection​

Object Processing​

Archive Extraction​

Details​

Ingestion Modes​

Poll Mode Properties​

The Cursor​

Date Tokens in prefix​

Replay​

Examples​

Basic​

Service Account​

Bucket Filter​

File Name Filter​

Archive Processing​