prometheus/head.go

184 lines
3.9 KiB
Go
Raw Normal View History

2016-12-04 04:16:11 -08:00
package tsdb
import (
"math"
2016-12-19 13:37:03 -08:00
"os"
"sort"
2016-12-04 04:16:11 -08:00
"sync"
"github.com/fabxc/tsdb/chunks"
2016-12-21 00:39:01 -08:00
"github.com/fabxc/tsdb/labels"
2016-12-04 04:16:11 -08:00
)
// HeadBlock handles reads and writes of time series data within a time window.
type HeadBlock struct {
mtx sync.RWMutex
descs map[uint64][]*chunkDesc // labels hash to possible chunks descs
index *memIndex
stats BlockStats
2016-12-04 04:16:11 -08:00
}
2016-12-09 04:41:38 -08:00
// NewHeadBlock creates a new empty head block.
func NewHeadBlock(baseTime int64) *HeadBlock {
b := &HeadBlock{
descs: make(map[uint64][]*chunkDesc, 2048),
index: newMemIndex(),
}
b.stats.MinTime = baseTime
return b
}
// Querier returns a new querier over the head block.
func (h *HeadBlock) Querier(mint, maxt int64) Querier {
return newBlockQuerier(h, h, mint, maxt)
}
// Chunk returns the chunk for the reference number.
func (h *HeadBlock) Chunk(ref uint32) (chunks.Chunk, error) {
c, ok := h.index.forward[ref]
if !ok {
return nil, errNotFound
}
return c.chunk, nil
}
func (h *HeadBlock) interval() (int64, int64) {
return h.stats.MinTime, h.stats.MaxTime
}
// Stats returns statisitics about the indexed data.
func (h *HeadBlock) Stats() (BlockStats, error) {
return h.stats, nil
}
// LabelValues returns the possible label values
func (h *HeadBlock) LabelValues(names ...string) (StringTuples, error) {
if len(names) != 1 {
return nil, errInvalidSize
}
var sl []string
for s := range h.index.values[names[0]] {
sl = append(sl, s)
}
sort.Strings(sl)
t := &stringTuples{
l: len(names),
s: sl,
}
return t, nil
}
// Postings returns the postings list iterator for the label pair.
2016-12-14 12:58:29 -08:00
func (h *HeadBlock) Postings(name, value string) (Postings, error) {
return h.index.Postings(term{name, value}), nil
}
// Series returns the series for the given reference.
func (h *HeadBlock) Series(ref uint32, mint, maxt int64) (Series, error) {
cd, ok := h.index.forward[ref]
if !ok {
return nil, errNotFound
}
if !intervalOverlap(cd.firsTimestamp, cd.lastTimestamp, mint, maxt) {
return nil, nil
}
s := &chunkSeries{
labels: cd.lset,
chunks: []ChunkMeta{
{MinTime: h.stats.MinTime, Ref: 0},
},
chunk: func(ref uint32) (chunks.Chunk, error) {
return cd.chunk, nil
},
}
return s, nil
}
2016-12-04 04:16:11 -08:00
// get retrieves the chunk with the hash and label set and creates
// a new one if it doesn't exist yet.
2016-12-21 00:39:01 -08:00
func (h *HeadBlock) get(hash uint64, lset labels.Labels) *chunkDesc {
2016-12-04 04:16:11 -08:00
cds := h.descs[hash]
for _, cd := range cds {
if cd.lset.Equals(lset) {
2016-12-09 01:00:14 -08:00
return cd
2016-12-04 04:16:11 -08:00
}
}
// None of the given chunks was for the series, create a new one.
cd := &chunkDesc{
lset: lset,
chunk: chunks.NewXORChunk(int(math.MaxInt64)),
}
h.index.add(cd)
2016-12-04 04:16:11 -08:00
// For the head block there's exactly one chunk per series.
h.stats.ChunkCount++
h.stats.SeriesCount++
2016-12-04 04:16:11 -08:00
h.descs[hash] = append(cds, cd)
2016-12-09 01:00:14 -08:00
return cd
2016-12-04 04:16:11 -08:00
}
2016-12-09 01:00:14 -08:00
// append adds the sample to the headblock.
2016-12-21 00:39:01 -08:00
func (h *HeadBlock) append(hash uint64, lset labels.Labels, ts int64, v float64) error {
2016-12-09 01:00:14 -08:00
if err := h.get(hash, lset).append(ts, v); err != nil {
return err
}
h.stats.SampleCount++
if ts > h.stats.MaxTime {
h.stats.MaxTime = ts
2016-12-09 01:00:14 -08:00
}
return nil
}
2016-12-19 13:37:03 -08:00
func (h *HeadBlock) persist(p string) (int64, error) {
sf, err := os.Create(chunksFileName(p))
if err != nil {
return 0, err
}
xf, err := os.Create(indexFileName(p))
if err != nil {
return 0, err
}
iw := newIndexWriter(xf)
sw := newSeriesWriter(sf, iw, h.stats.MinTime)
defer sw.Close()
defer iw.Close()
for ref, cd := range h.index.forward {
if err := sw.WriteSeries(ref, cd.lset, []*chunkDesc{cd}); err != nil {
2016-12-19 13:37:03 -08:00
return 0, err
}
}
if err := iw.WriteStats(h.stats); err != nil {
2016-12-19 13:37:03 -08:00
return 0, err
}
for n, v := range h.index.values {
s := make([]string, 0, len(v))
for x := range v {
s = append(s, x)
}
if err := iw.WriteLabelIndex([]string{n}, s); err != nil {
2016-12-19 13:37:03 -08:00
return 0, err
}
}
for t := range h.index.postings.m {
if err := iw.WritePostings(t.name, t.value, h.index.postings.get(t)); err != nil {
2016-12-19 13:37:03 -08:00
return 0, err
}
}
2016-12-19 13:37:03 -08:00
return iw.Size() + sw.Size(), nil
}