diff --git a/block.go b/block.go index cd2da7bfa..7fedc1916 100644 --- a/block.go +++ b/block.go @@ -1,10 +1,12 @@ package tsdb -import "sort" - -const ( - magicIndex = 0xCAFECAFE - magicSeries = 0xAFFEAFFE +import ( + "fmt" + "io/ioutil" + "os" + "path/filepath" + "sort" + "strconv" ) // Block handles reads against a block of time series data within a time window. @@ -17,6 +19,132 @@ const ( flagStd = 1 ) +type persistedBlock struct { + chunksf, indexf *mmapFile + + chunks *seriesReader + index *indexReader + + baseTime int64 +} + +func newPersistedBlock(path string) (*persistedBlock, error) { + // The directory must be named after the base timestamp for the block. + baset, err := strconv.ParseInt(filepath.Base(path), 10, 0) + if err != nil { + return nil, fmt.Errorf("unexpected directory name %q: %s", path, err) + } + + // 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 + } + + pb := &persistedBlock{ + chunksf: chunksf, + indexf: indexf, + chunks: sr, + index: ir, + baseTime: baset, + } + return pb, nil +} + +func (pb *persistedBlock) Close() error { + err0 := pb.chunksf.Close() + err1 := pb.indexf.Close() + + if err0 != nil { + return err0 + } + return err1 +} + +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].baseTime < p[j].baseTime } + +// findBlocks finds time-ordered persisted blocks within a directory. +func findPersistedBlocks(path string) ([]*persistedBlock, error) { + var pbs persistedBlocks + + files, err := ioutil.ReadDir(path) + if err != nil { + return nil, err + } + + for _, fi := range files { + pb, err := newPersistedBlock(fi.Name()) + if err != nil { + return nil, fmt.Errorf("error initializing block %q: %s", fi.Name(), err) + } + pbs = append(pbs, pb) + } + + // Order blocks by their base time so they represent a continous + // range of time. + sort.Sort(pbs) + + return pbs, nil +} + +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 { diff --git a/db.go b/db.go index 8842978dd..b0f68eb23 100644 --- a/db.go +++ b/db.go @@ -7,9 +7,12 @@ import ( "os" "path/filepath" "sort" + "strconv" "sync" "time" + "golang.org/x/sync/errgroup" + "github.com/cespare/xxhash" "github.com/fabxc/tsdb/chunks" "github.com/go-kit/kit/log" @@ -65,9 +68,15 @@ func Open(path string, l log.Logger, opts *Options) (*DB, error) { // TODO(fabxc): validate shard number to be power of 2, which is required // for the bitshift-modulo when finding the right shard. for i := 0; i < numSeriesShards; i++ { - path := filepath.Join(path, fmt.Sprintf("%d", i)) l := log.NewContext(l).With("shard", i) - c.shards = append(c.shards, NewSeriesShard(path, l)) + d := shardDir(path, i) + + s, err := NewSeriesShard(d, l) + if err != nil { + return nil, fmt.Errorf("initializing shard %q failed: %s", d, err) + } + + c.shards = append(c.shards, s) } // TODO(fabxc): run background compaction + GC. @@ -75,23 +84,21 @@ func Open(path string, l log.Logger, opts *Options) (*DB, error) { return c, nil } +func shardDir(base string, i int) string { + return filepath.Join(base, strconv.Itoa(i)) +} + // Close the database. func (db *DB) Close() error { - var wg sync.WaitGroup + var g errgroup.Group - for i, shard := range db.shards { - wg.Add(1) - go func(i int, shard *SeriesShard) { - if err := shard.Close(); err != nil { - // TODO(fabxc): handle with multi error. - panic(err) - } - wg.Done() - }(i, shard) + for _, shard := range db.shards { + // Fix closure argument to goroutine. + shard := shard + g.Go(shard.Close) } - wg.Wait() - return nil + return g.Wait() } // Appender adds a batch of samples. @@ -198,31 +205,50 @@ type SeriesShard struct { persistCh chan struct{} logger log.Logger - mtx sync.RWMutex - blocks []*Block - head *HeadBlock + mtx sync.RWMutex + persisted persistedBlocks + head *HeadBlock } // NewSeriesShard returns a new SeriesShard. -func NewSeriesShard(path string, logger log.Logger) *SeriesShard { +func NewSeriesShard(path string, logger log.Logger) (*SeriesShard, error) { + // Create directory if shard is new. + if _, err := os.Stat(path); os.IsNotExist(err) { + if err := os.MkdirAll(path, 0777); err != nil { + return nil, err + } + } + + // Initialize previously persisted blocks. + pbs, err := findPersistedBlocks(path) + if err != nil { + return nil, err + } + s := &SeriesShard{ path: path, persistCh: make(chan struct{}, 1), logger: logger, + persisted: pbs, // TODO(fabxc): restore from checkpoint. - // TODO(fabxc): provide access to persisted blocks. } // TODO(fabxc): get base time from pre-existing blocks. Otherwise // it should come from a user defined start timestamp. // Use actual time for now. s.head = NewHeadBlock(time.Now().UnixNano() / int64(time.Millisecond)) - return s + return s, nil } // Close the series shard. func (s *SeriesShard) Close() error { - return nil + var e MultiError + + for _, pb := range s.persisted { + e.Add(pb.Close()) + } + + return e.Err() } func (s *SeriesShard) appendBatch(ts int64, samples []Sample) error { @@ -463,3 +489,16 @@ func (es MultiError) Error() string { return buf.String() } + +func (es MultiError) Add(err error) { + if err != nil { + es = append(es, err) + } +} + +func (es MultiError) Err() error { + if len(es) == 0 { + return nil + } + return es +}