storage

Documentation

Index

• Variables
• func HashKey(parts ...string) string
• type Cache
• type CacheKey
• type Entry
  • func MakeEntry(d []byte, idx types.MessageIndex) *Entry
  • func (e *Entry) GetLogEntry() (*logpb.LogEntry, error)
  • func (e *Entry) GetStreamIndex() (types.MessageIndex, error)
• type ExpungeRequest
• type GetCallback
• type GetRequest
• type PutRequest
• type Storage

Variables

var (
	// ErrExists returned if an attempt is made to overwrite an existing record.
	ErrExists = errors.New("storage: record exists")

	// ErrDoesNotExist returned if an attempt is made to read a record that
	// doesn't exist.
	ErrDoesNotExist = errors.New("storage: record does not exist")

	// ErrBadData is an error returned when the stored data is invalid.
	ErrBadData = errors.New("storage: bad data")

	// ErrReadOnly can be returned by Storage methods to indicate that the Storage
	// is read-only.
	ErrReadOnly = errors.New("storage: read only")
)

Functions

func HashKey

func HashKey(parts ...string) string

HashKey composes a hex-encoded SHA256 string from the supplied parts. The local schema version and KeyType are included in the hash.

Types

type Cache

type Cache interface {
	// Put caches an Items into the cache. If an item already exists in the cache,
	// it will be overridden.
	//
	// The data in val may be directly retained by the cache, and the caller
	// promises not to mutate it after being passed to Put.
	//
	// If exp, the supplied expiration Duration, is >0, the cache should only
	// store the data if it can expire it after this period of time.
	//
	// This method does not return whether or not the caching storage was
	// successful.
	Put(c context.Context, key CacheKey, val []byte, exp time.Duration)

	// Get retrieves a cached entry. If the entry was present, the returned
	// boolean will be true.
	//
	// The returned byte slice may not be modified by the caller.
	//
	// This method does not distinguish between an error and missing data. Either
	// valid data is returned, or it is not.
	Get(c context.Context, key CacheKey) ([]byte, bool)
}

Cache is a simple cache interface. It is capable of storage and retrieval.

type CacheKey

type CacheKey struct {
	// Schema is the item's schema value. If empty, the item is schemaless.
	Schema string

	// Type is the item's type identifier. If empty, the item has no type.
	Type string

	// Key is the item's individual key. It uniquely identifies the Item within
	// the scope of the Schema and Type.
	Key string
}

CacheKey is a single cache item's key.

type Entry

type Entry struct {
	D []byte
	// contains filtered or unexported fields
}

Entry is a logpb.LogEntry wrapper that lazily evaluates / unmarshals the underlying LogEntry data as needed.

Entry is not goroutine-safe.

func MakeEntry

func MakeEntry(d []byte, idx types.MessageIndex) *Entry

MakeEntry creates a new Entry.

All Entry must be backed by data. The index, "idx", is optional. If <0, it will be calculated by unmarshalling the data into a LogEntry and pulling the value from there.

func (*Entry) GetLogEntry

func (e *Entry) GetLogEntry() (*logpb.LogEntry, error)

GetLogEntry returns the unmarshalled LogEntry data.

The first time this is called, the LogEntry will be unmarshalled from its underlying data.

func (*Entry) GetStreamIndex

func (e *Entry) GetStreamIndex() (types.MessageIndex, error)

GetStreamIndex returns the LogEntry's stream index.

If this needs to be calculated by unmarshalling the LogEntry, this will be done. If this fails, an error will be returned.

If GetLogEntry has succeeded, subsequent GetStreamIndex calls will always be fast and succeed.

type ExpungeRequest

type ExpungeRequest struct {
	// Project is the project name of the stream.
	Project string
	// Path is the stream path to remove from storage.
	Path types.StreamPath
}

ExpungeRequest is a request to expunge the data associated with this stream.

type GetCallback

type GetCallback func(*Entry) bool

GetCallback is invoked for each record in the Get request. If it returns false, iteration should stop.

The MessageIndex may be -1 if the message index isn't known. In this case, the caller will have to unmarshal the log entry data to determine its index.

type GetRequest

type GetRequest struct {
	// Project is the project name of the stream.
	Project string
	// Path is the stream path to retrieve.
	Path types.StreamPath
	// Index is the entry's stream index.
	Index types.MessageIndex

	// Limit is the maximum number of records to return before stopping iteration.
	// If <= 0, no maximum limit will be applied.
	//
	// The Storage instance may return fewer records than the supplied Limit as an
	// implementation detail.
	Limit int
	// KeysOnly, if true, allows (but doesn't require) the Storage instance to
	// omit entry data in its get callback. For scanning operations, this can be
	// much cheaper/faster than full data queries.
	KeysOnly bool
}

GetRequest is a request to retrieve a series of LogEntry records.

type PutRequest

type PutRequest struct {
	// Project is the project name of the stream.
	Project string
	// Path is the stream path to retrieve.
	Path types.StreamPath
	// Index is the entry's stream index.
	Index types.MessageIndex

	// Values are contiguous sequential records to add to the storage. The first
	// index in values corresponds to Index.
	Values [][]byte
}

PutRequest describes adding a single storage record to BigTable.

type Storage

type Storage interface {
	// Close shuts down this instance, releasing any allocated resources.
	Close()

	// Writes log record data to storage.
	//
	// If the data already exists, ErrExists will be returned.
	Put(context.Context, PutRequest) error

	// Get invokes a callback over a range of sequential LogEntry records.
	//
	// These log entries will be returned in order (e.g., seq(Rn) < seq(Rn+1)),
	// but, depending on ingest, may not be contiguous.
	//
	// The underlying Storage implementation may return fewer records than
	// requested based on availability or implementation details; consequently,
	// receiving fewer than requested records does not necessarily mean that more
	// records are not available.
	//
	// Returns nil if retrieval executed successfully, ErrDoesNotExist if
	// the requested stream does not exist, and an error if an error occurred
	// during retrieval.
	Get(context.Context, GetRequest, GetCallback) error

	// Expunge removes all entries related to the given project/path.
	Expunge(context.Context, ExpungeRequest) error

	// Tail retrieves the latest log in the stream. If the stream has no logs, it
	// will fail with ErrDoesNotExist.
	//
	// The MessageIndex may be -1 if the message index isn't known. In this case,
	// the caller will have to unmarshal the log entry data to determine its
	// index.
	Tail(ctx context.Context, projectName string, stream types.StreamPath) (*Entry, error)
}

Storage is an abstract LogDog storage implementation. Interfaces implementing this may be used to store and retrieve log records by the collection service layer.

All of these methods must be synchronous and goroutine-safe.

All methods may return errors.Transient errors if they encounter an error that may be transient.