stores

Documentation

Index

• Variables
• type ChunkFetcher
• type ChunkFetcherProvider
• type ChunkWriter
  • func NewChunkWriter(fetcher *fetcher.Fetcher, schemaCfg config.SchemaConfig, ...) ChunkWriter
• type CompositeStore
  • func NewCompositeStore(limits StoreLimits) *CompositeStore
  • func (c *CompositeStore) AddStore(start model.Time, fetcher *fetcher.Fetcher, index index.Reader, ...)
  • func (c CompositeStore) GetChunkFetcher(tm model.Time) *fetcher.Fetcher
  • func (c CompositeStore) GetChunks(ctx context.Context, userID string, from, through model.Time, ...) ([][]chunk.Chunk, []*fetcher.Fetcher, error)
  • func (c CompositeStore) GetSeries(ctx context.Context, userID string, from, through model.Time, ...) ([]labels.Labels, error)
  • func (c CompositeStore) GetShards(ctx context.Context, userID string, from, through model.Time, ...) (*logproto.ShardsResponse, error)
  • func (c CompositeStore) HasForSeries(from, through model.Time) (sharding.ForSeries, bool)
  • func (c CompositeStore) LabelNamesForMetricName(ctx context.Context, userID string, from, through model.Time, ...) ([]string, error)
  • func (c CompositeStore) LabelValuesForMetricName(ctx context.Context, userID string, from, through model.Time, ...) ([]string, error)
  • func (c CompositeStore) Put(ctx context.Context, chunks []chunk.Chunk) error
  • func (c CompositeStore) PutOne(ctx context.Context, from, through model.Time, chunk chunk.Chunk) error
  • func (c CompositeStore) SetChunkFilterer(chunkFilter chunk.RequestChunkFilterer)
  • func (c CompositeStore) Stats(ctx context.Context, userID string, from, through model.Time, ...) (*stats.Stats, error)
  • func (c CompositeStore) Stop()
  • func (c *CompositeStore) Stores() []Store
  • func (c CompositeStore) Volume(ctx context.Context, userID string, from, through model.Time, limit int32, ...) (*logproto.VolumeResponse, error)
• type Store
• type StoreLimits
• type Writer
  • func (c *Writer) Put(ctx context.Context, chunks []chunk.Chunk) error
  • func (c *Writer) PutOne(ctx context.Context, from, through model.Time, chk chunk.Chunk) error

Variables

var (
	DedupedChunksTotal = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: constants.Loki,
		Name:      "chunk_store_deduped_chunks_total",
		Help:      "Count of chunks which were not stored because they have already been stored by another replica.",
	})

	DedupedBytesTotal = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: constants.Loki,
		Name:      "chunk_store_deduped_bytes_total",
		Help:      "Count of bytes from chunks which were not stored because they have already been stored by another replica.",
	})

	IndexEntriesPerChunk = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: constants.Loki,
		Name:      "chunk_store_index_entries_per_chunk",
		Help:      "Number of entries written to storage per chunk.",
		Buckets:   prometheus.ExponentialBuckets(1, 2, 5),
	})
)

Types

type ChunkFetcher

type ChunkFetcher interface {
	GetChunks(
		ctx context.Context,
		userID string,
		from,
		through model.Time,
		predicate chunk.Predicate,
		storeChunksOverride *logproto.ChunkRefGroup,
	) ([][]chunk.Chunk, []*fetcher.Fetcher, error)
}

type ChunkFetcherProvider

type ChunkFetcherProvider interface {
	GetChunkFetcher(tm model.Time) *fetcher.Fetcher
}

type ChunkWriter

type ChunkWriter interface {
	Put(ctx context.Context, chunks []chunk.Chunk) error
	PutOne(ctx context.Context, from, through model.Time, chunk chunk.Chunk) error
}

func NewChunkWriter

func NewChunkWriter(fetcher *fetcher.Fetcher, schemaCfg config.SchemaConfig, indexWriter index.Writer, disableIndexDeduplication bool) ChunkWriter

