s3store

import "github.com/umbralcalc/stochadex/pkg/s3store"

Package s3store is the opt-in object-storage integration for the simulation’s data boundary: it moves runs between the engine and Amazon S3, or any S3-compatible store (MinIO, Cloudflare R2, Ceph). It is a SEPARATE Go module (github.com/umbralcalc/stochadex/pkg/s3store) so that the AWS SDK and its dependency tree — around fifteen modules — stay entirely out of the core engine’s go.mod. The engine module remains lean; you opt in only by importing this module.

A transport, not a format

This is the design decision that shapes the whole package. Object storage is orthogonal to what the bytes contain, so s3store does not know how to parse anything. It moves whole objects, and the caller pairs it with whichever reader or sink already handles that format — analysis.NewStateTimeStorageFromCsv, an Arrow reader, a JSON-log sink.

The alternative — an “S3 source” that reimplements CSV, JSON-log and Arrow handling against S3 — would need extending again for every future format, and would duplicate field handling (a CSV’s time column and state columns) that already exists. Moving bytes instead means every present and future format works over object storage for free.

Credentials

Credentials are never taken from configuration. They come from the standard AWS chain: environment variables, the shared config and credentials files, or an instance or pod IAM role. A stochadex config therefore names only a bucket and key, and can be committed and shared without leaking anything.

Config carries the two optional overrides. Region overrides what the SDK would resolve. Endpoint points the client at an S3-compatible store, and also switches on path-style addressing, which those stores expect — virtual-host style assumes the bucket is a subdomain of a real AWS endpoint.

Reading

Fetch downloads an object to a temporary file and returns its path with a cleanup function. Hand the path to the reader for the object’s format:

client, err := s3store.NewClient(ctx, s3store.Config{Region: "eu-west-2"})
path, cleanup, err := s3store.Fetch(ctx, client, "my-bucket", "runs/in.csv")
defer cleanup()
storage, err := analysis.NewStateTimeStorageFromCsv(path, 0, columns, true)

Writing

OutputFunction wraps any other sink: rows go to the inner sink, which writes to a local staging path, and the finished file is uploaded once the run ends. It implements simulator.FinalizingOutputFunction, and the transfer belongs at Finalize for two reasons — uploading per step would be one request per row, and a columnar file is not even readable until its buffers are sealed.

inner := simulator.NewJsonLogOutputFunction(staged)
sink := s3store.NewOutputFunction(inner, staged, "my-bucket", "runs/out.log", cfg)

Finalize reports transfer errors on stderr rather than panicking: the simulation has already completed successfully by that point, and losing a finished run to a transfer failure would be worse than reporting it.

From config

The distributed CLI (cmd/stochadex-full) registers this package against the config surface, so a run needs no Go at all:

data:
  source:
    s3: {bucket: my-bucket, key: runs/in.arrow, format: arrow}

simulation:
  output_function: {type: s3, bucket: my-bucket, key: runs/out.arrow, format: arrow}

The format key names what the object contains, and the transport pairs with the existing reader or sink for it.

Index

func Fetch

func Fetch(ctx context.Context, client *s3.Client, bucket, key string) (string, func(), error)

Fetch downloads an object to a temporary file and returns its path together with a cleanup function the caller must invoke when done. The temporary file keeps the key’s extension, so any format sniffing downstream still works.

Pair it with the reader for the object’s format, e.g. analysis.NewStateTimeStorageFromCsv.

func NewClient

func NewClient(ctx context.Context, cfg Config) (*s3.Client, error)

NewClient builds an S3 client from the standard AWS credential chain, applying any overrides in cfg.

func Upload

func Upload(ctx context.Context, client *s3.Client, bucket, key, path string) error

Upload copies a local file to an object.

type Config

Config carries the optional client overrides. Both fields may be empty, in which case the SDK’s own resolution applies.

type Config struct {
    // Region overrides the region the SDK would otherwise resolve.
    Region string
    // Endpoint points the client at an S3-compatible store — MinIO, Cloudflare R2, Ceph.
    // Setting it also switches on path-style addressing, which those stores expect;
    // virtual-host style assumes the bucket is a subdomain of a real AWS endpoint.
    Endpoint string
}

type OutputFunction

OutputFunction wraps any other sink so a run lands in object storage: rows go to the inner sink, which writes to a local staging path, and the finished file is uploaded once the run ends.

It implements simulator.FinalizingOutputFunction. Uploading per step would be one request per row, and a columnar file is not even readable until its buffers are sealed — so the transfer belongs at Finalize, after the inner sink has finalized itself.

type OutputFunction struct {
    // contains filtered or unexported fields
}

func NewOutputFunction

func NewOutputFunction(inner simulator.OutputFunction, staged, bucket, key string, config Config) *OutputFunction

NewOutputFunction wraps inner, which must write to the file at staged.

func (*OutputFunction) Configure

func (o *OutputFunction) Configure(settings *simulator.Settings)

func (*OutputFunction) Finalize

func (o *OutputFunction) Finalize()

Finalize seals the inner sink, uploads the staged file, and removes it. Errors are reported on stderr rather than panicking: the simulation itself has already completed successfully by this point, and losing the run to a transfer failure would be worse than reporting it.

func (*OutputFunction) Output

func (o *OutputFunction) Output(partitionName string, state []float64, cumulativeTimesteps float64)

Generated by gomarkdoc