blob

Documentation

Overview

Package blob implements simple storage of immutable, unstructured binary large objects (BLOBs).

Index

• Variables
• func AddSupportedStorage(urlScheme string, defaultConfigFunc func() interface{}, ...)
• func IterateAllPrefixesInParallel(ctx context.Context, parallelism int, st Storage, prefixes []ID, ...) error
• func WithUploadProgressCallback(ctx context.Context, callback ProgressFunc) context.Context
• type CancelFunc
• type ConnectionInfo
  • func (c ConnectionInfo) MarshalJSON() ([]byte, error)
  • func (c *ConnectionInfo) UnmarshalJSON(b []byte) error
• type ID
• type Metadata
  • func ListAllBlobs(ctx context.Context, st Storage, prefix ID) ([]Metadata, error)
  • func ListAllBlobsConsistent(ctx context.Context, st Storage, prefix ID, maxAttempts int) ([]Metadata, error)
• type ProgressFunc
  • func ProgressCallback(ctx context.Context) ProgressFunc
• type Storage
  • func NewStorage(ctx context.Context, cfg ConnectionInfo) (Storage, error)

Variables

var ErrBlobNotFound = errors.New("BLOB not found")

ErrBlobNotFound is returned when a BLOB cannot be found in storage.

Functions

func AddSupportedStorage

func AddSupportedStorage(
	urlScheme string,
	defaultConfigFunc func() interface{},
	createStorageFunc func(context.Context, interface{}) (Storage, error))

AddSupportedStorage registers factory function to create storage with a given type name.

func IterateAllPrefixesInParallel

func IterateAllPrefixesInParallel(ctx context.Context, parallelism int, st Storage, prefixes []ID, callback func(Metadata) error) error

IterateAllPrefixesInParallel invokes the provided callback and returns the first error returned by the callback or nil.

func WithUploadProgressCallback

func WithUploadProgressCallback(ctx context.Context, callback ProgressFunc) context.Context

WithUploadProgressCallback returns a context that passes callback function to be used storage upload progress.

Types

type CancelFunc

type CancelFunc func()

CancelFunc requests cancellation of a storage operation.

type ConnectionInfo

type ConnectionInfo struct {
	Type   string
	Config interface{}
}

ConnectionInfo represents JSON-serializable configuration of a blob storage.

func (ConnectionInfo) MarshalJSON

func (c ConnectionInfo) MarshalJSON() ([]byte, error)

MarshalJSON returns JSON-encoded storage configuration.

func (*ConnectionInfo) UnmarshalJSON

func (c *ConnectionInfo) UnmarshalJSON(b []byte) error

UnmarshalJSON parses the JSON-encoded data into ConnectionInfo.

type ID

type ID string

ID is a string that represents blob identifier.

type Metadata

type Metadata struct {
	BlobID    ID
	Length    int64
	Timestamp time.Time
}

Metadata represents metadata about a single BLOB in a storage.

func ListAllBlobs

func ListAllBlobs(ctx context.Context, st Storage, prefix ID) ([]Metadata, error)

ListAllBlobs returns Metadata for all blobs in a given storage that have the provided name prefix.

func ListAllBlobsConsistent

func ListAllBlobsConsistent(ctx context.Context, st Storage, prefix ID, maxAttempts int) ([]Metadata, error)

ListAllBlobsConsistent lists all blobs with given name prefix in the provided storage until the results are consistent. The results are consistent if the list result fetched twice is identical. This guarantees that while the first scan was in progress, no new blob was added or removed. maxAttempts specifies maximum number of list attempts (must be >= 2)

type ProgressFunc

type ProgressFunc func(desc string, completed, total int64)

ProgressFunc is used to report progress of a long-running storage operation.

func ProgressCallback

func ProgressCallback(ctx context.Context) ProgressFunc

ProgressCallback gets the progress callback function from the context.

type Storage

type Storage interface {
	// PutBlob uploads the blob with given data to the repository or replaces existing blob with the provided
	// id with given contents.
	PutBlob(ctx context.Context, blobID ID, data []byte) error

	// DeleteBlob removes the blob from storage. Future Get() operations will fail with ErrNotFound.
	DeleteBlob(ctx context.Context, blobID ID) error

	// GetBlob returns full or partial contents of a blob with given ID.
	// If length>0, the the function retrieves a range of bytes [offset,offset+length)
	// If length<0, the entire blob must be fetched.
	GetBlob(ctx context.Context, blobID ID, offset, length int64) ([]byte, error)

	// ListBlobs invokes the provided callback for each blob in the storage.
	// Iteration continues until the callback returns an error or until all matching blobs have been reported.
	ListBlobs(ctx context.Context, blobIDPrefix ID, cb func(bm Metadata) error) error

	// ConnectionInfo returns JSON-serializable data structure containing information required to
	// connect to storage.
	ConnectionInfo() ConnectionInfo

	// Close releases all resources associated with storage.
	Close(ctx context.Context) error
}

Storage encapsulates API for connecting to blob storage.

The underlying storage system must provide:

* high durability, availability and bit-rot protection * read-after-write - blob written using PubBlob() must be immediately readable using GetBlob() and ListBlobs() * atomicity - it mustn't be possible to observe partial results of PubBlob() via either GetBlob() or ListBlobs() * timestamps that don't go back in time (small clock skew up to minutes is allowed) * reasonably low latency for retrievals

The required semantics are provided by existing commercial cloud storage products (Google Cloud, AWS, Azure).

func NewStorage

func NewStorage(ctx context.Context, cfg ConnectionInfo) (Storage, error)

NewStorage creates new storage based on ConnectionInfo. The storage type must be previously registered using AddSupportedStorage.