diff --git a/block.go b/block.go index 8cf28fe0e..7dc3af9d8 100644 --- a/block.go +++ b/block.go @@ -195,7 +195,7 @@ func readMetaFile(dir string) (*BlockMeta, error) { if err := json.Unmarshal(b, &m); err != nil { return nil, err } - if m.Version != 1 && m.Version != 2 { + if m.Version != 1 { return nil, errors.Errorf("unexpected meta file version %d", m.Version) } @@ -203,6 +203,8 @@ func readMetaFile(dir string) (*BlockMeta, error) { } func writeMetaFile(dir string, meta *BlockMeta) error { + meta.Version = 1 + // Make any changes to the file appear atomic. path := filepath.Join(dir, metaFilename) tmp := path + ".tmp" @@ -253,7 +255,7 @@ func OpenBlock(dir string, pool chunkenc.Pool) (*Block, error) { if err != nil { return nil, err } - ir, err := index.NewFileReader(filepath.Join(dir, "index"), meta.Version) + ir, err := index.NewFileReader(filepath.Join(dir, "index")) if err != nil { return nil, err } diff --git a/compact.go b/compact.go index 713ddc315..8ed922946 100644 --- a/compact.go +++ b/compact.go @@ -431,7 +431,6 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe if err != nil { return errors.Wrap(err, "open index writer") } - meta.Version = indexw.Version if err := c.populateBlock(blocks, meta, indexw, chunkw); err != nil { return errors.Wrap(err, "write compaction") diff --git a/db.go b/db.go index 97ff63682..9c301470b 100644 --- a/db.go +++ b/db.go @@ -188,6 +188,9 @@ func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db if opts == nil { opts = DefaultOptions } + if err := repairBadIndexVersion(dir); err != nil { + return nil, err + } db = &DB{ dir: dir, diff --git a/index/index.go b/index/index.go index f6b680b36..4fdc0398c 100644 --- a/index/index.go +++ b/index/index.go @@ -37,6 +37,7 @@ const ( MagicIndex = 0xBAAAD700 indexFormatV1 = 1 + indexFormatV2 = 2 ) type indexWriterSeries struct { @@ -168,8 +169,6 @@ func NewWriter(fn string) (*Writer, error) { symbols: make(map[string]uint32, 1<<13), seriesOffsets: make(map[uint64]uint64, 1<<16), crc32: newCRC32(), - - Version: 2, } if err := iw.writeMeta(); err != nil { return nil, err @@ -195,7 +194,7 @@ func (w *Writer) write(bufs ...[]byte) error { return nil } -// addPadding adds zero byte padding until the file size is a multiple size_unit. +// addPadding adds zero byte padding until the file size is a multiple size. func (w *Writer) addPadding(size int) error { p := w.pos % uint64(size) if p == 0 { @@ -249,7 +248,7 @@ func (w *Writer) ensureStage(s indexWriterStage) error { func (w *Writer) writeMeta() error { w.buf1.reset() w.buf1.putBE32(MagicIndex) - w.buf1.putByte(indexFormatV1) + w.buf1.putByte(indexFormatV2) return w.write(w.buf1.get()) } @@ -266,7 +265,13 @@ func (w *Writer) AddSeries(ref uint64, lset labels.Labels, chunks ...chunks.Meta if _, ok := w.seriesOffsets[ref]; ok { return errors.Errorf("series with reference %d already added", ref) } + // We add padding to 16 bytes to increase the addressable space we get through 4 byte + // series references. w.addPadding(16) + + if w.pos%16 != 0 { + return errors.Errorf("series write not 16-byte aligned at %d", w.pos) + } w.seriesOffsets[ref] = w.pos / 16 w.buf2.reset() @@ -573,24 +578,20 @@ func (b realByteSlice) Sub(start, end int) ByteSlice { } // NewReader returns a new IndexReader on the given byte slice. -func NewReader(b ByteSlice, version int) (*Reader, error) { - return newReader(b, nil, version) +func NewReader(b ByteSlice) (*Reader, error) { + return newReader(b, nil) } // NewFileReader returns a new index reader against the given index file. -func NewFileReader(path string, version int) (*Reader, error) { +func NewFileReader(path string) (*Reader, error) { f, err := fileutil.OpenMmapFile(path) if err != nil { return nil, err } - return newReader(realByteSlice(f.Bytes()), f, version) + return newReader(realByteSlice(f.Bytes()), f) } -func newReader(b ByteSlice, c io.Closer, version int) (*Reader, error) { - if version != 1 && version != 2 { - return nil, errors.Errorf("unexpected file version %d", version) - } - +func newReader(b ByteSlice, c io.Closer) (*Reader, error) { r := &Reader{ b: b, c: c, @@ -598,16 +599,20 @@ func newReader(b ByteSlice, c io.Closer, version int) (*Reader, error) { labels: map[string]uint32{}, postings: map[labels.Label]uint32{}, crc32: newCRC32(), - version: version, } - // Verify magic number. - if b.Len() < 4 { + // Verify header. + if b.Len() < 5 { return nil, errors.Wrap(errInvalidSize, "index header") } if m := binary.BigEndian.Uint32(r.b.Range(0, 4)); m != MagicIndex { return nil, errors.Errorf("invalid magic number %x", m) } + r.version = int(r.b.Range(4, 5)[0]) + + if r.version != 1 && r.version != 2 { + return nil, errors.Errorf("unknown index file version %d", r.version) + } if err := r.readTOC(); err != nil { return nil, errors.Wrap(err, "read TOC") @@ -880,8 +885,10 @@ func (r *Reader) LabelIndices() ([][]string, error) { // Series reads the series with the given ID and writes its labels and chunks into lbls and chks. func (r *Reader) Series(id uint64, lbls *labels.Labels, chks *[]chunks.Meta) error { offset := id + // In version 2 series IDs are no longer exact references but series are 16-byte padded + // and the ID is the multiple of 16 of the actual position. if r.version == 2 { - offset = 16 * id + offset = id * 16 } d := r.decbufUvarintAt(int(offset)) if d.err() != nil { diff --git a/index/index_test.go b/index/index_test.go index 65ef9b6e6..83b6ef657 100644 --- a/index/index_test.go +++ b/index/index_test.go @@ -160,7 +160,7 @@ func TestIndexRW_Create_Open(t *testing.T) { testutil.Ok(t, err) testutil.Ok(t, iw.Close()) - ir, err := NewFileReader(fn, 1) + ir, err := NewFileReader(fn) testutil.Ok(t, err) testutil.Ok(t, ir.Close()) @@ -170,7 +170,7 @@ func TestIndexRW_Create_Open(t *testing.T) { _, err = f.WriteAt([]byte{0, 0}, 0) testutil.Ok(t, err) - _, err = NewFileReader(dir, 1) + _, err = NewFileReader(dir) testutil.NotOk(t, err) } @@ -213,7 +213,7 @@ func TestIndexRW_Postings(t *testing.T) { testutil.Ok(t, iw.Close()) - ir, err := NewFileReader(fn, 2) + ir, err := NewFileReader(fn) testutil.Ok(t, err) p, err := ir.Postings("a", "1") @@ -331,7 +331,7 @@ func TestPersistence_index_e2e(t *testing.T) { err = iw.Close() testutil.Ok(t, err) - ir, err := NewFileReader(filepath.Join(dir, "index"), 2) + ir, err := NewFileReader(filepath.Join(dir, "index")) testutil.Ok(t, err) for p := range mi.postings { diff --git a/repair.go b/repair.go new file mode 100644 index 000000000..dc063acdc --- /dev/null +++ b/repair.go @@ -0,0 +1,86 @@ +package tsdb + +import ( + "encoding/json" + "io" + "io/ioutil" + "os" + "path/filepath" + + "github.com/go-kit/kit/log" + "github.com/go-kit/kit/log/level" + "github.com/oklog/ulid" + "github.com/pkg/errors" + "github.com/prometheus/tsdb/fileutil" +) + +func repairBadIndexVersion(logger log.Logger, dir string) error { + // All blocks written by Prometheus 2.1 with a meta.json version of 2 are affected. + // We must actually set the index file version to 2 and revert the meta.json version back to 1. + subdirs, err := fileutil.ReadDir(dir) + if err != nil { + return err + } + for _, d := range subdirs { + // Skip non-block dirs. + if _, err := ulid.Parse(d); err != nil { + continue + } + + meta, err := readBogusMetaFile(d) + if err != nil { + return err + } + if meta.Version == 1 { + continue + } + level.Info(logger).Log("msg", "fixing broken block", "ulid", meta.ULID) + + repl, err := os.Create(filepath.Join(d, "index.repaired")) + if err != nil { + return err + } + broken, err := os.Open(filepath.Join(d, "index")) + if err != nil { + return err + } + if _, err := io.Copy(repl, broken); err != nil { + return err + } + // Set the 5th byte to 2 to indiciate the correct file format version. + if _, err := repl.WriteAt([]byte{2}, 4); err != nil { + return err + } + if err := fileutil.Fsync(repl); err != nil { + return err + } + if err := repl.Close(); err != nil { + return err + } + if err := renameFile(repl.Name(), broken.Name()); err != nil { + return err + } + // Reset version of meta.json to 1. + meta.Version = 1 + if err := writeMetaFile(d, meta); err != nil { + return err + } + } + return nil +} + +func readBogusMetaFile(dir string) (*BlockMeta, error) { + b, err := ioutil.ReadFile(filepath.Join(dir, metaFilename)) + if err != nil { + return nil, err + } + var m BlockMeta + + if err := json.Unmarshal(b, &m); err != nil { + return nil, err + } + if m.Version != 1 && m.Version != 2 { + return nil, errors.Errorf("unexpected meta file version %d", m.Version) + } + return &m, nil +} diff --git a/repair_test.go b/repair_test.go new file mode 100644 index 000000000..4d396ab3b --- /dev/null +++ b/repair_test.go @@ -0,0 +1,7 @@ +package tsdb + +import "testing" + +func TestRepairBadIndexVersion(t *testing.T) { + +}