[RFC] Google Cloud Storage (GCS) Source and Sink Plugins

[RFC] Google Cloud Storage (GCS) Source and Sink Plugins

Problem Statement

Data Prepper ships first-class object-storage support for AWS: the s3 source and s3
sink cover ingestion from and delivery to Amazon S3, including notification-driven
(SQS) and scheduled-scan ingestion, pluggable codecs, compression, and cluster-wide work
coordination.

There is no equivalent for Google Cloud Storage. A pipeline running entirely on GCP
has no native way to read objects from, or write objects to, a GCS bucket. Users today must
either route data through S3 (cross-cloud egress, extra credentials, latency) or write a
custom out-of-band process — neither is viable for a durable, GCP-native deployment.

This is the same class of capability the s3 plugins provide for AWS. Google Cloud needs
the analogous source and sink.

Current state

• data-prepper-plugins/s3-source — ingestion from S3 via SQS notifications or scheduled
  scan, with source-coordinator-based partitioning for multi-node clusters, pluggable
  codecs (newline, JSON, CSV, Parquet), compression, and S3 Select.
• data-prepper-plugins/s3-sink — delivery to S3 with buffering, size/time/event
  thresholds, dynamic key/prefix templating, and codecs.
• data-prepper-plugins/s3-common, aws-plugin-api — shared AWS client/credential wiring.

No gcs-* plugin modules exist.

Proposed Solution

Add two new plugins mirroring the S3 plugins' architecture and configuration surface, plus a
shared common module:

• gcs-source — reads objects from a GCS bucket.
• gcs-sink — writes objects to a GCS bucket.
• gcs-common — shared client construction, credential handling, and bucket/object
  helpers (parallel to s3-common).

Reusing existing codec, compression, buffer, and source-coordination abstractions means the
GCS plugins differ from their S3 counterparts only in the storage client and the
notification transport.

Authentication — reuse existing GCP credential model

