chunkenc

README

Chunk format

  |                 |             |
  | MagicNumber(4b) | version(1b) |
  |                 |             |
  --------------------------------------------------
  |         block-1 bytes         |  checksum (4b) |
  --------------------------------------------------
  |         block-2 bytes         |  checksum (4b) |
  --------------------------------------------------
  |         block-n bytes         |  checksum (4b) |
  --------------------------------------------------
  |         #blocks (uvarint)                      |
  --------------------------------------------------
  | #entries(uvarint) | mint, maxt (varint) | offset, len (uvarint) |
  -------------------------------------------------------------------
  | #entries(uvarint) | mint, maxt (varint) | offset, len (uvarint) |
  -------------------------------------------------------------------
  | #entries(uvarint) | mint, maxt (varint) | offset, len (uvarint) |
  -------------------------------------------------------------------
  | #entries(uvarint) | mint, maxt (varint) | offset, len (uvarint) |
  -------------------------------------------------------------------
  |                      checksum(from #blocks)                     |
  -------------------------------------------------------------------
  | metasOffset - offset to the point with #blocks |
  --------------------------------------------------

Documentation

Index

• Constants
• Variables
• func NewFacade(c Chunk) encoding.Chunk
• type BufioReaderPool
  • func (bufPool *BufioReaderPool) Get(r io.Reader) *bufio.Reader
  • func (bufPool *BufioReaderPool) Put(b *bufio.Reader)
• type Chunk
  • func NewDumbChunk() Chunk
• type CompressionPool
• type CompressionReader
• type CompressionWriter
• type Encoding
  • func (e Encoding) String() string
• type Facade
  • func (Facade) Encoding() encoding.Encoding
  • func (f Facade) LokiChunk() Chunk
  • func (f Facade) Marshal(w io.Writer) error
  • func (f *Facade) UnmarshalFromBuf(buf []byte) error
• type GzipPool
  • func (pool *GzipPool) GetReader(src io.Reader) (reader CompressionReader)
  • func (pool *GzipPool) GetWriter(dst io.Writer) (writer CompressionWriter)
  • func (pool *GzipPool) PutReader(reader CompressionReader)
  • func (pool *GzipPool) PutWriter(writer CompressionWriter)
• type LazyChunk
  • func (c *LazyChunk) Iterator(ctx context.Context, from, through time.Time, direction logproto.Direction, ...) (iter.EntryIterator, error)
• type MemChunk
  • func NewByteChunk(b []byte) (*MemChunk, error)
  • func NewMemChunk(enc Encoding) *MemChunk
  • func NewMemChunkSize(enc Encoding, blockSize int) *MemChunk
  • func (c *MemChunk) Append(entry *logproto.Entry) error
  • func (c *MemChunk) Bounds() (fromT, toT time.Time)
  • func (c *MemChunk) Bytes() ([]byte, error)
  • func (c *MemChunk) Close() error
  • func (c *MemChunk) Encoding() Encoding
  • func (c *MemChunk) Iterator(mintT, maxtT time.Time, direction logproto.Direction, filter logql.Filter) (iter.EntryIterator, error)
  • func (c *MemChunk) Size() int
  • func (c *MemChunk) SpaceFor(*logproto.Entry) bool

Constants

const GzipLogChunk = encoding.Encoding(128)

GzipLogChunk is a cortex encoding type for our chunks.

Variables

var (
	ErrChunkFull       = errors.New("chunk full")
	ErrOutOfOrder      = errors.New("entry out of order")
	ErrInvalidSize     = errors.New("invalid size")
	ErrInvalidFlag     = errors.New("invalid flag")
	ErrInvalidChecksum = errors.New("invalid checksum")
)

Errors returned by the chunk interface.

var (
	// Gzip is the gun zip compression pool
	Gzip GzipPool
	// BufReaderPool is bufio.Reader pool
	BufReaderPool = &BufioReaderPool{
		pool: sync.Pool{
			New: func() interface{} { return bufio.NewReader(nil) },
		},
	}
	// BytesBufferPool is a bytes buffer used for lines decompressed.
	// Buckets [0.5KB,1KB,2KB,4KB,8KB]
	BytesBufferPool = pool.New(1<<9, 1<<13, 2, func(size int) interface{} { return make([]byte, 0, size) })
)

Functions

func NewFacade

func NewFacade(c Chunk) encoding.Chunk

NewFacade makes a new Facade.

Types

type BufioReaderPool

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

BufioReaderPool is a bufio reader that uses sync.Pool.

func (*BufioReaderPool) Get

func (bufPool *BufioReaderPool) Get(r io.Reader) *bufio.Reader

Get returns a bufio.Reader which reads from r. The buffer size is that of the pool.

func (*BufioReaderPool) Put

func (bufPool *BufioReaderPool) Put(b *bufio.Reader)

Put puts the bufio.Reader back into the pool.

type Chunk

type Chunk interface {
	Bounds() (time.Time, time.Time)
	SpaceFor(*logproto.Entry) bool
	Append(*logproto.Entry) error
	Iterator(from, through time.Time, direction logproto.Direction, filter logql.Filter) (iter.EntryIterator, error)
	Size() int
	Bytes() ([]byte, error)
}

Chunk is the interface for the compressed logs chunk format.

func NewDumbChunk

func NewDumbChunk() Chunk

NewDumbChunk returns a new chunk that isn't very good.

type CompressionPool

type CompressionPool interface {
	GetWriter(io.Writer) CompressionWriter
	PutWriter(CompressionWriter)
	GetReader(io.Reader) CompressionReader
	PutReader(CompressionReader)
}

CompressionPool is a pool of CompressionWriter and CompressionReader This is used by every chunk to avoid unnecessary allocations.

type CompressionReader

type CompressionReader interface {
	Read(p []byte) (int, error)
	Reset(r io.Reader) error
}

CompressionReader reads the compressed data.

type CompressionWriter

type CompressionWriter interface {
	Write(p []byte) (int, error)
	Close() error
	Flush() error
	Reset(w io.Writer)
}

CompressionWriter is the writer that compresses the data passed to it.

type Encoding

type Encoding uint8

Encoding is the identifier for a chunk encoding.

const (
	EncNone Encoding = iota
	EncGZIP
	EncDumb
)

The different available encodings.

func (Encoding) String

func (e Encoding) String() string

type Facade

type Facade struct {
	encoding.Chunk
	// contains filtered or unexported fields
}

Facade for compatibility with cortex chunk type, so we can use its chunk store.

func (Facade) Encoding

func (Facade) Encoding() encoding.Encoding

Encoding implements encoding.Chunk.

func (Facade) LokiChunk

func (f Facade) LokiChunk() Chunk

LokiChunk returns the chunkenc.Chunk.

func (Facade) Marshal

func (f Facade) Marshal(w io.Writer) error

Marshal implements encoding.Chunk.

func (*Facade) UnmarshalFromBuf

func (f *Facade) UnmarshalFromBuf(buf []byte) error

UnmarshalFromBuf implements encoding.Chunk.

type GzipPool

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

GzipPool is a gun zip compression pool

func (*GzipPool) GetReader

func (pool *GzipPool) GetReader(src io.Reader) (reader CompressionReader)

GetReader gets or creates a new CompressionReader and reset it to read from src

func (*GzipPool) GetWriter

func (pool *GzipPool) GetWriter(dst io.Writer) (writer CompressionWriter)

GetWriter gets or creates a new CompressionWriter and reset it to write to dst

func (*GzipPool) PutReader

func (pool *GzipPool) PutReader(reader CompressionReader)

PutReader places back in the pool a CompressionReader

func (*GzipPool) PutWriter

func (pool *GzipPool) PutWriter(writer CompressionWriter)

PutWriter places back in the pool a CompressionWriter

type LazyChunk

type LazyChunk struct {
	Chunk   chunk.Chunk
	Fetcher *chunk.Fetcher
}

LazyChunk loads the chunk when it is accessed.

func (*LazyChunk) Iterator

func (c *LazyChunk) Iterator(ctx context.Context, from, through time.Time, direction logproto.Direction, filter logql.Filter) (iter.EntryIterator, error)

Iterator returns an entry iterator.

type MemChunk

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

MemChunk implements compressed log chunks.

func NewByteChunk

func NewByteChunk(b []byte) (*MemChunk, error)

NewByteChunk returns a MemChunk on the passed bytes.

func NewMemChunk

func NewMemChunk(enc Encoding) *MemChunk

NewMemChunk returns a new in-mem chunk for query.

func NewMemChunkSize

func NewMemChunkSize(enc Encoding, blockSize int) *MemChunk

NewMemChunkSize returns a new in-mem chunk. Mainly for config push size.

func (*MemChunk) Append

func (c *MemChunk) Append(entry *logproto.Entry) error

Append implements Chunk.

func (*MemChunk) Bounds

func (c *MemChunk) Bounds() (fromT, toT time.Time)

Bounds implements Chunk.

func (*MemChunk) Bytes

func (c *MemChunk) Bytes() ([]byte, error)

Bytes implements Chunk.

func (*MemChunk) Close

func (c *MemChunk) Close() error

Close implements Chunk. TODO: Fix this to check edge cases.

func (*MemChunk) Encoding

func (c *MemChunk) Encoding() Encoding

Encoding implements Chunk.

func (*MemChunk) Iterator

func (c *MemChunk) Iterator(mintT, maxtT time.Time, direction logproto.Direction, filter logql.Filter) (iter.EntryIterator, error)

Iterator implements Chunk.

func (*MemChunk) Size

func (c *MemChunk) Size() int

Size implements Chunk.

func (*MemChunk) SpaceFor

func (c *MemChunk) SpaceFor(*logproto.Entry) bool

SpaceFor implements Chunk.