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