Data Prepper has no GCP authentication code today, so there is nothing in this repo to
import directly. But we have already built and reviewed a GCP credential model in
ml-commons (the google_cloud connector, PR #4921 / RFC #4915), and this RFC deliberately
reuses that design so GCP is configured identically across the OpenSearch ecosystem. It
also shares implementation within data-prepper with the proposed GCP Secret Manager
provider (RFC #7010), which needs the same credentials.

Shared module. Rather than each GCS plugin (and #7010) re-implementing credential
construction, factor a shared gcp-plugin-api / gcp-plugin auth module, mirroring how
aws-plugin-api / aws-plugin are shared by the S3 and other AWS plugins. This module owns
the google-auth-library-oauth2-http–based credential provider and is consumed by
gcs-source, gcs-sink, and the Secret Manager provider alike.

Injection interface — mirroring AwsCredentialsSupplier. The AWS module exposes an
AwsCredentialsSupplier interface (in aws-plugin-api) via a Data Prepper extension
(@DataPrepperExtensionPlugin); plugins declare it as a constructor dependency and use the
returned AwsCredentialsProvider to build their SDK clients. The GCP module follows the same
shape:

// gcp-plugin-api
public interface GcpCredentialsSupplier {
    // Returns a google-auth-library credential for use in google-cloud-* SDK clients.
    GoogleCredentials getCredentials(GcpCredentialsOptions options);
    Optional<String> getDefaultProjectId();
}

gcs-source / gcs-sink (and the #7010 Secret Manager provider) receive a
GcpCredentialsSupplier by injection and pass the resulting GoogleCredentials to the
google-cloud-storage (or Secret Manager) client builder — exactly as s3-source passes an
AwsCredentialsProvider into its S3Client. The GcpCredentialsOptions carries the two
modes (service-account key vs. auth_mode: adc), scopes, and token_uri, so the credential
model above lives in one place.

Credential model — matching ml-commons google_cloud. Two modes, same field names and
defaults as the connector, so users configure GCP the same way in a connector and in a
pipeline:

• Service-account key — private_key + client_email, token_uri defaulting to
  https://oauth2.googleapis.com/token.
• ADC / Workload Identity — auth_mode: adc, no credentials (uses
  GoogleCredentials.getApplicationDefault()).
• scopes defaulting to https://www.googleapis.com/auth/cloud-platform.
• The same token_uri SSRF guard ml-commons applies (HTTPS + host restricted to
  *.googleapis.com), since token_uri is user-supplied.

The GCS plugins themselves use the official com.google.cloud:google-cloud-storage client,
authenticated with credentials from the shared module.

gcp:
  project_id: my-project
  # Service-account key mode (fields match ml-commons google_cloud connector):
  private_key: ${{gcp_secrets:sa-key:private_key}}
  client_email: my-sa@my-project.iam.gserviceaccount.com
  # token_uri: https://oauth2.googleapis.com/token   # default
  # scopes: https://www.googleapis.com/auth/cloud-platform   # default
  #
  # OR ADC / Workload Identity — omit credentials entirely:
  # auth_mode: adc

gcs-source

Two ingestion modes, matching s3-source:

1. Scan mode (scan:) — scheduled enumeration of a bucket/prefix. This is the mode the
batch-inference flow depends on (scan for staged input and for job-output objects). Uses the
Data Prepper source coordinator for partition assignment so scanning is safe across a
multi-node cluster, mirroring S3ScanPartitionCreationSupplier / ScanObjectWorker.

2. Notification mode (notification_type: pubsub) — event-driven ingestion. GCS
publishes object-create notifications to a Pub/Sub topic; the source subscribes and
reads newly created objects. This is the GCS analog of S3 → SQS.

Config surface (parallel to S3SourceConfig):

source:
  gcs:
    codec:
      newline:              # newline | json | csv | parquet
    compression: none       # none | gzip | automatic
    scan:
      buckets:
        - bucket:
            name: my-bucket
            key_prefix:
              include: [ "input/" ]
      scheduling:
        interval: PT5M
    # OR notification-driven:
    # notification_type: pubsub
    # pubsub:
    #   subscription: projects/my-project/subscriptions/gcs-events
    workers: 1
    acknowledgments: false
    gcp:
      project_id: my-project

gcs-sink

Delivery to GCS with the same buffering/threshold/key-templating model as s3-sink:

sink:
  gcs:
    bucket: my-bucket
    object_key:
      path_prefix: "output/%{yyyy}/%{MM}/%{dd}/"
    codec:
      ndjson:
    threshold:
      event_count: 10000
      maximum_size: 50mb
      event_collect_timeout: PT5M
    gcp:
      project_id: my-project

Scope

In scope

• gcs-source scan mode with source-coordinator partitioning.
• gcs-source Pub/Sub notification mode.
• gcs-sink with buffering, thresholds, and key templating.
• Service-account-key and ADC/Workload-Identity auth.
• Reuse of existing codecs (newline, JSON, CSV, Parquet) and compression.

Out of scope (initial)

• A GCS analog of S3 Select (server-side query pushdown). GCS has no direct equivalent;
  can be a later enhancement.
• BigQuery source/sink (separate proposal).

Alternatives Considered

1. Route GCP data through S3. Rejected — defeats the purpose of a GCP-native stack;
   adds cross-cloud egress cost, latency, and a second cloud's credentials.
2. A single generic "object storage" plugin abstracting S3/GCS/Azure. Rejected for now —
   the existing plugin ecosystem is per-service (s3-source, s3-sink), and the auth,
   notification, and client models differ enough that a premature abstraction would be
   leaky. A per-service plugin matching the established pattern is the lower-risk path;
   shared code can be factored into a common module later if a third provider lands.
3. Scan-only, no Pub/Sub mode. Rejected as the target — scan alone would satisfy the
   batch-inference use case, but notification-driven ingestion is a first-class S3 feature
   and omitting it would leave GCS a second-class citizen. (It could reasonably be phased:
   scan first, Pub/Sub second — see open questions.)

Open Questions

1. Phasing. Ship gcs-source (scan) + gcs-sink first as the minimum for the
   batch-inference use case, and add Pub/Sub notification mode in a follow-up? Or land all
   three modes together for feature parity with S3 from day one?
2. gcs-common boundary. How much can genuinely be shared with s3-common vs. how much
   is S3-specific? Proposing a separate gcs-common rather than widening s3-common.
3. Shared GCP auth module ownership. The gcp-plugin auth module is also needed by the
   GCP Secret Manager provider (RFC #7010). Which RFC/PR lands it first, and does it live as
   gcp-plugin + gcp-plugin-api (parallel to aws-plugin/aws-plugin-api)? These two
   efforts should coordinate to avoid two divergent GCP credential implementations.
4. Codec parity. Confirm the existing codec plugins are storage-agnostic enough to reuse
   unchanged (expected yes — they operate on streams, not S3 clients).
5. Integration testing. S3 plugin ITs are gated behind live AWS resources and excluded
   from the default build. Propose the same treatment for GCS (live GCS bucket + Pub/Sub
   subscription supplied via -D properties).

Request for Comments

Feedback sought on: the two-plugin + common-module split, the Pub/Sub notification design,
the phasing question, and the auth config shape (gcp: block) relative to the existing
aws: block.

Start by reading data-prepper-plugins/s3-source, s3-sink, s3-common, and aws-plugin-api to understand the existing plugin and credential patterns. Before implementation, resolve the RFC's phasing, gcp-common/auth ownership, codec reuse, and integration-testing questions; done means an agreed scope and implementation plan for the GCS source and sink.