block: sanity check metas for common corruptions

Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>

pkg/block/fetcher.go

@@ -11,6 +11,7 @@ import (
 	"path"
 	"path/filepath"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -434,6 +435,10 @@ func (f *BaseFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Met
 		return nil, errors.Wrapf(ErrorSyncMetaCorrupted, "meta.json %v unmarshal: %v", metaFile, err)
 	}
 
+	if err := sanityCheckFilesForMeta(m.Thanos.Files); err != nil {
+		return nil, errors.Wrapf(ErrorSyncMetaCorrupted, "meta.json %v not sane: %v", metaFile, err)
+	}
+
 	if m.Version != metadata.TSDBVersion1 {
 		return nil, errors.Errorf("unexpected meta file: %s version: %d", metaFile, m.Version)
 	}
@@ -451,6 +456,42 @@ func (f *BaseFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Met
 	return m, nil
 }
 
+func sanityCheckFilesForMeta(files []metadata.File) error {
+	var (
+		numChunks    int
+		highestChunk int
+		hasIndex     bool
+	)
+
+	for _, f := range files {
+		if f.RelPath == "index" {
+			hasIndex = true
+		}
+		dir, name := path.Split(f.RelPath)
+		if dir == "chunks" {
+			numChunks++
+			idx, err := strconv.Atoi(name)
+			if err != nil {
+				return errors.Wrap(err, "unexpected chunk name")
+			}
+			if idx > highestChunk {
+				highestChunk = idx
+			}
+		}
+	}
+
+	if !hasIndex {
+		return errors.New("no index file in meta")
+	}
+	if numChunks == 0 {
+		return errors.New("no chunks in meta")
+	}
+	if numChunks != highestChunk {
+		return errors.New("incomplete chunks in meta")
+	}
+	return nil
+}
+
 type response struct {
 	metas   map[ulid.ULID]*metadata.Meta
 	partial map[ulid.ULID]error