GCS reader

Reads an object from Google Cloud Storage and emits it as a file stream, for example for a CSV or ASN.1 decoder. Every object is a batch: once it is processed and committed downstream, the object can be deleted, moved or tagged; a failed object can be moved to an error location. The object comes from an incoming object file info (GCS enumerator) or from the configuration. If the object does not exist, the input is passed to NoData.

Pins

Pin	Kind	Types	Description
Input	input	Viaducta.TypeSystem.Metadata.ViaFileInfo	Triggers the read. A ViaFileInfo whose FilePath is a gs:// URL selects the object; any other input reads the configured object.
Output	output	Viaducta.Plugins.Files.Types.ViaFileStream	The object stream with name, size and date, sent in the batch of the object.
NoData	no-data output	same as input	Called with the input record when the bucket or object does not exist (HTTP 404).

Configuration

Field	Type	Default	Description
Authentication	dropdown: 0 ServiceAccountKey, 1 ApplicationDefault	0	How to sign in.
ServiceAccountKey	password		0: the whole service-account JSON key file. Required for 0, except with ServiceUrl.
ServiceUrl	string		JSON API endpoint of a GCS emulator or a private endpoint, for example http://localhost:4443. Empty: Google Cloud Storage.
BucketName	string		Bucket of the object, used when the input is not an object ViaFileInfo.
ObjectName	string		Object name including folders, for example daily/export.csv, used when the input is not an object ViaFileInfo.
PostProcessing	dropdown: 0 None, 1 Delete, 2 Move, 3 Tag	0	What happens to the object after its batch committed.
PostProcessBucket	string		2 Move: target bucket (must exist); empty keeps the object's bucket.
PostProcessPrefix	string		2 Move: put in front of the name, for example done/. Move needs this or PostProcessBucket.
PostProcessTag	string	viaducta-processed	3 Tag: custom metadata entry to set, name or name=value; without a value the UTC time (yyyy-MM-ddTHH:mm:ssZ) is the value.
ErrorBucket	string		Bucket a failed object is moved to (must exist); empty keeps the object's bucket.
ErrorPrefix	string		Put in front of the name of a failed object, for example error/. With ErrorBucket and ErrorPrefix empty, failed objects stay where they are.

Behaviour

• The storage client is created at flow start; sign-in and post-processing settings are checked then (Move without target, Tag without name fail the start). Sign-in modes and the credential-less emulator access as in GCS enumerator. The identity needs storage.objects.get, plus storage.objects.delete for Delete, storage.objects.create and storage.objects.delete for Move (and the error location), storage.objects.update for Tag (role Storage Object User covers all).
• Per input the object to read is chosen: when the input is a ViaFileInfo whose FilePath is gs://<bucket>/<name> (as emitted by GCS enumerator), bucket and name are taken from it; otherwise BucketName/ObjectName are used. The endpoint always comes from the sign-in. With neither, the input fails with an error.
• The object's metadata is read, then the content of exactly that generation is opened, so stream, size, date and version belong together. A 404 for either call (bucket or object missing, or deleted in between) calls NoData with the input record (no batch); other errors (authentication, network) raise an error.
• Otherwise a batch labelled with the object name is opened and Output is called with a ViaFileStream in that batch: Stream (the response body, read forward as it downloads), FilePath = <bucket>/<name>, Name = object name, Extension from the name, Size = size in bytes, CreateDateTime and LastAccessTime = when the content was written (UTC). When Output returns the stream is closed and the batch completes: downstream sinks (database writers, ...) commit.
• After the commit the object is post-processed: 1 Delete deletes it; 2 Move copies it (server-side rewrite, with content type and metadata) to <PostProcessBucket or the same bucket>/<PostProcessPrefix><name> and deletes the original; 3 Tag sets PostProcessTag as a custom metadata entry (other metadata is kept). All three only act on the generation that was read: if the object was replaced meanwhile they fail with HTTP 412 and the object is kept. A post-processing error raises an error; the data is already committed and the object is read again by the next run.
• When Output or the commit fails, the batch fails (sinks roll back) and the object is not post-processed. With ErrorBucket or ErrorPrefix set it is moved to <ErrorBucket or the same bucket>/<ErrorPrefix><name> (a failed move is logged). The error is raised in both cases.

Notes

• Downstream components must consume the stream before Output returns (decoders do); it is closed afterwards. The stream is forward-only (not seekable).
• Moving within the listed bucket: give the target a prefix outside the enumerator's Prefix (for example list in/, move to done/), or the next run lists the moved objects again. Tag + the enumerator's SkipTag keeps the objects in place.
• GCS has no object tags; Tag writes custom metadata (x-goog-meta-<name>), which the enumerator reads from its listing.
• Google Cloud Storage checks the generation atomically (ifGenerationMatch on the delete, ifSourceGenerationMatch on the copy, ifGenerationMatch on the metadata patch). Before each, the plugin also reads the object's current generation, so emulators that ignore these conditions (fake-gcs-server) fail a changed object too, with a short window.
• In a bucket with object versioning Delete and Move keep the old generation as a noncurrent version; with soft delete it stays restorable for the retention period.
• Archive, Coldline and Nearline objects can be read directly (retrieval fees apply).
• Store the service-account key as an expression setting (=Secrets.GcsServiceAccountKey).

Example

{ "Authentication": 1, "PostProcessing": 2, "PostProcessPrefix": "done/", "ErrorPrefix": "error/" }

Timer → GCS enumerator (Prefix in/) → GCS reader → CSV decoder → Table writer (bucket and name from the enumerator)

Timer → GCS reader (BucketName, ObjectName set) → CSV decoder → Table writer; (NoData) Log