Upload blocks with the lowest minTime first

Fixes thanos-io#1670

Signed-off-by: Olivier Biesmans <olivier.biesmans@blablacar.com>

CHANGELOG.md

@@ -19,6 +19,7 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 
 - [#1656](https://github.com/thanos-io/thanos/pull/1656) Thanos Store now starts metric and status probe HTTP server earlier in its start-up sequence. `/-/healthy` endpoint now starts to respond with success earlier. `/metrics` endpoint starts serving metrics earlier as well. Make sure to point your readiness probes to the `/-/ready` endpoint rather than `/metrics`.
 - [#1669](https://github.com/thanos-io/thanos/pull/1669) Fixed store sharding. Now it does not load excluded meta.jsons and load/fetch index-cache.json files.
+- [#1670](https://github.com/thanos-io/thanos/pull/1670) Fixed un-ordered blocks upload. Sidecar now uploads the oldest blocks first.
 
 ### Changed
 

pkg/shipper/shipper.go

@@ -380,10 +380,12 @@ func (s *Shipper) upload(ctx context.Context, meta *metadata.Meta) error {
 	return block.Upload(ctx, s.logger, s.bucket, updir)
 }
 
-// iterBlockMetas calls f with the block meta for each block found in dir. It logs
-// an error and continues if it cannot access a meta.json file.
+// iterBlockMetas calls f with the block meta for each block found in dir
+// sorted by minTime asc. It logs an error and continues if it cannot access a
+// meta.json file.
 // If f returns an error, the function returns with the same error.
 func (s *Shipper) iterBlockMetas(f func(m *metadata.Meta) error) error {
+	var metas []*metadata.Meta
 	names, err := fileutil.ReadDir(s.dir)
 	if err != nil {
 		return errors.Wrap(err, "read dir")
@@ -407,6 +409,13 @@ func (s *Shipper) iterBlockMetas(f func(m *metadata.Meta) error) error {
 			level.Warn(s.logger).Log("msg", "reading meta file failed", "err", err)
 			continue
 		}
+		metas = append(metas, m)
+	}
+	sort.Slice(metas, func(i, j int) bool {
+		return metas[i].BlockMeta.MinTime < metas[j].BlockMeta.MinTime
+	})
+	for _, m := range metas {
+
 		if err := f(m); err != nil {
 			return err
 		}

pkg/shipper/shipper_test.go

@@ -3,8 +3,10 @@ package shipper
 import (
 	"io/ioutil"
 	"math"
+	"math/rand"
 	"os"
 	"path"
+	"sort"
 	"testing"
 
 	"github.com/go-kit/kit/log"
@@ -77,3 +79,85 @@ func TestShipperTimestamps(t *testing.T) {
 	testutil.Equals(t, int64(1000), mint)
 	testutil.Equals(t, int64(2000), maxt)
 }
+
+func TestIterBlockMetas(t *testing.T) {
+	var metas []*metadata.Meta
+	dir, err := ioutil.TempDir("", "shipper-test")
+	testutil.Ok(t, err)
+	defer func() {
+		testutil.Ok(t, os.RemoveAll(dir))
+	}()
+
+	for i := 0; i < 10; i++ {
+		id := ulid.MustNew(uint64(i), nil)
+		testutil.Ok(t, os.Mkdir(path.Join(dir, id.String()), os.ModePerm))
+		testutil.Ok(t,
+			metadata.Write(
+				log.NewNopLogger(),
+				path.Join(dir, id.String()),
+				&metadata.Meta{
+					BlockMeta: tsdb.BlockMeta{
+						ULID:    id,
+						MaxTime: int64((i + 1) * 1000),
+						MinTime: int64(i * 1000),
+						Version: 1,
+					},
+				},
+			),
+		)
+	}
+	rand.Shuffle(len(metas), func(i, j int) {
+		metas[i], metas[j] = metas[j], metas[i]
+	})
+
+	shipper := New(nil, nil, dir, nil, nil, metadata.TestSource)
+	if err := shipper.iterBlockMetas(func(m *metadata.Meta) error {
+		metas = append(metas, m)
+		return nil
+	}); err != nil {
+		testutil.Ok(t, err)
+	}
+	testutil.Equals(t, sort.SliceIsSorted(metas, func(i, j int) bool {
+		return metas[i].BlockMeta.MinTime < metas[j].BlockMeta.MinTime
+	}), true)
+}
+
+func BenchmarkIterBlockMetas(b *testing.B) {
+	var metas []*metadata.Meta
+	dir, err := ioutil.TempDir("", "shipper-test")
+	testutil.Ok(b, err)
+	defer func() {
+		testutil.Ok(b, os.RemoveAll(dir))
+	}()
+
+	for i := 0; i < 100; i++ {
+		id := ulid.MustNew(uint64(i), nil)
+		testutil.Ok(b, os.Mkdir(path.Join(dir, id.String()), os.ModePerm))
+		testutil.Ok(b,
+			metadata.Write(
+				log.NewNopLogger(),
+				path.Join(dir, id.String()),
+				&metadata.Meta{
+					BlockMeta: tsdb.BlockMeta{
+						ULID:    id,
+						MaxTime: int64((i + 1) * 1000),
+						MinTime: int64(i * 1000),
+						Version: 1,
+					},
+				},
+			),
+		)
+	}
+	rand.Shuffle(len(metas), func(i, j int) {
+		metas[i], metas[j] = metas[j], metas[i]
+	})
+	b.ResetTimer()
+
+	shipper := New(nil, nil, dir, nil, nil, metadata.TestSource)
+	if err := shipper.iterBlockMetas(func(m *metadata.Meta) error {
+		metas = append(metas, m)
+		return nil
+	}); err != nil {
+		testutil.Ok(b, err)
+	}
+}