faucet-source-gcs

Google Cloud Storage source connector for the faucet-stream ecosystem. Lists objects in a bucket (with an optional prefix) or reads an explicit list of object keys, fetches each one over the official google-cloud-storage SDK, and parses it as JSON Lines, JSON Array, or raw text — yielding records as serde_json::Value.

Reach for it when your data already lives in GCS — event exports, log dumps, analytics extracts, daily snapshots — and you want to move it into any faucet-stream sink (a database, a warehouse, a queue, a file) with one declarative config and no glue code. Objects are read concurrently and JSONL/raw-text bodies stream line-by-line, so a multi-gigabyte export lands without buffering the whole bucket in memory.

Feature highlights

• Three file formats — json_lines (one record per line), json_array (one array per object), and raw_text (each object becomes a {key, content} record).
• List or explicit keys — scan a bucket by prefix, or skip listing entirely by passing an exact object_keys list.
• Concurrent reads — objects are fetched in parallel via buffer_unordered(concurrency) (default 10), so wall-clock time is bounded by your slowest objects, not their sum.
• Bounded-memory streaming — json_lines and raw_text decode straight off the GCS body reader, so peak memory is O(batch_size) regardless of total file size.
• Compression auto-detect — behind the compression feature, .gz / .zst objects are transparently decompressed; the codec resolves per object key, so one run can mix compressed and uncompressed objects.
• Read-integrity verification — every object's byte length is checked against the size GCS reports (verify_length, default on), so a cleanly-truncated transfer is rejected instead of silently parsed as a complete object. Opt into CRC-32C / MD5 checksum verification with verify_checksum. Both checks auto-skip GCS-transcoded (Content-Encoding: gzip) objects.
• Four credential modes — Application Default Credentials, a service-account key file, inline service-account JSON, or anonymous (for emulators). The shared GcsCredentials enum is re-exported from faucet-common-gcs so it matches the GCS sink byte-for-byte.
• Clients built once — the data-plane and control-plane clients are constructed in new() and reused for every list and read.
• Matrix-aware prefixes — ${parent.path} placeholders in prefix are resolved per-record at runtime, so a parent matrix row can fan out into many per-record bucket scans.

Installation

# As a library:
cargo add faucet-source-gcs

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-gcs

# With transparent gzip/zstd decompression:
cargo install faucet-cli --features "source-gcs,compression"

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: gcs
    config:
      bucket: my-bucket
      prefix: events/2026/
      auth:
        type: service_account_json_file
        config:
          path: /run/secrets/gcp-sa.json
      file_format: json_lines
  sink:
    type: jsonl
    config:
      path: ./events.jsonl

faucet run pipeline.yaml

Configuration reference

Core

Performance

Format & testing

Authentication

auth uses the shared GcsCredentials enum from faucet-common-gcs (the project-wide { type, config } shape):

# Application Default Credentials (workload identity, gcloud, GOOGLE_APPLICATION_CREDENTIALS)
auth:
  type: application_default

# Service-account key file
auth:
  type: service_account_json_file
  config:
    path: /run/secrets/gcp-sa.json

# Inline service-account JSON via env indirection
auth:
  type: service_account_json_inline
  config:
    json: ${env:GCP_SA_JSON}

HMAC-key auth, signed-URL generation, and KMS/CMEK encryption configuration are out of scope.

File formats

Parse errors are precise: a JSONL failure carries the object key and the 1-based line number; a JSON-array failure carries the key. A json_array object whose top-level value isn't an array fails with an "expected JSON array" message. Non-UTF-8 bodies surface as FaucetError::Source with a "not valid UTF-8" hint.

Examples

Scan a prefix as JSON Lines (the 80% path)

source:
  type: gcs
  config:
    bucket: analytics-exports
    prefix: events/dt=2026-06-16/
    auth: { type: application_default }
    file_format: json_lines
    concurrency: 20
    batch_size: 5000

Read an explicit set of objects, no listing

source:
  type: gcs
  config:
    bucket: my-bucket
    object_keys:
      - lookups/countries.json
      - lookups/currencies.json
    auth:
      type: service_account_json_file
      config: { path: /run/secrets/gcp-sa.json }
    file_format: json_array
    batch_size: 0      # emit one page per object — ideal for small lookup tables

Raw text files, compressed, with a cap

source:
  type: gcs
  config:
    bucket: log-archive
    prefix: app/2026/06/
    auth: { type: application_default }
    file_format: raw_text
    compression: auto    # requires the `compression` feature; decompresses .gz / .zst objects
    max_objects: 100     # read at most the first 100 objects
    concurrency: 4       # raw_text holds a whole object in memory — keep concurrency low

Date-templated prefix driven by the run clock

source:
  type: gcs
  config:
    bucket: analytics-exports
    prefix: events/dt=${now.date}/   # resolves to e.g. events/dt=2026-06-16/ at run time
    auth: { type: application_default }
    file_format: json_lines

Streaming & batching

The source overrides Source::stream_pages. It lists object keys once, then walks them in order:

