Add loading of persisted blocks

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 {

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
+}