prometheus/block.go

219 lines
4.4 KiB
Go
Raw Normal View History

package tsdb
2016-12-15 07:31:26 +00:00
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"sort"
2016-12-22 14:54:39 +00:00
"strconv"
"github.com/pkg/errors"
)
// Block handles reads against a block of time series data within a time window.
type block interface {
2016-12-13 14:26:58 +00:00
Querier(mint, maxt int64) Querier
interval() (int64, int64)
2016-12-13 14:26:58 +00:00
}
type BlockStats struct {
MinTime int64
MaxTime int64
SampleCount uint64
SeriesCount uint32
ChunkCount uint32
}
const (
flagNone = 0
flagStd = 1
)
2016-12-15 07:31:26 +00:00
type persistedBlock struct {
chunksf, indexf *mmapFile
chunks *seriesReader
index *indexReader
stats BlockStats
2016-12-15 07:31:26 +00:00
}
func newPersistedBlock(path string) (*persistedBlock, error) {
// TODO(fabxc): validate match of name and stats time, validate magic.
2016-12-15 07:31:26 +00:00
// mmap files belonging to the block.
chunksf, err := openMmapFile(chunksFileName(path))
if err != nil {
return nil, err
}
indexf, err := openMmapFile(indexFileName(path))
if err != nil {
return nil, err
}
sr, err := newSeriesReader(chunksf.b)
if err != nil {
return nil, err
}
ir, err := newIndexReader(sr, indexf.b)
if err != nil {
return nil, err
}
stats, err := ir.Stats()
if err != nil {
return nil, err
}
2016-12-15 07:31:26 +00:00
pb := &persistedBlock{
chunksf: chunksf,
indexf: indexf,
chunks: sr,
index: ir,
stats: stats,
2016-12-15 07:31:26 +00:00
}
return pb, nil
}
func (pb *persistedBlock) Close() error {
err0 := pb.chunksf.Close()
err1 := pb.indexf.Close()
if err0 != nil {
return err0
}
return err1
}
func (pb *persistedBlock) Querier(mint, maxt int64) Querier {
return &blockQuerier{
mint: mint,
maxt: maxt,
index: pb.index,
series: pb.chunks,
}
}
func (pb *persistedBlock) interval() (int64, int64) {
return pb.stats.MinTime, pb.stats.MaxTime
}
2016-12-15 07:31:26 +00:00
type persistedBlocks []*persistedBlock
func (p persistedBlocks) Len() int { return len(p) }
func (p persistedBlocks) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p persistedBlocks) Less(i, j int) bool { return p[i].stats.MinTime < p[j].stats.MinTime }
2016-12-15 07:31:26 +00:00
// findBlocks finds time-ordered persisted blocks within a directory.
2016-12-22 14:54:39 +00:00
func findBlocks(path string) ([]*persistedBlock, *HeadBlock, error) {
2016-12-15 07:31:26 +00:00
var pbs persistedBlocks
files, err := ioutil.ReadDir(path)
if err != nil {
2016-12-22 14:54:39 +00:00
return nil, nil, err
2016-12-15 07:31:26 +00:00
}
2016-12-22 14:54:39 +00:00
var head *HeadBlock
2016-12-15 07:31:26 +00:00
for _, fi := range files {
p := filepath.Join(path, fi.Name())
2016-12-22 14:54:39 +00:00
if _, err := os.Stat(chunksFileName(p)); os.IsNotExist(err) {
fmt.Println("found head dir", p)
if head != nil {
return nil, nil, errors.Errorf("found two head blocks")
}
ts, err := strconv.Atoi(filepath.Base(p))
if err != nil {
return nil, nil, errors.Errorf("invalid directory name")
}
2016-12-22 19:00:24 +00:00
head, err = OpenHeadBlock(p, int64(ts))
2016-12-22 14:54:39 +00:00
if err != nil {
return nil, nil, err
}
continue
}
pb, err := newPersistedBlock(p)
2016-12-15 07:31:26 +00:00
if err != nil {
2016-12-22 14:54:39 +00:00
return nil, nil, fmt.Errorf("error initializing block %q: %s", p, err)
2016-12-15 07:31:26 +00:00
}
pbs = append(pbs, pb)
}
// Order blocks by their base time so they represent a continous
// range of time.
sort.Sort(pbs)
2016-12-22 14:54:39 +00:00
return pbs, head, nil
2016-12-15 07:31:26 +00:00
}
func chunksFileName(path string) string {
return filepath.Join(path, "series")
}
func indexFileName(path string) string {
return filepath.Join(path, "index")
}
type mmapFile struct {
f *os.File
b []byte
}
func openMmapFile(path string) (*mmapFile, error) {
f, err := os.Open(path)
if err != nil {
return nil, err
}
info, err := f.Stat()
if err != nil {
return nil, err
}
b, err := mmap(f, int(info.Size()))
if err != nil {
return nil, err
}
return &mmapFile{f: f, b: b}, nil
}
func (f *mmapFile) Close() error {
err0 := munmap(f.b)
err1 := f.f.Close()
if err0 != nil {
return err0
}
return err1
}
// A skiplist maps offsets to values. The values found in the data at an
// offset are strictly greater than the indexed value.
type skiplist interface {
// offset returns the offset to data containing values of x and lower.
offset(x int64) (uint32, bool)
}
// simpleSkiplist is a slice of plain value/offset pairs.
type simpleSkiplist []skiplistPair
type skiplistPair struct {
value int64
offset uint32
}
func (sl simpleSkiplist) offset(x int64) (uint32, bool) {
// Search for the first offset that contains data greater than x.
i := sort.Search(len(sl), func(i int) bool { return sl[i].value >= x })
// If no element was found return false. If the first element is found,
// there's no previous offset actually containing values that are x or lower.
if i == len(sl) || i == 0 {
return 0, false
}
return sl[i-1].offset, true
}