storage

Documentation

Index

• Variables
• func GCSSourceFactory(c stiface.Client) factory.SourceFactory
• func GetStorageClient(writeAccess bool) (stiface.Client, error)
• func NewRowWriter(ctx context.Context, client stiface.Client, bucket string, path string) (row.Sink, error)
• func NewSinkFactory(client stiface.Client, outputBucket string) factory.SinkFactory
• func NewTestSource(client stiface.Client, dp etl.DataPath, label string) (etl.TestSource, error)
• func ObjectWriter(ctx context.Context, client stiface.Client, bucket string, path string) stiface.Writer
• type Closer
  • func (t *Closer) Close() error
• type GCSSource
  • func (src *GCSSource) Date() civil.Date
  • func (src *GCSSource) Detail() string
  • func (src *GCSSource) NextTest(maxSize int64) (string, []byte, error)
  • func (src *GCSSource) Type() string
• type RowWriter
  • func (rw *RowWriter) Close() error
  • func (rw *RowWriter) Commit(rows []interface{}, label string) (int, error)
• type SinkFactory
  • func (sf *SinkFactory) Get(ctx context.Context, path etl.DataPath) (row.Sink, etl.ProcessingError)
• type TarReader

Variables

var ErrOversizeFile = errors.New("Oversize file")

ErrOversizeFile is returned when exceptionally large files are skipped.

Functions

func GCSSourceFactory

func GCSSourceFactory(c stiface.Client) factory.SourceFactory

GCSSourceFactory returns the default SourceFactory

func GetStorageClient

func GetStorageClient(writeAccess bool) (stiface.Client, error)

GetStorageClient provides a storage reader client. This contacts the backend server, so should be used infrequently.

func NewRowWriter

func NewRowWriter(ctx context.Context, client stiface.Client, bucket string, path string) (row.Sink, error)

NewRowWriter creates a RowWriter.

func NewSinkFactory

func NewSinkFactory(client stiface.Client, outputBucket string) factory.SinkFactory

NewSinkFactory returns the default SinkFactory

func NewTestSource

func NewTestSource(client stiface.Client, dp etl.DataPath, label string) (etl.TestSource, error)

NewTestSource creates an TestSource suitable for injecting into Task. Caller is responsible for calling Close on the returned object.

uri should be of form gs://bucket/filename.tar or gs://bucket/filename.tgz FYI Using a persistent client saves about 80 msec, and 220 allocs, totalling 70kB.

func ObjectWriter

func ObjectWriter(ctx context.Context, client stiface.Client, bucket string, path string) stiface.Writer

ObjectWriter creates a writer to a named object. It may overwrite an existing object. Caller must Close() the writer, or cancel the context.

Types

type Closer

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

Closer handles gzip files.

func (*Closer) Close

func (t *Closer) Close() error

Close invokes the gzip and body Close() functions.

type GCSSource

type GCSSource struct {
	FilePath      string
	Size          int64
	TarReader                   // TarReader interface provided by an embedded struct.
	io.Closer                   // Closer interface to be provided by an embedded struct.
	RetryBaseTime time.Duration // The base time for backoff and retry.
	TableBase     string        // TableBase is BQ table associated with this source, or "invalid".
	PathDate      civil.Date    // Date associated with YYYY/MM/DD in FilePath.
}

GCSSource wraps a gsutil tar file containing tests.

func (*GCSSource) Date

func (src *GCSSource) Date() civil.Date

Date returns a civil.Date associated with the GCSSource archive path.

func (*GCSSource) Detail

func (src *GCSSource) Detail() string

Detail returns a string for use in logs.

func (*GCSSource) NextTest

func (src *GCSSource) NextTest(maxSize int64) (string, []byte, error)

NextTest reads the next test object from the tar file. Skips reading contents of any file larger than maxSize, returning empty data and storage.ErrOversizeFile. Returns io.EOF when there are no more tests.

func (*GCSSource) Type

func (src *GCSSource) Type() string

Type returns a string for use in metrics and logs.

type RowWriter

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

RowWriter implements row.Sink to a GCS file backend.

func (*RowWriter) Close

func (rw *RowWriter) Close() error

Close synchronizes on the tokens, and closes the backing file.

func (*RowWriter) Commit

func (rw *RowWriter) Commit(rows []interface{}, label string) (int, error)

Commit commits rows, in order, to the GCS object. The GCS object is not available until Close is called, at which point the entire object becomes available atomically. The returned int is the number of rows written (and pending), or, if error is not nil, an estimate of the number of rows written.

type SinkFactory

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

SinkFactory implements factory.SinkFactory.

func (*SinkFactory) Get

func (sf *SinkFactory) Get(ctx context.Context, path etl.DataPath) (row.Sink, etl.ProcessingError)

Get implements factory.SinkFactory

type TarReader

type TarReader interface {
	Next() (*tar.Header, error)
	Read(b []byte) (int, error)
}

TarReader provides Next and Read functions.