type CompositeStore

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

CompositeStore is a Store which delegates to various stores depending on when they were activated.

func NewCompositeStore

func NewCompositeStore(limits StoreLimits) *CompositeStore

NewCompositeStore creates a new Store which delegates to different stores depending on time.

func (*CompositeStore) AddStore

func (c *CompositeStore) AddStore(start model.Time, fetcher *fetcher.Fetcher, index index.Reader, writer ChunkWriter, stop func())

func (CompositeStore) GetChunkFetcher

func (c CompositeStore) GetChunkFetcher(tm model.Time) *fetcher.Fetcher

func (CompositeStore) GetChunks

func (c CompositeStore) GetChunks(
	ctx context.Context,
	userID string,
	from,
	through model.Time,
	predicate chunk.Predicate,
	storeChunksOverride *logproto.ChunkRefGroup,
) ([][]chunk.Chunk, []*fetcher.Fetcher, error)

func (CompositeStore) GetSeries

func (c CompositeStore) GetSeries(ctx context.Context, userID string, from, through model.Time, matchers ...*labels.Matcher) ([]labels.Labels, error)

func (CompositeStore) GetShards

func (c CompositeStore) GetShards(
	ctx context.Context,
	userID string,
	from, through model.Time,
	targetBytesPerShard uint64,
	predicate chunk.Predicate,
) (*logproto.ShardsResponse, error)

func (CompositeStore) HasForSeries

func (c CompositeStore) HasForSeries(from, through model.Time) (sharding.ForSeries, bool)

func (CompositeStore) LabelNamesForMetricName

func (c CompositeStore) LabelNamesForMetricName(ctx context.Context, userID string, from, through model.Time, metricName string, matchers ...*labels.Matcher) ([]string, error)

LabelNamesForMetricName retrieves all label names for a metric name.

func (CompositeStore) LabelValuesForMetricName

func (c CompositeStore) LabelValuesForMetricName(ctx context.Context, userID string, from, through model.Time, metricName string, labelName string, matchers ...*labels.Matcher) ([]string, error)

LabelValuesForMetricName retrieves all label values for a single label name and metric name.

func (CompositeStore) Put

func (c CompositeStore) Put(ctx context.Context, chunks []chunk.Chunk) error

func (CompositeStore) PutOne

func (c CompositeStore) PutOne(ctx context.Context, from, through model.Time, chunk chunk.Chunk) error

func (CompositeStore) SetChunkFilterer

func (c CompositeStore) SetChunkFilterer(chunkFilter chunk.RequestChunkFilterer)

func (CompositeStore) Stats

func (c CompositeStore) Stats(ctx context.Context, userID string, from, through model.Time, matchers ...*labels.Matcher) (*stats.Stats, error)

func (CompositeStore) Stop

func (c CompositeStore) Stop()

func (*CompositeStore) Stores

func (c *CompositeStore) Stores() []Store

func (CompositeStore) Volume

func (c CompositeStore) Volume(ctx context.Context, userID string, from, through model.Time, limit int32, targetLabels []string, aggregateBy string, matchers ...*labels.Matcher) (*logproto.VolumeResponse, error)

type Store

type Store interface {
	index.BaseReader
	index.StatsReader
	index.Filterable
	ChunkWriter
	ChunkFetcher
	ChunkFetcherProvider
	Stop()
}

type StoreLimits

type StoreLimits interface {
	MaxChunksPerQueryFromStore(string) int
	MaxQueryLength(context.Context, string) time.Duration
}

type Writer

type Writer struct {
	DisableIndexDeduplication bool
	// contains filtered or unexported fields
}

func (*Writer) Put

func (c *Writer) Put(ctx context.Context, chunks []chunk.Chunk) error

Put implements Store

func (*Writer) PutOne

func (c *Writer) PutOne(ctx context.Context, from, through model.Time, chk chunk.Chunk) error

PutOne implements Store