compact: clean up directories properly

I couldn't stop thinking about this code for some reason and I have
figured that I had missed one case in thanos-io#3031. We need to also clean up
the directories in compaction groups. A compaction could fail leaving
some new directory with a random ULID on the disk. Before attempting to
do another compaction loop, we need to remove it as well because the
compaction process always produces a new, unique directory.

Signed-off-by: Giedrius Statkevičius <giedriuswork@gmail.com>

pkg/compact/compact.go

@@ -957,6 +957,16 @@ func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
 
 		ignoreDirs := []string{}
 		for _, gr := range groups {
+			groupIDs := []string{}
+			for _, grID := range gr.IDs() {
+				groupIDs = append(groupIDs, grID.String())
+			}
+
+			groupDir := filepath.Join(c.compactDir, gr.Key())
+			if err := runutil.DeleteAll(groupDir, groupIDs...); err != nil {
+				level.Warn(c.logger).Log("msg", "failed deleting non-compaction group sub-directories/files, some disk space usage might have leaked. Continuing", "err", err, "dir", groupDir)
+			}
+
 			ignoreDirs = append(ignoreDirs, gr.Key())
 		}
 

pkg/runutil/runutil.go

@@ -164,14 +164,17 @@ func ExhaustCloseWithErrCapture(err *error, r io.ReadCloser, format string, a ..
 // dir except for the ignoreDirs directories.
 func DeleteAll(dir string, ignoreDirs ...string) error {
 	entries, err := ioutil.ReadDir(dir)
+	if os.IsNotExist(err) {
+		return nil
+	}
 	if err != nil {
 		return errors.Wrap(err, "read dir")
 	}
 	var groupErrs errutil.MultiError
 
 	for _, d := range entries {
 		if !d.IsDir() {
-			if err := os.RemoveAll(filepath.Join(dir, d.Name())); err != nil {
+			if err := os.RemoveAll(filepath.Join(dir, d.Name())); err != nil && !os.IsNotExist(err) {
 				groupErrs.Add(err)
 			}
 			continue
@@ -186,7 +189,7 @@ func DeleteAll(dir string, ignoreDirs ...string) error {
 		}
 
 		if !found {
-			if err := os.RemoveAll(filepath.Join(dir, d.Name())); err != nil {
+			if err := os.RemoveAll(filepath.Join(dir, d.Name())); err != nil && !os.IsNotExist(err) {
 				groupErrs.Add(err)
 			}
 		}

test/e2e/compact_test.go

@@ -539,8 +539,26 @@ func TestCompactWithStoreGateway(t *testing.T) {
 
 	// No replica label with overlaps should halt compactor. This test is sequential
 	// because we do not want two Thanos Compact instances deleting the same partially
-	// uploaded blocks and blocks with deletion marks.
+	// uploaded blocks and blocks with deletion marks. We also check that Thanos Compactor
+	// deletes directories inside of a compaction group that do not belong there.
 	{
+		// Precreate a directory. It should be deleted.
+		// In a hypothetical scenario, the directory could be a left-over from
+		// a compaction that had crashed.
+		p := filepath.Join(s.SharedDir(), "data", "compact", "expect-to-halt", "compact")
+
+		testutil.Assert(t, len(blocksWithHashes) > 0)
+
+		m, err := block.DownloadMeta(ctx, logger, bkt, blocksWithHashes[0])
+		testutil.Ok(t, err)
+
+		randBlockDir := filepath.Join(p, compact.DefaultGroupKey(m.Thanos), "ITISAVERYRANDULIDFORTESTS0")
+		testutil.Ok(t, os.MkdirAll(randBlockDir, os.ModePerm))
+
+		f, err := os.Create(filepath.Join(randBlockDir, "index"))
+		testutil.Ok(t, err)
+		testutil.Ok(t, f.Close())
+
 		c, err := e2ethanos.NewCompactor(s.SharedDir(), "expect-to-halt", svcConfig, nil)
 		testutil.Ok(t, err)
 		testutil.Ok(t, s.StartAndWaitReady(c))
@@ -578,6 +596,10 @@ func TestCompactWithStoreGateway(t *testing.T) {
 		ensureGETStatusCode(t, http.StatusOK, "http://"+path.Join(c.HTTPEndpoint(), "loaded"))
 
 		testutil.Ok(t, s.Stop(c))
+
+		_, err = os.Stat(randBlockDir)
+		testutil.NotOk(t, err)
+		testutil.Assert(t, os.IsNotExist(err))
 	}
 
 	// Sequential because we want to check that Thanos Compactor does not