indexer

Documentation

Overview

Package indexer processing logs and stores them for querying. This is probably confusing several concerns and needs work.

On start up, each indexer is assigned a new UUID. The indexer creates shards, collections of shard files that span a liited time duration (shardDuration). Each shard created by an indexer is assigned a UUID. Each incoming stream is assigned a UUID. A sharfile is created for each stream. Each message in a stream is assigned its ordinal position in the stream

The dataDir should be laid out as follows: dataDir/shardID/streamID.pb.log

Index

• func NewArchive(opts ...ArchiveOpt) (*shardArchive, error)
• func NewSimpleJSONHandler(src GrafanaSimpleJSON, opts ...sjcOpt) *sjcSrc
• func WithGrafanaBasicAuth(user, pass string) sjcOpt
• type Annotation
• type ArchiveOpt
  • func WithArchiveDataDir(datadir string) ArchiveOpt
  • func WithArchiveEncryptTo(ent []openpgp.Entity) ArchiveOpt
  • func WithArchiveGzipCompression(level int) ArchiveOpt
  • func WithArchiveRetention(d time.Duration) ArchiveOpt
  • func WithArchiveStowConfig(scfg stow.ConfigMap) ArchiveOpt
• type Data
• type GrafanaSimpleJSON
• type Indexer
  • func New(opts ...Opt) (*Indexer, error)
  • func (idx *Indexer) AddSource(ctx context.Context, id string, labels map[string]string) (sinks.MessageWriter, error)
  • func (idx *Indexer) Close() error
  • func (i *Indexer) Collect(ch chan<- prometheus.Metric)
  • func (i *Indexer) Describe(ch chan<- *prometheus.Desc)
  • func (idx *Indexer) GrafanaAnnotations(ctx context.Context, from, to time.Time, query string) ([]Annotation, error)
  • func (idx *Indexer) GrafanaQuery(ctx context.Context, from, to time.Time, interval time.Duration, maxDPs int, ...) (map[string][]Data, error)
  • func (idx *Indexer) GrafanaSearch(ctx context.Context, target string) ([]string, error)
  • func (idx *Indexer) LabelValues(name string, from, to time.Time, count int) ([]string, int, error)
  • func (idx *Indexer) Labels(from, to time.Time) ([]string, error)
  • func (idx *Indexer) Search(ctx context.Context, msgFunc logspray.MessageFunc, matcher ql.MatchFunc, ...) error
• type MessageWriter
  • func (w *MessageWriter) Close() error
  • func (w *MessageWriter) WriteMessage(ctx context.Context, m *logspray.Message) error
• type Opt
  • func WithDataDir(d string) Opt
  • func WithSharDuration(d time.Duration) Opt
• type Shard
  • func (s *Shard) LabelValues(name string) []string
  • func (s *Shard) Labels() []string
  • func (s *Shard) Search(ctx context.Context, msgFunc logspray.MessageFunc, matcher ql.MatchFunc, ...) error
• type ShardFile
  • func (s *ShardFile) Close() error
  • func (s *ShardFile) Search(ctx context.Context, msgFunc logspray.MessageFunc, matcher ql.MatchFunc, ...) error
• type ShardFileByStartTime
  • func (so ShardFileByStartTime) Len() int
  • func (so ShardFileByStartTime) Less(i, j int) bool
  • func (so ShardFileByStartTime) Swap(i, j int)

Functions

func NewArchive

func NewArchive(opts ...ArchiveOpt) (*shardArchive, error)

func NewSimpleJSONHandler

func NewSimpleJSONHandler(src GrafanaSimpleJSON, opts ...sjcOpt) *sjcSrc

NewSimpleJSONHandler returns a HTTP handler that uses data from src to serve Grafana SimpleJSON queries.

func WithGrafanaBasicAuth

func WithGrafanaBasicAuth(user, pass string) sjcOpt

WithGrafanaBasicAuth sets basic auth params for the simpleJson endpoint

Types

type Annotation

type Annotation struct {
	Time  simpleJSONDPTime `json:"time"`
	Title string           `json:"title"`
	Text  string           `json:"text"`
	Tags  []string         `json:"tags"`
}

Annotation describes a Grafana Simple JSON annotation

type ArchiveOpt

type ArchiveOpt func(a *shardArchive) (*shardArchive, error)

func WithArchiveDataDir

func WithArchiveDataDir(datadir string) ArchiveOpt

func WithArchiveEncryptTo

func WithArchiveEncryptTo(ent []openpgp.Entity) ArchiveOpt

func WithArchiveGzipCompression

func WithArchiveGzipCompression(level int) ArchiveOpt

func WithArchiveRetention

func WithArchiveRetention(d time.Duration) ArchiveOpt

func WithArchiveStowConfig

func WithArchiveStowConfig(scfg stow.ConfigMap) ArchiveOpt

