storage

Documentation

Index

• Constants
• type Config
  • func GetDefaultConfig() Config
• type Entry
  • func (e Entry) GetOrigin() string
  • func (e Entry) GetType() string
• type HttpConfig
• type LDB
  • func NewLDB(config LDBConfig) (*LDB, error)
  • func (ldb *LDB) Read(ctx context.Context, query Query) (<-chan Entry, <-chan error)
  • func (ldb *LDB) Start(ctx context.Context, wg *sync.WaitGroup) error
  • func (ldb *LDB) Store(bucket int64, entries []Entry) error
• type LDBConfig
• type Labels
• type Processor
  • func NewProcessor(config Config) (*Processor, error)
  • func (p Processor) GetStorage() Storage
  • func (p Processor) Process(entry ingest.Entry) error
  • func (p Processor) Start(ctx context.Context, wg *sync.WaitGroup) error
• type Query
• type RemoteConfig
• type RemoteStorage
  • func NewRemoteStorage(config RemoteConfig, localStorage Storage) (*RemoteStorage, error)
  • func (w *RemoteStorage) Start(ctx context.Context, wg *sync.WaitGroup) error
  • func (w *RemoteStorage) Write(bucket int64, entry Entry) error
• type Storage

Constants

const (
	EntBucketHeader    = "X-ent-bucket"
	EntTimestampHeader = "X-ent-timestamp"
	EntOriginHeader    = "X-ent-origin"
)

Types

type Config

type Config struct {
	Buckets      map[string]int64 `yaml:"buckets"`
	Type         string           `yaml:"type"`
	LDBConfig    LDBConfig        `yaml:"ldb"`
	RemoteConfig RemoteConfig     `yaml:"remote"`
}

func GetDefaultConfig

func GetDefaultConfig() Config

type Entry

type Entry struct {
	Timestamp int64
	Data      []byte
	Labels    Labels
}

func (Entry) GetOrigin

func (e Entry) GetOrigin() string

func (Entry) GetType

func (e Entry) GetType() string

type HttpConfig

type HttpConfig struct {
	Url            string `yaml:"url"`
	ConnectTimeout string `yaml:"connect_timeout"`
	RequestTimeout string `yaml:"request_timeout"`
}

type LDB

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

func NewLDB

func NewLDB(config LDBConfig) (*LDB, error)

func (*LDB) Read

func (ldb *LDB) Read(ctx context.Context, query Query) (<-chan Entry, <-chan error)

func (*LDB) Start

func (ldb *LDB) Start(ctx context.Context, wg *sync.WaitGroup) error

func (*LDB) Store

func (ldb *LDB) Store(bucket int64, entries []Entry) error

type LDBConfig

type LDBConfig struct {
	Path string `yaml:"path"`
	Sync bool   `yaml:"sync"`
}

type Labels

type Labels map[string]string

type Processor

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

func NewProcessor

func NewProcessor(config Config) (*Processor, error)

func (Processor) GetStorage

func (p Processor) GetStorage() Storage

func (Processor) Process

func (p Processor) Process(entry ingest.Entry) error

func (Processor) Start

func (p Processor) Start(ctx context.Context, wg *sync.WaitGroup) error

type Query

type Query struct {
	Bucket int64
	From   int64
	To     int64
}

type RemoteConfig

type RemoteConfig struct {
	Enabled    bool       `yaml:"enabled"`
	HttpConfig HttpConfig `yaml:"http"`
}

type RemoteStorage

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

func NewRemoteStorage

func NewRemoteStorage(config RemoteConfig, localStorage Storage) (*RemoteStorage, error)

func (*RemoteStorage) Start

func (w *RemoteStorage) Start(ctx context.Context, wg *sync.WaitGroup) error

func (*RemoteStorage) Write

func (w *RemoteStorage) Write(bucket int64, entry Entry) error

type Storage

type Storage interface {
	Store(bucket int64, entry []Entry) error
	Read(ctx context.Context, query Query) (<-chan Entry, <-chan error)
	Start(ctx context.Context, wg *sync.WaitGroup) error
}