Check for unshipped blocks before deleting

Signed-off-by: Ganesh Vernekar <[email protected]>

Author: codesome

pkg/ingester/ingester_v2.go

@@ -2,8 +2,10 @@ package ingester
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"net/http"
 	"os"
@@ -12,6 +14,7 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/log/level"
+	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -120,6 +123,38 @@ func (u *userTSDB) setLastUpdate(t time.Time) {
 	u.lastUpdate.Store(t.Unix())
 }
 
+func (u *userTSDB) getShippedBlocksULID() ([]ulid.ULID, error) {
+	b, err := ioutil.ReadFile(filepath.Join(u.Dir(), shipper.MetaFilename))
+	if err != nil {
+		return nil, errors.Wrap(err, "read shipper meta file")
+	}
+	var shipperMeta shipper.Meta
+	if err := json.Unmarshal(b, &shipperMeta); err != nil {
+		return nil, errors.Wrap(err, "unmarshal shipper meta file to json")
+	}
+
+	return shipperMeta.Uploaded, nil
+}
+
+func (u *userTSDB) getUnshippedBlocksULID() (unshipped []ulid.ULID, err error) {
+	shippedBlocks, err := u.getShippedBlocksULID()
+	if err != nil {
+		return nil, errors.Wrap(err, "get shipped blocks")
+	}
+
+Outer:
+	for _, b := range u.Blocks() {
+		for _, uid := range shippedBlocks {
+			if uid == b.Meta().ULID {
+				continue Outer
+			}
+		}
+		unshipped = append(unshipped, b.Meta().ULID)
+	}
+
+	return unshipped, nil
+}
+
 // TSDBState holds data structures used by the TSDB storage engine
 type TSDBState struct {
 	dbs         map[string]*userTSDB // tsdb sharded by userID

pkg/ingester/ingester_v2_backfill.go

@@ -471,7 +471,20 @@ func (i *Ingester) closeOldBackfillTSDBsAndDelete(gracePeriod int64) error {
 			if err := db.Close(); err != nil {
 				return errors.Wrap(err, "close backfill TSDB")
 			}
-			// TODO(codesome): check if the blocks are shipped.
+
+			unshippedBlocks, err := db.getUnshippedBlocksULID()
+			if err != nil {
+				return errors.Wrap(err, "get unshipped blocks")
+			}
+			if len(unshippedBlocks) > 0 {
+				// Ship the unshipped blocks.
+				uploaded, err := db.shipper.Sync(context.Background())
+				if err != nil {
+					return errors.Wrap(err, "ship block")
+				}
+				level.Debug(util.Logger).Log("msg", "shipper successfully synchronized backfill TSDB blocks with storage", "user", db.userID, "uploaded", uploaded, "bucket_dir", db.Dir())
+			}
+
 			if err := os.RemoveAll(db.Dir()); err != nil {
 				return errors.Wrap(err, "delete backfill TSDB dir")
 			}