store

Documentation

Overview

Package store contains the implementation of Store. Store provides a simple append-only index of telemetry events. On append it marshals the data into JSON and appends it to the log file. The events are managed based on a key. Key is provided by the caller. It also tracks whether the event has been processed or not. This is useful for determining if the event needs to be sent to telemetry or not.

Index

• type Cursor
  • func NewCursor(items []map[string]interface{}) *Cursor
  • func (c *Cursor) Len() int
  • func (c *Cursor) Next() bool
  • func (c *Cursor) Value(v IndexMarshaller) error
• type FSStore
  • func New(opts *Options) *FSStore
  • func (s *FSStore) AddDefaultField(k string, v interface{})
  • func (s *FSStore) Append(ctx context.Context, value IndexMarshaller) error
  • func (s *FSStore) Get(ctx context.Context, key string, value IndexMarshaller) error
  • func (s *FSStore) GetAll(ctx context.Context) *Cursor
  • func (s *FSStore) GetUnprocessed(ctx context.Context) *Cursor
  • func (s *FSStore) Init(ctx context.Context) error
  • func (s *FSStore) MarkProcessed(ctx context.Context, recs []IndexMarshaller) error
• type IndexMarshaller
• type Options
• type Store
• type TestClosableBuffer
  • func (TestClosableBuffer) Close() error

Types

type Cursor

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

Cursor implements iteration over generic event data.

func NewCursor

func NewCursor(items []map[string]interface{}) *Cursor

NewCursor creates new cursor instance for given items.

func (*Cursor) Len

func (c *Cursor) Len() int

Len returns number of items in cursor.

func (*Cursor) Next

func (c *Cursor) Next() bool

Next moves cursor to next item.

func (*Cursor) Value

func (c *Cursor) Value(v IndexMarshaller) error

Value sets v to current value. Returns error if value cannot be unmarshaled.

type FSStore

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

FSStore is the concrete implementation of Store.

func New

func New(opts *Options) *FSStore

New creates a new FSStore instance.

func (*FSStore) AddDefaultField

func (s *FSStore) AddDefaultField(k string, v interface{})

AddDefaultField adds a default field to the store. These fields are added to all events.

func (*FSStore) Append

func (s *FSStore) Append(ctx context.Context, value IndexMarshaller) error

Append adds an event to the store.

func (*FSStore) Get

func (s *FSStore) Get(ctx context.Context, key string, value IndexMarshaller) error

Get gets an event from the store.

func (*FSStore) GetAll

func (s *FSStore) GetAll(ctx context.Context) *Cursor

GetAll returns all the events in the store. The latest versions of the events.

func (*FSStore) GetUnprocessed

func (s *FSStore) GetUnprocessed(ctx context.Context) *Cursor

GetUnprocessed returns all the events in the store that have not been processed.

func (*FSStore) Init

func (s *FSStore) Init(ctx context.Context) error

Init goes through all the log files in the log dir and loads the entries into the store. It either creates a log file, or uses the last one if it exists.

func (*FSStore) MarkProcessed

func (s *FSStore) MarkProcessed(ctx context.Context, recs []IndexMarshaller) error

MarkProcessed marks the events as processed.

type IndexMarshaller

type IndexMarshaller interface {
	Key() string
	MarshalRecord(addField func(name string, value interface{}))
	UnmarshalRecord(data map[string]interface{}) error
}

IndexMarshaller is the interface for marshalling and unmarshalling index data.

type Options

type Options struct {
	LogDir     string
	LogFS      fs.FS
	OpenAppend func(path string) (io.WriteCloser, error)
}

Options hold the store configuration.

type Store

type Store interface {
	Init(context.Context) error

	AddDefaultField(string, interface{})

	Append(context.Context, IndexMarshaller) error

	Get(context.Context, string, IndexMarshaller) error
	GetAll(context.Context) *Cursor
	GetUnprocessed(context.Context) *Cursor

	MarkProcessed(context.Context, []IndexMarshaller) error
}

Store provides a generic append-only index of data.

type TestClosableBuffer

type TestClosableBuffer struct {
	bytes.Buffer
}

TestClosableBuffer wraps a bytes.Buffer and implements io.Closer.

func (TestClosableBuffer) Close

func (TestClosableBuffer) Close() error

Close implements io.Closer. Does nothing.