bucket

README

bucket

This package contains some helpful abstractions for interacting with GCS buckets.

Features

Basic GCS Operations

var err error // error handling omitted

b := bucket.NewBucket("my-bucket")

// upload file to bucket
err = b.Upload("/tmp/my-file.json", "my-object.json")

// download file from bucket
err = b.Upload("my-object.json", "/tmp/my-file.json")

// write to bucket
err = b.Write("my-object.json", []byte(`{"foo":"bar"}`))

// read from bucket
data, err := b.Read("my-object.json")

// delete object
err = b.Delete("my-object.json")

See the Bucket interface in bucket.go for a full list of supported operations.

Distributed Locking

var err error // error handling omitted

b := bucket.NewBucket("my-bucket")

// create a new Locker associated with `gs://my-bucket/my-lock-object`
locker := b.NewLocker("my-lock-object", 30 * time.Second, func(options *lock.Options) {
  options.ExpiresAfter = 300 * time.Second // if left unset, the lock will never expire
})

// Wait up to 30 seconds to acquire the lock, returning an error if we time out.
// If there is an existing lock, but it's more than 300 seconds old, 
// it will be expired (deleted so that we can claim the lock).
lockId, err := locker.Lock()

// do stuff while we hold the lock
// ...

// release the lock
err = locker.Unlock(lockId)

Smoke Testing

Often, it's useful to test code that depends on GCS against a real GCS bucket. This package includes a TestBucket feature that creates virtual Buckets, backed by the shared thelma-integration-test bucket (in the dsp-tools-k8s project).

Every TestBucket uses a random path prefix, so that Read() and Write() calls for the path my-object.json will translate to <random-prefix>/my-object.json. This prevents different tests (and even multiple concurrent runs of the same test) from stepping on each other.

Any objects written to the TestBucket are automatically cleaned up when the test finishes.

Two packages, assert and require, are provided, to support assertions on objects in the bucket. (require includes all the same assertions as assert but causes the test to fail immediately if an assertion fails, just like Testify's require package).

func TestSomething(t *testing.T) {
  // create a new TestBucket
  b := bucket.NewTestBucket(t)

  // do something with bucket, eg. b.Upload("/tmp/my-file.json", "my-object.json")
  // ...
  
  // use the testing/assert and testing
  assert.ObjectExists(t, b, "my-object.json")
  
  // no need to clean up my-object.json, it will be deleted automatically
}

See tests/example_smoke_test.go for a complete, runnable example.

Instrumentation

All GCS operations are logged at debug level. Log messages include:

• the type of operation that is being performad
• url for the object that is being updated
• whether the operation succeeded or failed
• duration of the call in milliseconds

Example message:

{
  "level": "debug",
  "bucket": {
    "name": "my-bucket",
    "prefix": "my-prefix/"
  },
  "object": {
    "name": "my-object",
    "url": "gs://my-bucket/my-prefix/my-object"
  },
  "call": {
    "kind": "write",
    "id": "ba7b5c"
  },
  "attrs":{
    "cache-control": "public, max-age=1337"
  },
  "time": "2022-02-25T13:31:14-05:00",
  "duration": 98.633113,
  "status":"ok",
  "message": "write finished in 98.633113ms"
}

Documentation

Index

• type Bucket
  • func NewBucket(bucketName string, options ...BucketOption) (Bucket, error)
• type BucketOption
• type BucketOptions
• type Locker
• type LockerOption

Types

type Bucket

type Bucket interface {
	// Name returns the name of the bucket
	Name() string

	// Close closes gcs client associated with this bucket
	Close() error

	// Exists returns true if the object exists, false otherwise
	Exists(objectName string) (bool, error)

	// Upload uploads a local file to the bucket
	Upload(localPath string, objectName string, attrs ...object.AttrSetter) error

	// Download downloads an object in the bucket to a local file
	Download(objectName string, localPath string) error

	// Read reads object contents
	Read(objectName string) ([]byte, error)

	// Write replaces object contents with given content
	Write(objectName string, content []byte, attrs ...object.AttrSetter) error

	// Delete deletes the object from the bucket
	Delete(objectName string) error

	// Attrs returns the attributes of an object (eg. creation time, cache control)
	Attrs(objectName string) (*storage.ObjectAttrs, error)

	// Update updates the attributes  of an object (eg. cache control)
	Update(objectName string, attrs ...object.AttrSetter) error

	// NewLocker returns a Locker instance for the given object
	NewLocker(objectName string, maxWait time.Duration, options ...LockerOption) Locker
}

Bucket offers a simple interface for operations on GCS buckets

func NewBucket

func NewBucket(bucketName string, options ...BucketOption) (Bucket, error)

type BucketOption

type BucketOption func(options *BucketOptions)

type BucketOptions

type BucketOptions struct {
	// Prefix is an optionally prefix to add to all object names in the bucket. Eg.
	// For bucket called "my-bucket" with a prefix of "my-prefix-",
	//    bucket.Read("foo") will read the object "gs://my-bucket/my-prefix-foo"
	Prefix string
}

BucketOptions optional configuration for a Bucket

type Locker

type Locker interface {
	// ObjectName returns the name of the object associated with this lock
	ObjectName() string
	// Lock waits to acquire the lock, timing out after maxTime. It returns a lock id / object generation number
	// that must be passed in to Unlock
	Lock() (int64, error)
	// Unlock releases the lock.
	Unlock(lockId int64) error
}

Locker is distributed locking mechanism implemented over GCS. Every Locker is associated with an object in a GCS bucket.

type LockerOption

type LockerOption func(*lock.Options)

LockerOption used for configuring Locker options