gcsbuf

This Go library gives you an io.Writer to streamily write an object to Google Cloud Storage, and at the same time, streamily read that same object using an io.Reader.

Writing

Writing works by buffering the input data and periodically flushing updates to GCS. Object updates are implemented using composite objects. When appending a new chunk of data, that data is first uploaded to a new temp object, then composed with the existing object contents over itself.

When the object reaches the max number of compositions, the count is reset by copying and then renaming the object back to its original location -- this is purely a metadata operation and does not require the client to read or write all the data again.

When the writer calls Close(), the object is finalized by updating object metadata, to tell consumers that the object is done being written.

Each compose/append and copy/reset operation is guarded with a precondition, so that the operations fail if the object changed due to outside operations, preventing silent corruption.

You can configure the flush timeout (default: 5s) and buffer size (default: 4MB), and the writer will flush when either of those conditions are met.

w := gcsbuf.NewWriter(...)
fmt.Fprintln(w, "hello")
_ = w.Flush() // immediately flush to GCS
fmt.Fprintln(w, "world")
_ = w.Close() // flush and finalize

Reading

Reading works by iteratively making HTTP Range requests to the object and polling until either newly appended data is found, or until the object is marked as finalized.

r := gcsbuf.NewReader(...)
_, _ = io.Copy(os.Stdout, r) // reads until the object is finalized

In the example above, if the writer is still writing data, io.Copy will block until the writer completes, and receive new content each time it flushes.

If the writer has already written and finalized its data, io.Copy simply streams all of the data in one request.

Rate limits

GCS enforces a 1 mutation per second per object rate limit. Each flush cycle performs one compose on the main object (temp object operations target separate objects), so the default 5-second flush interval stays well within this limit. If you use manual Flush() calls or a low FlushInterval or write data faster than your FlushSize per second, you may hit rate limits; all GCS operations are retried automatically with exponential backoff, but this will slow down your writes.

Observability

Both WriterOptions and ReaderOptions accept an optional OnOp callback, called after each GCS API operation with the operation name, duration, and error (nil on success). Operations: upload, attrs, compose, delete, rewrite, patch, read.

w := gcsbuf.NewWriter(ctx, gcsbuf.WriterOptions{
    Client: client,
    Bucket: "my-bucket",
    Object: "my-object",
    OnOp: func(op string, d time.Duration, err error) {
        slog.Info("gcs", "op", op, "dur", d, "err", err)
    },
})

You can use this to export Prometheus metrics for each operation.

Testing

You can test this with ./cmd/gcscat, which reads from stdin, streamily writes it to GCS, and simultaneously reads it back out, and logs operations.

echo "hello world" | go run ./cmd/gcscat <bucket> <object>

Background

I originally used a hack scheme like this when creating Google Cloud Build to efficiently write logs to durable storage, and to allow clients like gcloud to simultaneously stream the logs objects to users until build completion. Logs were separately written to Google Cloud Logging, which didn't support a fast logs tailing API at the time.