Don't write empty blocks

CHANGELOG.md

@@ -3,3 +3,4 @@
  - `LastCheckpoint` used to return just the segment name and now it returns the full relative path.
  - `NewSegmentsRangeReader` can now read over miltiple wal ranges by using the new `SegmentRange` struct.
  - `CorruptionErr` now also exposes the Segment `Dir` which is added when displaying any errors.
+ - \[CHANGE\] Empty blocks are not written during compaction [#374](https://github.com/prometheus/tsdb/pull/374)

block.go

@@ -182,6 +182,9 @@ type BlockMetaCompaction struct {
 	Level int `json:"level"`
 	// ULIDs of all source head blocks that went into the block.
 	Sources []ulid.ULID `json:"sources,omitempty"`
+	// Indicates that during compaction it resulted in a block without any samples
+	// so it should be deleted on the next reload.
+	Deletable bool `json:"deletable,omitempty"`
 	// Short descriptions of the direct blocks that were used to create
 	// this block.
 	Parents []BlockDesc `json:"parents,omitempty"`

compact.go

@@ -55,12 +55,17 @@ type Compactor interface {
 	Plan(dir string) ([]string, error)
 
 	// Write persists a Block into a directory.
+	// No Block is written when resulting Block has 0 samples, and returns empty ulid.ULID{}.
 	Write(dest string, b BlockReader, mint, maxt int64, parent *BlockMeta) (ulid.ULID, error)
 
 	// Compact runs compaction against the provided directories. Must
 	// only be called concurrently with results of Plan().
 	// Can optionally pass a list of already open blocks,
 	// to avoid having to reopen them.
+	// When resulting Block has 0 samples
+	//  * No block is written.
+	//  * The source dirs are marked Deletable.
+	//  * Returns empty ulid.ULID{}.
 	Compact(dest string, dirs []string, open []*Block) (ulid.ULID, error)
 }
 
@@ -366,15 +371,34 @@ func (c *LeveledCompactor) Compact(dest string, dirs []string, open []*Block) (u
 	meta := compactBlockMetas(uid, metas...)
 	err = c.write(dest, meta, blocks...)
 	if err == nil {
-		level.Info(c.logger).Log(
-			"msg", "compact blocks",
-			"count", len(blocks),
-			"mint", meta.MinTime,
-			"maxt", meta.MaxTime,
-			"ulid", meta.ULID,
-			"sources", fmt.Sprintf("%v", uids),
-			"duration", time.Since(start),
-		)
+		if meta.Stats.NumSamples == 0 {
+			for _, b := range bs {
+				b.meta.Compaction.Deletable = true
+				if err = writeMetaFile(b.dir, &b.meta); err != nil {
+					level.Error(c.logger).Log(
+						"msg", "Failed to write 'Deletable' to meta file after compaction",
+						"ulid", b.meta.ULID,
+					)
+				}
+			}
+			uid = ulid.ULID{}
+			level.Info(c.logger).Log(
+				"msg", "compact blocks [resulted in empty block]",
+				"count", len(blocks),
+				"sources", fmt.Sprintf("%v", uids),
+				"duration", time.Since(start),
+			)
+		} else {
+			level.Info(c.logger).Log(
+				"msg", "compact blocks",
+				"count", len(blocks),
+				"mint", meta.MinTime,
+				"maxt", meta.MaxTime,
+				"ulid", meta.ULID,
+				"sources", fmt.Sprintf("%v", uids),
+				"duration", time.Since(start),
+			)
+		}
 		return uid, nil
 	}
 
@@ -413,6 +437,10 @@ func (c *LeveledCompactor) Write(dest string, b BlockReader, mint, maxt int64, p
 		return uid, err
 	}
 
+	if meta.Stats.NumSamples == 0 {
+		return ulid.ULID{}, nil
+	}
+
 	level.Info(c.logger).Log("msg", "write block", "mint", meta.MinTime, "maxt", meta.MaxTime, "ulid", meta.ULID)
 	return uid, nil
 }
@@ -490,11 +518,6 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 	if err := c.populateBlock(blocks, meta, indexw, chunkw); err != nil {
 		return errors.Wrap(err, "write compaction")
 	}
-
-	if err = writeMetaFile(tmp, meta); err != nil {
-		return errors.Wrap(err, "write merged meta")
-	}
-
 	// We are explicitly closing them here to check for error even
 	// though these are covered under defer. This is because in Windows,
 	// you cannot delete these unless they are closed and the defer is to
@@ -506,6 +529,18 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 		return errors.Wrap(err, "close index writer")
 	}
 
+	// Populated block is empty, so cleanup and exit.
+	if meta.Stats.NumSamples == 0 {
+		if err := os.RemoveAll(tmp); err != nil {
+			return errors.Wrap(err, "remove tmp folder after empty block failed")
+		}
+		return nil
+	}
+
+	if err = writeMetaFile(tmp, meta); err != nil {
+		return errors.Wrap(err, "write merged meta")
+	}
+
 	// Create an empty tombstones file.
 	if err := writeTombstoneFile(tmp, newMemTombstones()); err != nil {
 		return errors.Wrap(err, "write new tombstones file")

db.go

@@ -409,7 +409,8 @@ func (db *DB) compact() (err error) {
 			// from the block interval here.
 			maxt: maxt - 1,
 		}
-		if _, err = db.compactor.Write(db.dir, head, mint, maxt, nil); err != nil {
+		uid, err := db.compactor.Write(db.dir, head, mint, maxt, nil)
+		if err != nil {
 			return errors.Wrap(err, "persist head block")
 		}
 
@@ -418,6 +419,17 @@ func (db *DB) compact() (err error) {
 		if err := db.reload(); err != nil {
 			return errors.Wrap(err, "reload blocks")
 		}
+		if (uid == ulid.ULID{}) { // No block created.
+			// After this was fixed https://github.com/prometheus/tsdb/issues/309,
+			// Case 1: It is possible to have 0 blocks after compaction.
+			//         db.reload() doesn't truncate the head when the block count is zero.
+			// Case 2: If there were blocks on disk, head will be truncated to a
+			//         wrong value based on old blocks.
+			// Hence we need to truncate manually.
+			if err = db.head.Truncate(maxt); err != nil {
+				return errors.Wrap(err, "head truncate failed (in compact)")
+			}
+		}
 		runtime.GC()
 	}
 
@@ -506,6 +518,9 @@ func (db *DB) reload() (err error) {
 			deleteable[meta.ULID] = struct{}{}
 			continue
 		}
+		if meta.Compaction.Deletable {
+			deleteable[meta.ULID] = struct{}{}
+		}
 		for _, b := range meta.Compaction.Parents {
 			deleteable[b.ULID] = struct{}{}
 		}

db_test.go

@@ -1309,6 +1309,112 @@ func TestInitializeHeadTimestamp(t *testing.T) {
 	})
 }
 
+func TestNoEmptyBlocks(t *testing.T) {
+	db, close := openTestDB(t, nil)
+	defer close()
+	defer db.Close()
+	db.DisableCompactions()
+
+	// Test no blocks after compact with empty head.
+	testutil.Ok(t, db.compact())
+	bb, err := blockDirs(db.Dir())
+	testutil.Ok(t, err)
+	testutil.Equals(t, len(db.Blocks()), len(bb))
+	testutil.Equals(t, 0, len(bb))
+
+	// Test no blocks after deleting all samples from head.
+	blockRange := DefaultOptions.BlockRanges[0]
+	label := labels.FromStrings("foo", "bar")
+
+	app := db.Appender()
+	for i := int64(0); i < 6; i++ {
+		_, err := app.Add(label, i*blockRange, 0)
+		testutil.Ok(t, err)
+		_, err = app.Add(label, i*blockRange+1000, 0)
+		testutil.Ok(t, err)
+	}
+	testutil.Ok(t, app.Commit())
+
+	testutil.Ok(t, db.Delete(math.MinInt64, math.MaxInt64, labels.NewEqualMatcher("foo", "bar")))
+	uid, err := db.compactor.Write(db.dir, db.head, db.head.MinTime(), db.head.MaxTime(), nil)
+	testutil.Ok(t, err)
+	testutil.Equals(t, ulid.ULID{}, uid)
+	testutil.Ok(t, db.reload())
+	testutil.Ok(t, db.head.Truncate(db.head.MaxTime()))
+	// No blocks created.
+	bb, err = blockDirs(db.Dir())
+	testutil.Ok(t, err)
+	testutil.Equals(t, len(db.Blocks()), len(bb))
+	testutil.Equals(t, 0, len(bb))
+
+	app = db.Appender()
+	for i := int64(7); i < 25; i++ {
+		for j := int64(0); j < 10; j++ {
+			_, err := app.Add(label, i*blockRange+j, 0)
+			testutil.Ok(t, err)
+		}
+	}
+	testutil.Ok(t, app.Commit())
+
+	testutil.Ok(t, db.compact())
+	bb, err = blockDirs(db.Dir())
+	testutil.Ok(t, err)
+	testutil.Equals(t, len(db.Blocks()), len(bb))
+	testutil.Assert(t, len(bb) > 0, "No blocks created when compacting with >0 samples")
+
+	// When no new block is created from head, and there are some blocks on disk,
+	// compaction should not run into infinite loop (was seen during development).
+	oldBlocks := db.Blocks()
+	app = db.Appender()
+	for i := int64(26); i < 30; i++ {
+		_, err := app.Add(label, i*blockRange, 0)
+		testutil.Ok(t, err)
+	}
+	testutil.Ok(t, app.Commit())
+	testutil.Ok(t, db.head.Delete(math.MinInt64, math.MaxInt64, labels.NewEqualMatcher("foo", "bar")))
+	testutil.Ok(t, db.compact())
+	testutil.Equals(t, oldBlocks, db.Blocks())
+
+	// Test no blocks remaining after deleting all samples from disk.
+	testutil.Ok(t, db.Delete(math.MinInt64, math.MaxInt64, labels.NewEqualMatcher("foo", "bar")))
+
+	// Mimicking Plan() of compactor and getting list
+	// of all block directories to pass for compaction.
+	plan := []string{}
+	for _, b := range db.Blocks() {
+		plan = append(plan, b.Dir())
+	}
+
+	// Blocks are not set deletable before compaction.
+	for _, b := range db.Blocks() {
+		meta, err := readMetaFile(b.dir)
+		testutil.Ok(t, err)
+		testutil.Assert(t, meta.Compaction.Deletable == false, "Block was marked deletable before compaction")
+	}
+
+	// No new blocks are created by Compact, and marks all old blocks as deletable.
+	oldBlocks = db.Blocks()
+	_, err = db.compactor.Compact(db.dir, plan, db.Blocks())
+	testutil.Ok(t, err)
+	// Blocks are the same.
+	testutil.Equals(t, oldBlocks, db.Blocks())
+
+	// Marked as deletable.
+	for _, b := range db.Blocks() {
+		meta, err := readMetaFile(b.dir)
+		testutil.Ok(t, err)
+		testutil.Assert(t, meta.Compaction.Deletable, "Block was not marked deletable after compaction")
+	}
+
+	// Deletes the deletable blocks.
+	testutil.Ok(t, db.reload())
+	// All samples are deleted. No blocks should be remaining after compact.
+	bb, err = blockDirs(db.Dir())
+	testutil.Ok(t, err)
+	testutil.Equals(t, len(db.Blocks()), len(bb))
+	testutil.Equals(t, 0, len(bb))
+}
+
 func TestDB_LabelNames(t *testing.T) {
 	tests := []struct {
 		// Add 'sampleLabels1' -> Test Head -> Compact -> Test Disk ->