Fix S3 BucketWithRetries upload empty content issue

CHANGELOG.md

@@ -48,6 +48,7 @@
 * [BUGFIX] Query-frontend: Fix shardable instant queries do not produce sorted results for `sort`, `sort_desc`, `topk`, `bottomk` functions. #5148, #5170
 * [BUGFIX] Querier: Fix `/api/v1/series` returning 5XX instead of 4XX when limits are hit. #5169
 * [BUGFIX] Compactor: Fix issue that shuffle sharding planner return error if block is under visit by other compactor. #5188
+* [BUGFIX] Fix S3 BucketWithRetries upload empty content issue #5217
 * [FEATURE] Alertmanager: Add support for time_intervals. #5102
 
 ## 1.14.0 2022-12-02

pkg/storage/bucket/s3/bucket_client.go

@@ -1,6 +1,7 @@
 package s3
 
 import (
+	"bytes"
 	"context"
 	"io"
 	"time"
@@ -153,8 +154,18 @@ func (b *BucketWithRetries) Exists(ctx context.Context, name string) (exists boo
 }
 
 func (b *BucketWithRetries) Upload(ctx context.Context, name string, r io.Reader) error {
+	// Convert Reader to Seeker
+	var buf bytes.Buffer
+	if _, err := buf.ReadFrom(r); err != nil {
+		return err
+	}
+	s := bytes.NewReader(buf.Bytes())
 	return b.retry(ctx, func() error {
-		return b.bucket.Upload(ctx, name, r)
+		err := b.bucket.Upload(ctx, name, s)
+		if _, err := s.Seek(0, io.SeekStart); err != nil {
+			return err
+		}
+		return err
 	})
 }
 

pkg/storage/bucket/s3/bucket_client_test.go

@@ -0,0 +1,96 @@
+package s3
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/objstore"
+)
+
+func TestBucketWithRetries_Upload(t *testing.T) {
+	t.Parallel()
+
+	m := mockBucket{
+		MaxFailCount: 3,
+	}
+	b := BucketWithRetries{
+		bucket:           &m,
+		operationRetries: 5,
+		retryMinBackoff:  10 * time.Millisecond,
+		retryMaxBackoff:  time.Second,
+	}
+
+	input := []byte("test input")
+	err := b.Upload(context.Background(), "dummy", bytes.NewReader(input))
+	require.NoError(t, err)
+	require.Equal(t, input, m.uploadedContent)
+}
+
+type mockBucket struct {
+	MaxFailCount    int
+	uploadedContent []byte
+}
+
+// Upload mocks objstore.Bucket.Upload()
+func (m *mockBucket) Upload(ctx context.Context, name string, r io.Reader) error {
+	var buf bytes.Buffer
+	if _, err := buf.ReadFrom(r); err != nil {
+		return err
+	}
+	m.uploadedContent = buf.Bytes()
+	if m.MaxFailCount > 0 {
+		m.MaxFailCount--
+		return fmt.Errorf("failed upload: %d", m.MaxFailCount)
+	}
+	return nil
+}
+
+// Delete mocks objstore.Bucket.Delete()
+func (m *mockBucket) Delete(ctx context.Context, name string) error {
+	return nil
+}
+
+// Name mocks objstore.Bucket.Name()
+func (m *mockBucket) Name() string {
+	return "mock"
+}
+
+// Iter mocks objstore.Bucket.Iter()
+func (m *mockBucket) Iter(ctx context.Context, dir string, f func(string) error, options ...objstore.IterOption) error {
+	return nil
+}
+
+// Get mocks objstore.Bucket.Get()
+func (m *mockBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
+	return nil, nil
+}
+
+// GetRange mocks objstore.Bucket.GetRange()
+func (m *mockBucket) GetRange(ctx context.Context, name string, off, length int64) (io.ReadCloser, error) {
+	return nil, nil
+}
+
+// Exists mocks objstore.Bucket.Exists()
+func (m *mockBucket) Exists(ctx context.Context, name string) (bool, error) {
+	return false, nil
+}
+
+// IsObjNotFoundErr mocks objstore.Bucket.IsObjNotFoundErr()
+func (m *mockBucket) IsObjNotFoundErr(err error) bool {
+	return false
+}
+
+// ObjectSize mocks objstore.Bucket.Attributes()
+func (m *mockBucket) Attributes(ctx context.Context, name string) (objstore.ObjectAttributes, error) {
+	return objstore.ObjectAttributes{Size: 0, LastModified: time.Now()}, nil
+}
+
+// Close mocks objstore.Bucket.Close()
+func (m *mockBucket) Close() error {
+	return nil
+}