type Data

type Data struct {
	Time  time.Time
	Value float64
}

Data represents a Grafana Simple JSON data point

type GrafanaSimpleJSON

type GrafanaSimpleJSON interface {
	GrafanaQuery(ctx context.Context, from, to time.Time, interval time.Duration, maxDPs int, targets []string) (map[string][]Data, error)
	GrafanaAnnotations(ctx context.Context, from, to time.Time, query string) ([]Annotation, error)
	GrafanaSearch(ctx context.Context, target string) ([]string, error)
}

GrafanaSimpleJSON describes requirements for serving a Grafana Simple JSON endpoint.

type Indexer

type Indexer struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Indexer implements a queryable index for storage of logspray mssages.

func New

func New(opts ...Opt) (*Indexer, error)

New creates a new index.

func (*Indexer) AddSource

func (idx *Indexer) AddSource(ctx context.Context, id string, labels map[string]string) (sinks.MessageWriter, error)

AddSource adds a new source to the remote server

func (*Indexer) Close

func (idx *Indexer) Close() error

Close this index

func (*Indexer) Collect

func (i *Indexer) Collect(ch chan<- prometheus.Metric)

Collect implements the prom metrics collection Collector interface

func (*Indexer) Describe

func (i *Indexer) Describe(ch chan<- *prometheus.Desc)

Describe implements the prometheus describe interfaces for metric collection

func (*Indexer) GrafanaAnnotations

func (idx *Indexer) GrafanaAnnotations(ctx context.Context, from, to time.Time, query string) ([]Annotation, error)

GrafanaAnnotations implements the grafana Simple JSON Annotations request

func (*Indexer) GrafanaQuery

func (idx *Indexer) GrafanaQuery(ctx context.Context, from, to time.Time, interval time.Duration, maxDPs int, targets []string) (map[string][]Data, error)

GrafanaQuery implements the Grafana Simple JSON Query request

func (*Indexer) GrafanaSearch

func (idx *Indexer) GrafanaSearch(ctx context.Context, target string) ([]string, error)

GrafanaSearch implements the Grafana Simple JSON search query

func (*Indexer) LabelValues

func (idx *Indexer) LabelValues(name string, from, to time.Time, count int) ([]string, int, error)

LabelValues returns all the known values for a given label.

func (*Indexer) Labels

func (idx *Indexer) Labels(from, to time.Time) ([]string, error)

Labels lists all the label names in the current index

func (*Indexer) Search

func (idx *Indexer) Search(ctx context.Context, msgFunc logspray.MessageFunc, matcher ql.MatchFunc, from, to time.Time, count, offset uint64, reverse bool) error

Search queries the index for documents matching the provided search query.

type MessageWriter

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

MessageWriter is a sinks.MessageWriter that writes to a remote server

func (*MessageWriter) Close

func (w *MessageWriter) Close() error

Close closes the remote stream.

func (*MessageWriter) WriteMessage

func (w *MessageWriter) WriteMessage(ctx context.Context, m *logspray.Message) error

WriteMessage writes a message to the log stream.

type Opt

type Opt func(i *Indexer) error

Opt defines an index option function.

func WithDataDir

func WithDataDir(d string) Opt

WithDataDir lets you set the base filesystem path to store data to.

func WithSharDuration

func WithSharDuration(d time.Duration) Opt

WithSharDuration allows you to set the time duration of a shard.

type Shard

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

Shard represents a logging session.

func (*Shard) LabelValues

func (s *Shard) LabelValues(name string) []string

LabelValues returns all the known values for a given label.

func (*Shard) Labels

func (s *Shard) Labels() []string

Labels retursn the list of labels in the label cache

func (*Shard) Search

func (s *Shard) Search(ctx context.Context, msgFunc logspray.MessageFunc, matcher ql.MatchFunc, from, to time.Time, count, offset uint64, reverse bool) error

Search this shard for queries between the provided time frames for message matched by the provided match function.

type ShardFile

type ShardFile struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

ShardFile represents an individual stream of data in a shard.

func (*ShardFile) Close

func (s *ShardFile) Close() error

Close the backing files for a shard

func (*ShardFile) Search

func (s *ShardFile) Search(ctx context.Context, msgFunc logspray.MessageFunc, matcher ql.MatchFunc, from, to time.Time, count, offset uint64, reverse bool) error

Search searches the shard file for messages in the provided time range, matched by matcher, and passes them to msgFunc. If the reverse is true the file will be searched in reverse order

type ShardFileByStartTime

type ShardFileByStartTime []*ShardFile

func (ShardFileByStartTime) Len

func (so ShardFileByStartTime) Len() int

func (ShardFileByStartTime) Less

func (so ShardFileByStartTime) Less(i, j int) bool

func (ShardFileByStartTime) Swap

func (so ShardFileByStartTime) Swap(i, j int)