influxdb/tsdb/engine/tsm1/file_store.go

930 lines
22 KiB
Go

package tsm1
import (
"expvar"
"fmt"
"log"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/influxdata/influxdb"
)
type TSMFile interface {
// Path returns the underlying file path for the TSMFile. If the file
// has not be written or loaded from disk, the zero value is returne.
Path() string
// Read returns all the values in the block where time t resides
Read(key string, t time.Time) ([]Value, error)
// ReadAt returns all the values in the block identified by entry.
ReadAt(entry *IndexEntry, values []Value) ([]Value, error)
ReadFloatBlockAt(entry *IndexEntry, values []FloatValue) ([]FloatValue, error)
ReadIntegerBlockAt(entry *IndexEntry, values []IntegerValue) ([]IntegerValue, error)
ReadStringBlockAt(entry *IndexEntry, values []StringValue) ([]StringValue, error)
ReadBooleanBlockAt(entry *IndexEntry, values []BooleanValue) ([]BooleanValue, error)
// Entries returns the index entries for all blocks for the given key.
Entries(key string) []*IndexEntry
// Returns true if the TSMFile may contain a value with the specified
// key and time
ContainsValue(key string, t time.Time) bool
// Contains returns true if the file contains any values for the given
// key.
Contains(key string) bool
// TimeRange returns the min and max time across all keys in the file.
TimeRange() (time.Time, time.Time)
// KeyRange returns the min and max keys in the file.
KeyRange() (string, string)
// Keys returns all keys contained in the file.
Keys() []string
// Type returns the block type of the values stored for the key. Returns one of
// BlockFloat64, BlockInt64, BlockBoolean, BlockString. If key does not exist,
// an error is returned.
Type(key string) (byte, error)
// Delete removes the keys from the set of keys available in this file.
Delete(keys []string) error
// HasTombstones returns true if file contains values that have been deleted.
HasTombstones() bool
// TombstoneFiles returns the tombstone filestats if there are any tombstones
// written for this file.
TombstoneFiles() []FileStat
// Close the underlying file resources
Close() error
// Size returns the size of the file on disk in bytes.
Size() uint32
// Remove deletes the file from the filesystem
Remove() error
// Stats returns summary information about the TSM file.
Stats() FileStat
// BlockIterator returns an iterator pointing to the first block in the file and
// allows sequential iteration to each every block.
BlockIterator() *BlockIterator
}
// Statistics gathered by the FileStore.
const (
statFileStoreBytes = "diskBytes"
)
type FileStore struct {
mu sync.RWMutex
lastModified time.Time
currentGeneration int
dir string
files []TSMFile
Logger *log.Logger
traceLogging bool
statMap *expvar.Map
}
type FileStat struct {
Path string
HasTombstone bool
Size uint32
LastModified time.Time
MinTime, MaxTime time.Time
MinKey, MaxKey string
}
func (f FileStat) OverlapsTimeRange(min, max time.Time) bool {
return (f.MinTime.Equal(max) || f.MinTime.Before(max)) &&
(f.MaxTime.Equal(min) || f.MaxTime.After(min))
}
func (f FileStat) OverlapsKeyRange(min, max string) bool {
return min != "" && max != "" && f.MinKey <= max && f.MaxKey >= min
}
func (f FileStat) ContainsKey(key string) bool {
return f.MinKey >= key || key <= f.MaxKey
}
func NewFileStore(dir string) *FileStore {
return &FileStore{
dir: dir,
lastModified: time.Now(),
Logger: log.New(os.Stderr, "[filestore] ", log.LstdFlags),
statMap: influxdb.NewStatistics("tsm1_filestore:"+dir, "tsm1_filestore", map[string]string{"path": dir}),
}
}
// Returns the number of TSM files currently loaded
func (f *FileStore) Count() int {
f.mu.RLock()
defer f.mu.RUnlock()
return len(f.files)
}
// Files returns TSM files currently loaded.
func (f *FileStore) Files() []TSMFile {
f.mu.RLock()
defer f.mu.RUnlock()
return f.files
}
// CurrentGeneration returns the current generation of the TSM files
func (f *FileStore) CurrentGeneration() int {
f.mu.RLock()
defer f.mu.RUnlock()
return f.currentGeneration
}
// NextGeneration returns the max file ID + 1
func (f *FileStore) NextGeneration() int {
f.mu.Lock()
defer f.mu.Unlock()
f.currentGeneration++
return f.currentGeneration
}
func (f *FileStore) Add(files ...TSMFile) {
f.mu.Lock()
defer f.mu.Unlock()
for _, file := range files {
f.statMap.Add(statFileStoreBytes, int64(file.Size()))
}
f.files = append(f.files, files...)
sort.Sort(tsmReaders(f.files))
}
// Remove removes the files with matching paths from the set of active files. It does
// not remove the paths from disk.
func (f *FileStore) Remove(paths ...string) {
f.mu.Lock()
defer f.mu.Unlock()
var active []TSMFile
for _, file := range f.files {
keep := true
for _, remove := range paths {
if remove == file.Path() {
keep = false
break
}
}
if keep {
active = append(active, file)
} else {
// Removing the file, remove the file size from the total file store bytes
f.statMap.Add(statFileStoreBytes, -int64(file.Size()))
}
}
f.files = active
sort.Sort(tsmReaders(f.files))
}
func (f *FileStore) Keys() []string {
f.mu.RLock()
defer f.mu.RUnlock()
uniqueKeys := map[string]struct{}{}
for _, f := range f.files {
for _, key := range f.Keys() {
uniqueKeys[key] = struct{}{}
}
}
var keys []string
for key := range uniqueKeys {
keys = append(keys, key)
}
sort.Strings(keys)
return keys
}
func (f *FileStore) Type(key string) (byte, error) {
f.mu.RLock()
defer f.mu.RUnlock()
for _, f := range f.files {
if f.Contains(key) {
return f.Type(key)
}
}
return 0, fmt.Errorf("unknown type for %v", key)
}
func (f *FileStore) Delete(keys []string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.lastModified = time.Now()
for _, file := range f.files {
if err := file.Delete(keys); err != nil {
return err
}
}
return nil
}
func (f *FileStore) Open() error {
f.mu.Lock()
defer f.mu.Unlock()
// Not loading files from disk so nothing to do
if f.dir == "" {
return nil
}
files, err := filepath.Glob(filepath.Join(f.dir, fmt.Sprintf("*.%s", TSMFileExtension)))
if err != nil {
return err
}
// struct to hold the result of opening each reader in a goroutine
type res struct {
r *TSMReader
err error
}
readerC := make(chan *res)
for i, fn := range files {
// Keep track of the latest ID
generation, _, err := ParseTSMFileName(fn)
if err != nil {
return err
}
if generation >= f.currentGeneration {
f.currentGeneration = generation + 1
}
file, err := os.OpenFile(fn, os.O_RDONLY, 0666)
if err != nil {
return fmt.Errorf("error opening file %s: %v", fn, err)
}
// Accumulate file store size stat
if fi, err := file.Stat(); err == nil {
f.statMap.Add(statFileStoreBytes, fi.Size())
}
go func(idx int, file *os.File) {
start := time.Now()
df, err := NewTSMReaderWithOptions(TSMReaderOptions{
MMAPFile: file,
})
if f.traceLogging {
f.Logger.Printf("%s (#%d) opened in %v", file.Name(), idx, time.Now().Sub(start))
}
if err != nil {
readerC <- &res{r: df, err: fmt.Errorf("error opening memory map for file %s: %v", file.Name(), err)}
return
}
readerC <- &res{r: df}
}(i, file)
}
for range files {
res := <-readerC
if res.err != nil {
return res.err
}
f.files = append(f.files, res.r)
}
close(readerC)
sort.Sort(tsmReaders(f.files))
return nil
}
func (f *FileStore) Close() error {
f.mu.Lock()
defer f.mu.Unlock()
for _, f := range f.files {
f.Close()
}
f.files = nil
return nil
}
func (f *FileStore) Read(key string, t time.Time) ([]Value, error) {
f.mu.RLock()
defer f.mu.RUnlock()
for _, f := range f.files {
// Can this file possibly contain this key and timestamp?
if !f.Contains(key) {
continue
}
// May have the key and time we are looking for so try to find
v, err := f.Read(key, t)
if err != nil {
return nil, err
}
if len(v) > 0 {
return v, nil
}
}
return nil, nil
}
func (f *FileStore) KeyCursor(key string, t time.Time, ascending bool) *KeyCursor {
f.mu.RLock()
defer f.mu.RUnlock()
return newKeyCursor(f, key, t, ascending)
}
func (f *FileStore) Stats() []FileStat {
f.mu.RLock()
defer f.mu.RUnlock()
stats := make([]FileStat, len(f.files))
for i, fd := range f.files {
stats[i] = fd.Stats()
}
return stats
}
func (f *FileStore) Replace(oldFiles, newFiles []string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.lastModified = time.Now()
// Copy the current set of active files while we rename
// and load the new files. We copy the pointers here to minimize
// the time that locks are held as well as to ensure that the replacement
// is atomic.©
var updated []TSMFile
for _, t := range f.files {
updated = append(updated, t)
}
// Rename all the new files to make them live on restart
for _, file := range newFiles {
var newName = file
if strings.HasSuffix(file, ".tmp") {
// The new TSM files have a tmp extension. First rename them.
newName = file[:len(file)-4]
if err := os.Rename(file, newName); err != nil {
return err
}
}
fd, err := os.Open(newName)
if err != nil {
return err
}
tsm, err := NewTSMReaderWithOptions(TSMReaderOptions{
MMAPFile: fd,
})
if err != nil {
return err
}
updated = append(updated, tsm)
}
// We need to prune our set of active files now
var active []TSMFile
for _, file := range updated {
keep := true
for _, remove := range oldFiles {
if remove == file.Path() {
keep = false
if err := file.Close(); err != nil {
return err
}
if err := file.Remove(); err != nil {
return err
}
break
}
}
if keep {
active = append(active, file)
}
}
f.files = active
sort.Sort(tsmReaders(f.files))
// Recalculate the disk size stat
var totalSize int64
for _, file := range f.files {
totalSize += int64(file.Size())
}
sizeStat := new(expvar.Int)
sizeStat.Set(totalSize)
f.statMap.Set(statFileStoreBytes, sizeStat)
return nil
}
// LastModified returns the last time the file store was updated with new
// TSM files or a delete
func (f *FileStore) LastModified() time.Time {
f.mu.RLock()
defer f.mu.RUnlock()
return f.lastModified
}
// BlockCount returns number of values stored in the block at location idx
// in the file at path. If path does not match any file in the store, 0 is
// returned. If idx is out of range for the number of blocks in the file,
// 0 is returned.
func (f *FileStore) BlockCount(path string, idx int) int {
f.mu.RLock()
defer f.mu.RUnlock()
if idx < 0 {
return 0
}
for _, fd := range f.files {
if fd.Path() == path {
iter := fd.BlockIterator()
for i := 0; i < idx; i++ {
if !iter.Next() {
return 0
}
}
_, _, _, block, _ := iter.Read()
return BlockCount(block)
}
}
return 0
}
// locations returns the files and index blocks for a key and time. ascending indicates
// whether the key will be scan in ascending time order or descenging time order.
func (f *FileStore) locations(key string, t time.Time, ascending bool) []*location {
var locations []*location
f.mu.RLock()
filesSnapshot := make([]TSMFile, len(f.files))
for i := range f.files {
filesSnapshot[i] = f.files[i]
}
f.mu.RUnlock()
for _, fd := range filesSnapshot {
minTime, maxTime := fd.TimeRange()
// If we ascending and the max time of the file is before where we want to start
// skip it.
if ascending && maxTime.Before(t) {
continue
// If we are descending and the min time fo the file is after where we want to start,
// then skip it.
} else if !ascending && minTime.After(t) {
continue
}
// This file could potential contain points we are looking for so find the blocks for
// the given key.
for _, ie := range fd.Entries(key) {
// If we ascending and the max time of a block is before where we are looking, skip
// it since the data is out of our range
if ascending && ie.MaxTime.Before(t) {
continue
// If we descending and the min time of a block is after where we are looking, skip
// it since the data is out of our range
} else if !ascending && minTime.After(t) {
continue
}
// Otherwise, add this file and block location
locations = append(locations, &location{
r: fd,
entry: ie,
})
}
}
return locations
}
// ParseTSMFileName parses the generation and sequence from a TSM file name.
func ParseTSMFileName(name string) (int, int, error) {
base := filepath.Base(name)
idx := strings.Index(base, ".")
if idx == -1 {
return 0, 0, fmt.Errorf("file %s is named incorrectly", name)
}
id := base[:idx]
parts := strings.Split(id, "-")
if len(parts) != 2 {
return 0, 0, fmt.Errorf("file %s is named incorrectly", name)
}
generation, err := strconv.ParseUint(parts[0], 10, 32)
sequence, err := strconv.ParseUint(parts[1], 10, 32)
return int(generation), int(sequence), err
}
type KeyCursor struct {
key string
fs *FileStore
// seeks is all the file locations that we need to return during iteration.
seeks []*location
// current is the set of blocks possibly containing the next set of points.
// Normally this is just one entry, but there may be multiple if points have
// been overwritten.
current []*location
buf []Value
// pos is the index within seeks. Based on ascending, it will increment or
// decrement through the size of seeks slice.
pos int
ascending bool
// duplicates is a hint that there are overlapping blocks for this key in
// multiple files (e.g. points have been overwritten but not fully compacted)
// If this is true, we need to scan the duplicate blocks and dedup the points
// as query time until they are compacted.
duplicates bool
}
type location struct {
r TSMFile
entry *IndexEntry
// Has this location been read before
read bool
}
// newKeyCursor returns a new instance of KeyCursor.
func newKeyCursor(fs *FileStore, key string, t time.Time, ascending bool) *KeyCursor {
c := &KeyCursor{
key: key,
fs: fs,
seeks: fs.locations(key, t, ascending),
ascending: ascending,
}
c.duplicates = c.hasOverlappingBlocks()
c.seek(t)
return c
}
// Close removes all references on the cursor.
func (c *KeyCursor) Close() {
c.buf = nil
c.seeks = nil
c.fs = nil
c.current = nil
}
// hasOverlappingBlocks returns true if blocks have overlapping time ranges.
// This result is computed once and stored as the "duplicates" field.
func (c *KeyCursor) hasOverlappingBlocks() bool {
if len(c.seeks) == 0 {
return false
}
for i := 1; i < len(c.seeks); i++ {
prev := c.seeks[i-1]
cur := c.seeks[i]
if prev.entry.MaxTime.Equal(cur.entry.MinTime) || prev.entry.MaxTime.After(cur.entry.MinTime) {
return true
}
}
return false
}
// seek positions the cursor at the given time.
func (c *KeyCursor) seek(t time.Time) {
if len(c.seeks) == 0 {
return
}
c.current = nil
if c.ascending {
c.seekAscending(t)
} else {
c.seekDescending(t)
}
}
func (c *KeyCursor) seekAscending(t time.Time) {
for i, e := range c.seeks {
if t.Before(e.entry.MinTime) || e.entry.Contains(t) {
// Record the position of the first block matching our seek time
if len(c.current) == 0 {
c.pos = i
}
c.current = append(c.current, e)
// Exit if we don't have duplicates.
// Otherwise, keep looking for additional blocks containing this point.
if !c.duplicates {
return
}
}
}
}
func (c *KeyCursor) seekDescending(t time.Time) {
for i := len(c.seeks) - 1; i >= 0; i-- {
e := c.seeks[i]
if t.After(e.entry.MaxTime) || e.entry.Contains(t) {
// Record the position of the first block matching our seek time
if len(c.current) == 0 {
c.pos = i
}
c.current = append(c.current, e)
// Exit if we don't have duplicates.
// Otherwise, keep looking for additional blocks containing this point.
if !c.duplicates {
return
}
}
}
}
// Next moves the cursor to the next position.
// Data should be read by the ReadBlock functions.
func (c *KeyCursor) Next() {
c.current = c.current[:0]
if c.ascending {
c.nextAscending()
} else {
c.nextDescending()
}
}
func (c *KeyCursor) nextAscending() {
for {
c.pos++
if c.pos >= len(c.seeks) {
return
} else if !c.seeks[c.pos].read {
break
}
}
// Append the first matching block
c.current = []*location{c.seeks[c.pos]}
// We're done if there are no overlapping blocks.
if !c.duplicates {
return
}
// If we have ovelapping blocks, append all their values so we can dedup
first := c.seeks[c.pos]
for i := c.pos + 1; i < len(c.seeks); i++ {
if c.seeks[i].read {
continue
}
if c.seeks[i].entry.MinTime.Before(first.entry.MaxTime) || c.seeks[i].entry.MinTime.Equal(first.entry.MaxTime) {
c.current = append(c.current, c.seeks[i])
}
}
}
func (c *KeyCursor) nextDescending() {
for {
c.pos--
if c.pos < 0 {
return
} else if !c.seeks[c.pos].read {
break
}
}
// Append the first matching block
c.current = []*location{c.seeks[c.pos]}
// We're done if there are no overlapping blocks.
if !c.duplicates {
return
}
// If we have ovelapping blocks, append all their values so we can dedup
first := c.seeks[c.pos]
for i := c.pos; i >= 0; i-- {
if c.seeks[i].read {
continue
}
if c.seeks[i].entry.MaxTime.After(first.entry.MinTime) || c.seeks[i].entry.MaxTime.Equal(first.entry.MinTime) {
c.current = append(c.current, c.seeks[i])
}
}
}
// ReadFloatBlock reads the next block as a set of float values.
func (c *KeyCursor) ReadFloatBlock(buf []FloatValue) ([]FloatValue, error) {
// No matching blocks to decode
if len(c.current) == 0 {
return nil, nil
}
// First block is the oldest block containing the points we're search for.
first := c.current[0]
values, err := first.r.ReadFloatBlockAt(first.entry, buf[:0])
first.read = true
// Only one block with this key and time range so return it
if len(c.current) == 1 {
return values, err
}
// Otherwise, search the remaining blocks that overlap and append their values so we can
// dedup them.
for i := 1; i < len(c.current); i++ {
cur := c.current[i]
if c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos++
v, err := cur.r.ReadFloatBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(values, v...)
} else if !c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos--
v, err := cur.r.ReadFloatBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(v, values...)
}
}
return FloatValues(values).Deduplicate(), err
}
// ReadIntegerBlock reads the next block as a set of integer values.
func (c *KeyCursor) ReadIntegerBlock(buf []IntegerValue) ([]IntegerValue, error) {
// No matching blocks to decode
if len(c.current) == 0 {
return nil, nil
}
// First block is the oldest block containing the points we're search for.
first := c.current[0]
values, err := first.r.ReadIntegerBlockAt(first.entry, buf[:0])
first.read = true
// Only one block with this key and time range so return it
if len(c.current) == 1 {
return values, err
}
// Otherwise, search the remaining blocks that overlap and append their values so we can
// dedup them.
for i := 1; i < len(c.current); i++ {
cur := c.current[i]
if c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos++
v, err := cur.r.ReadIntegerBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(values, v...)
} else if !c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos--
v, err := cur.r.ReadIntegerBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(v, values...)
}
}
return IntegerValues(values).Deduplicate(), err
}
// ReadStringBlock reads the next block as a set of string values.
func (c *KeyCursor) ReadStringBlock(buf []StringValue) ([]StringValue, error) {
// No matching blocks to decode
if len(c.current) == 0 {
return nil, nil
}
// First block is the oldest block containing the points we're search for.
first := c.current[0]
values, err := first.r.ReadStringBlockAt(first.entry, buf[:0])
first.read = true
// Only one block with this key and time range so return it
if len(c.current) == 1 {
return values, err
}
// Otherwise, search the remaining blocks that overlap and append their values so we can
// dedup them.
for i := 1; i < len(c.current); i++ {
cur := c.current[i]
if c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos++
v, err := cur.r.ReadStringBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(values, v...)
} else if !c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos--
v, err := cur.r.ReadStringBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(v, values...)
}
}
return StringValues(values).Deduplicate(), err
}
// ReadBooleanBlock reads the next block as a set of boolean values.
func (c *KeyCursor) ReadBooleanBlock(buf []BooleanValue) ([]BooleanValue, error) {
// No matching blocks to decode
if len(c.current) == 0 {
return nil, nil
}
// First block is the oldest block containing the points we're search for.
first := c.current[0]
values, err := first.r.ReadBooleanBlockAt(first.entry, buf[:0])
first.read = true
// Only one block with this key and time range so return it
if len(c.current) == 1 {
return values, err
}
// Otherwise, search the remaining blocks that overlap and append their values so we can
// dedup them.
for i := 1; i < len(c.current); i++ {
cur := c.current[i]
if c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos++
v, err := cur.r.ReadBooleanBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(values, v...)
} else if !c.ascending && cur.entry.OverlapsTimeRange(first.entry.MinTime, first.entry.MaxTime) && !cur.read {
cur.read = true
c.pos--
v, err := cur.r.ReadBooleanBlockAt(cur.entry, nil)
if err != nil {
return nil, err
}
values = append(v, values...)
}
}
return BooleanValues(values).Deduplicate(), err
}
type tsmReaders []TSMFile
func (a tsmReaders) Len() int { return len(a) }
func (a tsmReaders) Less(i, j int) bool { return a[i].Path() < a[j].Path() }
func (a tsmReaders) Swap(i, j int) { a[i], a[j] = a[j], a[i] }