prometheus/block.go

169 lines
3.5 KiB
Go
Raw Normal View History

package tsdb
2016-12-15 07:31:26 +00:00
import (
"os"
"path/filepath"
"sort"
2017-01-07 17:02:17 +00:00
"sync"
2016-12-22 14:54:39 +00:00
2017-01-03 09:09:20 +00:00
"github.com/coreos/etcd/pkg/fileutil"
2016-12-22 14:54:39 +00:00
"github.com/pkg/errors"
)
// Block handles reads against a Block of time series data.
type Block interface {
Dir() string
Stats() BlockStats
Index() IndexReader
Series() SeriesReader
Persisted() bool
2016-12-13 14:26:58 +00:00
}
// BlockStats provides stats on a data block.
type BlockStats struct {
MinTime, MaxTime int64 // time range of samples in the block
2017-01-07 17:02:17 +00:00
SampleCount uint64
2017-01-07 17:02:17 +00:00
SeriesCount uint64
ChunkCount uint64
mtx sync.RWMutex
}
const (
flagNone = 0
flagStd = 1
)
2016-12-15 07:31:26 +00:00
type persistedBlock struct {
dir string
stats *BlockStats
2017-01-03 09:09:20 +00:00
2016-12-15 07:31:26 +00:00
chunksf, indexf *mmapFile
chunkr *seriesReader
indexr *indexReader
2016-12-15 07:31:26 +00:00
}
func newPersistedBlock(dir 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(dir))
2016-12-15 07:31:26 +00:00
if err != nil {
2017-01-03 14:43:26 +00:00
return nil, errors.Wrap(err, "open chunk file")
2016-12-15 07:31:26 +00:00
}
indexf, err := openMmapFile(indexFileName(dir))
2016-12-15 07:31:26 +00:00
if err != nil {
2017-01-03 14:43:26 +00:00
return nil, errors.Wrap(err, "open index file")
2016-12-15 07:31:26 +00:00
}
sr, err := newSeriesReader(chunksf.b)
if err != nil {
2017-01-03 14:43:26 +00:00
return nil, errors.Wrap(err, "create series reader")
2016-12-15 07:31:26 +00:00
}
ir, err := newIndexReader(sr, indexf.b)
if err != nil {
2017-01-03 14:43:26 +00:00
return nil, errors.Wrap(err, "create index reader")
2016-12-15 07:31:26 +00:00
}
stats, err := ir.Stats()
if err != nil {
2017-01-03 14:43:26 +00:00
return nil, errors.Wrap(err, "read stats")
}
2016-12-15 07:31:26 +00:00
pb := &persistedBlock{
dir: dir,
chunksf: chunksf,
indexf: indexf,
chunkr: sr,
indexr: 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) Dir() string { return pb.dir }
func (pb *persistedBlock) Persisted() bool { return true }
func (pb *persistedBlock) Index() IndexReader { return pb.indexr }
func (pb *persistedBlock) Series() SeriesReader { return pb.chunkr }
func (pb *persistedBlock) Stats() BlockStats { return *pb.stats }
2016-12-15 07:31:26 +00:00
func chunksFileName(path string) string {
return filepath.Join(path, "chunks-000")
2016-12-15 07:31:26 +00:00
}
func indexFileName(path string) string {
return filepath.Join(path, "index-000")
2016-12-15 07:31:26 +00:00
}
type mmapFile struct {
2017-01-03 09:09:20 +00:00
f *fileutil.LockedFile
2016-12-15 07:31:26 +00:00
b []byte
}
func openMmapFile(path string) (*mmapFile, error) {
2017-01-03 09:09:20 +00:00
f, err := fileutil.TryLockFile(path, os.O_RDONLY, 0666)
2016-12-15 07:31:26 +00:00
if err != nil {
return nil, err
}
info, err := f.Stat()
if err != nil {
return nil, err
}
2017-01-03 09:09:20 +00:00
b, err := mmap(f.File, int(info.Size()))
2016-12-15 07:31:26 +00:00
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
}