--- id: google-cloud--gcs-reader title: "GCS reader" summary: "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`." audience: user keywords: "input, output, nodata, authentication, serviceaccountkey, serviceurl, bucketname, objectname, postprocessing, postprocessbucket, postprocessprefix, postprocesstag, errorbucket, errorprefix" --- # 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:///` (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` = `/`, `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 `/` 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 `/` (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-`), 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 ```json { "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