Don't write empty blocks

block.go

@@ -178,6 +178,10 @@ type BlockMetaCompaction struct {
 	Level int `json:"level"`
 	// ULIDs of all source head blocks that went into the block.
 	Sources []ulid.ULID `json:"sources,omitempty"`
+	// As we dont write empty blocks, we need this to mark
+	// if the block is deletable if in compaction this block
+	// resulted in an empty block.
+	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

@@ -349,14 +349,29 @@ func (c *LeveledCompactor) Compact(dest string, dirs ...string) (uid ulid.ULID,
 	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),
-		)
+
+		if meta.Stats.NumSamples == 0 {
+			level.Info(c.logger).Log(
+				"msg", "compact blocks [resulted in empty block]",
+				"count", len(blocks),
+				"sources", fmt.Sprintf("%v", uids),
+			)
+			for _, b := range bs {
+				b.meta.Compaction.Deletable = true
+				writeMetaFile(b.dir, &b.meta)
+			}
+			uid = ulid.ULID{}
+		} 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),
+			)
+		}
+
 		return uid, nil
 	}
 
@@ -471,17 +486,24 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 		return errors.Wrap(err, "write compaction")
 	}
 
-	if err = writeMetaFile(tmp, meta); err != nil {
-		return errors.Wrap(err, "write merged meta")
-	}
-
 	if err = chunkw.Close(); err != nil {
 		return errors.Wrap(err, "close chunk writer")
 	}
 	if err = indexw.Close(); err != nil {
 		return errors.Wrap(err, "close index writer")
 	}
 
+	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

@@ -383,6 +383,22 @@ func (db *DB) compact() (changes bool, err error) {
 		if err := db.reload(); err != nil {
 			return changes, errors.Wrap(err, "reload blocks")
 		}
+		// After this is fixed https://github.com/prometheus/tsdb/issues/309,
+		// if the compaction of head did not create any new blocks (possible
+		// if the samples were deleted in that range), and if number of
+		// blocks are 0, db.reload() does not truncate head. Hence MinTime
+		// of head is not set properly. So only for that situation, we need
+		// to call db.head.Truncate(..) manually.
+		db.mtx.RLock()
+		l := len(db.blocks)
+		db.mtx.RUnlock()
+		if l == 0 {
+			err = db.head.Truncate(maxt)
+			if err != nil {
+				return changes, errors.Wrap(err, "head truncate failed (in compact)")
+
+			}
+		}
 		runtime.GC()
 	}
 
@@ -468,6 +484,9 @@ func (db *DB) reload() (err error) {
 			corrupted[ulid] = err
 			continue
 		}
+		if meta.Compaction.Deletable {
+			deleteable[meta.ULID] = struct{}{}
+		}
 		if db.beyondRetention(meta) {
 			deleteable[meta.ULID] = struct{}{}
 			continue

db_test.go

@@ -1301,3 +1301,62 @@ func TestInitializeHeadTimestamp(t *testing.T) {
 		testutil.Equals(t, int64(15000), db.head.MaxTime())
 	})
 }
+
+func TestNoEmptyBlock(t *testing.T) {
+	db, close := openTestDB(t, nil)
+	defer close()
+	defer db.Close()
+
+	// Test no blocks after compact with empty head.
+	_, err := db.compact()
+	testutil.Ok(t, err)
+	testutil.Equals(t, 0, len(db.blocks))
+
+	// 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 < 3; i++ {
+		_, err := app.Add(label, i*blockRange, 0)
+		testutil.Ok(t, err)
+		_, err = app.Add(label, i*blockRange+1000, 0)
+		testutil.Ok(t, err)
+	}
+	err = app.Commit()
+	testutil.Ok(t, err)
+
+	testutil.Ok(t, db.Delete(math.MinInt64, math.MaxInt64, labels.NewEqualMatcher("foo", "bar")))
+	_, err = db.compact()
+	testutil.Ok(t, err)
+	// No blocks created.
+	testutil.Equals(t, 0, len(db.blocks))
+
+	// Test no blocks remaining after small samples are deleted from disk.
+	app = db.Appender()
+	for i := int64(3); i < 20; i++ {
+		for j := int64(0); j < 10; j++ {
+			_, err := app.Add(label, i*blockRange+j, 0)
+			testutil.Ok(t, err)
+		}
+	}
+	err = app.Commit()
+	testutil.Ok(t, err)
+
+	_, err = db.compact()
+	testutil.Ok(t, err)
+	testutil.Assert(t, len(db.blocks) > 0, "No blocks created")
+
+	testutil.Ok(t, db.Delete(math.MinInt64, math.MaxInt64, labels.NewEqualMatcher("foo", "bar")))
+
+	// Mimicking small part of compaction.
+	plan := []string{}
+	for _, b := range db.blocks {
+		plan = append(plan, b.Dir())
+	}
+	_, err = db.compactor.Compact(db.dir, plan...)
+	testutil.Ok(t, err)
+	testutil.Ok(t, db.reload())
+	// All samples are deleted. No blocks should be remeianing after compact.
+	testutil.Equals(t, 0, len(db.blocks))
+}