storage

Documentation

Index

• Variables
• type Config
• 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 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")
)

Types

type Config

type Config struct {
	// MaxLogAge is the maximium amount of time that a log entry must be kept for.
	// The Storage instance is welcome to delete any log entries that exceed this
	// age.
	MaxLogAge time.Duration
}

Config is the set of runtime configuration parameters for this storage instance.

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 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 cfgtypes.ProjectName
	// 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 cfgtypes.ProjectName
	// 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(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 requsted 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(GetRequest, GetCallback) 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(cfgtypes.ProjectName, types.StreamPath) (*Entry, error)

	// Config installs the supplied configuration parameters into the storage
	// instance.
	Config(Config) 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.