fix: avoid aws-chunked uploads for compatible object stores (#51349)

issue: #50567

use `minio.disableAWSChunkedEncoding` to disable the aws chunked
Encoding.

Signed-off-by: jiaqizho <jiaqi.zhou@zilliz.com>

internal/storage/minio_object_storage.go

@@ -76,10 +76,19 @@ func (minioObjectStorage *MinioObjectStorage) GetObject(ctx context.Context, buc
 }
 
 func (minioObjectStorage *MinioObjectStorage) PutObject(ctx context.Context, bucketName, objectName string, reader io.Reader, objectSize int64) error {
-	_, err := minioObjectStorage.Client.PutObject(ctx, bucketName, objectName, reader, objectSize, minio.PutObjectOptions{})
+	_, err := minioObjectStorage.Client.PutObject(ctx, bucketName, objectName, reader, objectSize, minioObjectStorage.putObjectOptions())
 	return mapObjectStorageError(objectName, err)
 }
 
+func (minioObjectStorage *MinioObjectStorage) putObjectOptions() minio.PutObjectOptions {
+	if !paramtable.Get().MinioCfg.DisableAWSChunkedEncoding.GetAsBool() {
+		return minio.PutObjectOptions{}
+	}
+	return minio.PutObjectOptions{
+		DisableContentSha256: true,
+	}
+}
+
 func (minioObjectStorage *MinioObjectStorage) StatObject(ctx context.Context, bucketName, objectName string) (int64, error) {
 	info, err := minioObjectStorage.Client.StatObject(ctx, bucketName, objectName, minio.StatObjectOptions{})
 	return info.Size, mapObjectStorageError(objectName, err)

internal/storage/minio_object_storage_test.go

@@ -31,8 +31,27 @@ import (
 
 	"github.com/milvus-io/milvus/pkg/v3/objectstorage"
 	"github.com/milvus-io/milvus/pkg/v3/util/merr"
+	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
+func TestMinioObjectStoragePutObjectOptions(t *testing.T) {
+	params := paramtable.Get()
+	params.Save(params.MinioCfg.DisableAWSChunkedEncoding.Key, "false")
+	t.Cleanup(func() {
+		params.Reset(params.MinioCfg.DisableAWSChunkedEncoding.Key)
+	})
+
+	storage := &MinioObjectStorage{}
+	opts := storage.putObjectOptions()
+	assert.False(t, opts.DisableContentSha256)
+	assert.False(t, opts.SendContentMd5)
+
+	params.Save(params.MinioCfg.DisableAWSChunkedEncoding.Key, "true")
+	opts = storage.putObjectOptions()
+	assert.True(t, opts.DisableContentSha256)
+	assert.False(t, opts.SendContentMd5)
+}
+
 func TestMinioObjectStorage(t *testing.T) {
 	ctx := context.Background()
 	config := objectstorage.Config{

pkg/util/paramtable/service_param.go

@@ -1496,6 +1496,8 @@ type MinioConfig struct {
 	MaxConnections     ParamItem `refreshable:"false"`
 	ListObjectsMaxKeys ParamItem `refreshable:"true"`
 	UseCRC32C          ParamItem `refreshable:"false"`
+
+	DisableAWSChunkedEncoding ParamItem `refreshable:"false"`
 }
 
 func (p *MinioConfig) Init(base *BaseTable) {
@@ -1570,6 +1572,16 @@ The default value applies to MinIO or S3 service that started with the default d
 	}
 	p.UseSSL.Init(base.mgr)
 
+	p.DisableAWSChunkedEncoding = ParamItem{
+		Key:          "minio.disableAWSChunkedEncoding",
+		Version:      "2.6.20",
+		DefaultValue: "false",
+		Doc: `When enabled, PutObject requests use UNSIGNED-PAYLOAD to support S3-compatible endpoints that are incompatible with AWS chunked encoding.
+HTTPS is recommended because payload integrity is then protected by TLS rather than a signed payload hash.`,
+		Export: false,
+	}
+	p.DisableAWSChunkedEncoding.Init(base.mgr)
+
 	p.SslCACert = ParamItem{
 		Key:          "minio.ssl.tlsCACert",
 		Version:      "2.3.12",

pkg/util/paramtable/service_param_test.go

@@ -366,6 +366,8 @@ func TestServiceParam(t *testing.T) {
 
 		assert.Equal(t, Params.UseSSL.GetAsBool(), false)
 
+		assert.False(t, Params.DisableAWSChunkedEncoding.GetAsBool())
+
 		assert.Empty(t, Params.SslCACert.GetValue())
 
 		assert.Equal(t, Params.UseIAM.GetAsBool(), false)