• json_lines decodes the (optionally decompressed) body line-by-line off an async buffered reader, emitting a StreamPage every batch_size records. Memory stays at O(batch_size) no matter how large the file is.
• raw_text emits one {key, content} record per object, streamed straight into a single String (no separate raw + decompressed copies for compressed objects).
• json_array buffers each object fully (a JSON array isn't parseable until its closing ]), then chunks its records into pages.

For a non-zero batch_size, records from multiple objects can share a page (cross-object flattening) — the page boundary follows the record count, not the object boundary. This matches the faucet-source-s3 source and is intentional.

batch_size: 0 is the no-batching sentinel: every page contains exactly one complete object's records, with no within-object chunking and no cross-object accumulation. Useful for small lookup tables or when a downstream sink prefers one large request per object.

Memory ceiling — raw_text / json_array. Both hold one whole decoded object in memory at a time (inherent: a raw-text record is the whole file, and a JSON array isn't valid until its closing ]). Because objects are fetched concurrently, peak memory is bounded by roughly concurrency × (largest object's decoded size), not by batch_size. For large raw_text / json_array objects, lower concurrency to cap peak memory, or re-emit the data as json_lines upstream so it streams at O(batch_size).

This is a one-shot scan source — it has no incremental bookmark / resume support, so each run re-lists and re-reads the matching objects. For incremental loads, advance the prefix between runs (e.g. a dated events/dt=${now.date}/ prefix) so each run reads only fresh objects.

Compression

Behind the crate-local compression Cargo feature. Adds the compression config field with values none, gzip, zstd, or auto (the default). auto detects .gz / .zst from the object key; explicit codecs apply to every object.

source:
  type: gcs
  config:
    bucket: log-archive
    prefix: app/
    auth: { type: application_default }
    compression: auto     # or 'gzip' | 'zstd' | 'none'

The codec resolves per object key, so a single source can read a mix of compressed and uncompressed objects in one run. A one-shot warning fires when an explicit codec disagrees with the object's filename suffix.

Read-integrity verification

Object bodies are read through a verifying reader that validates the transfer at EOF, so a stream that ends early but cleanly is rejected instead of being parsed and emitted as a complete object — silent data loss otherwise.

• Length (verify_length, default true) — counts the bytes read and compares them against the size GCS reports for the object. A mismatch fails the page with FaucetError::Source.
• Checksum (verify_checksum, default false) — verifies the body against the CRC-32C (preferred) or MD5 digest GCS reports. Costs a hash over the full body. A one-shot warning is logged if the object advertises neither.

Both checks are automatically skipped for a transcoded object — one stored with a non-empty Content-Encoding (e.g. gzip), which GCS may decompress on read so the received bytes match neither the stored size nor the stored checksum. They operate on the bytes GCS delivers (below the client-side compression feature's decompression), so .gz / .zst objects served without a Content-Encoding header verify correctly.

pipeline:
  source:
    type: gcs
    config:
      bucket: my-bucket
      prefix: events/
      verify_checksum: true   # length check is already on by default

Config loading & schema

Config loads from YAML/JSON or environment. Inspect the full JSON Schema with:

faucet schema source gcs

Library usage

use faucet_core::Source;
use faucet_source_gcs::{GcsCredentials, GcsFileFormat, GcsSource, GcsSourceConfig};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let cfg = GcsSourceConfig::new("analytics-exports")
    .prefix("events/dt=2026-06-16/")
    .auth(GcsCredentials::ApplicationDefault)
    .file_format(GcsFileFormat::JsonLines)
    .concurrency(20)
    .with_batch_size(5000);

let records = GcsSource::new(cfg).await?.fetch_all().await?;
println!("records: {}", records.len());
# Ok(())
# }

How it works

1. new() resolves GcsCredentials and builds both a data-plane Storage client and a control-plane StorageControl client once, reusing them across calls.
2. Listing uses the control-plane list_objects paginator (page size 1000), filtered by prefix and capped at max_objects. An explicit object_keys list skips listing entirely.
3. Each object's body is opened as an async buffered reader; with the compression feature it is wrapped in a per-key decompressor.
4. Objects are read concurrently via buffer_unordered(concurrency) and parsed per file_format.
5. stream_pages re-frames the decoded records into batch_size pages and yields them to the pipeline, keeping peak memory bounded for the streaming formats.

Dataset discovery

The source supports live introspection via Source::discover(): one control-plane delimiter (/) listing under the configured prefix (bucket root when unset) enumerates the "directories" directly below it, returning one dataset descriptor per common prefix with:

• name — the full object-name prefix (e.g. raw/orders/), kind: prefix
• config_patch — { "prefix": "raw/orders/" }, ready to deep-merge over the connection config as a matrix row

When the listing returns no common prefixes but does return objects directly under the prefix (a "leaf directory"), each object becomes a descriptor instead (kind: object, config_patch: { "object_keys": ["<full name>"] } — the exact-match field, which makes prefix inert), capped at the single listing page of 1000. schema and estimated_rows are never set — either would require reading or paging the whole listing. Discovery issues exactly one listing call and never recurses.

Lineage dataset URI

gs://<bucket> or gs://<bucket>/<prefix> — e.g. gs://my-bucket/events/2026/.

Feature flags

Enable the connector itself in the CLI/umbrella via the source-gcs feature.

Troubleshooting / FAQ

See also

• Connector reference — the full source/sink capability matrix.
• Authentication cookbook — shared auth providers and the {type, config} shape.
• Compression cookbook — gzip/zstd across file connectors.
• faucet-sink-gcs — the matching GCS sink.
• faucet-source-s3 — the AWS S3 equivalent with the same format semantics.
• faucet-common-gcs — the shared credentials enum and client builders.

Sharded execution (cluster Mode B)

Under faucet serve --cluster, a top-level shard: { count: N } block splits this source across cluster workers automatically — no connector config needed. Each worker reads the objects whose key hashes to its shard index (stable FNV-1a modulo count), so the partition is disjoint and complete: every object is read by exactly one worker, and the partition stays stable as new objects appear. Outside the cluster coordinator a run reads every object, unchanged.

max_objects is applied before the shard filter, so it caps the run's total object set (matching single-worker semantics) rather than multiplying by the shard count.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.