From 87332eb60bb61186379135833ba27c86bbe51c5d Mon Sep 17 00:00:00 2001 From: jackusm Date: Tue, 15 Sep 2026 06:09:08 +0100 Subject: [PATCH] Cloud/remote storage & tiering: configurable multipart upload/download concurrency (#11319) * pb: add multipart concurrency fields to RemoteConf and tier move requests RemoteConf gains upload_concurrency/download_concurrency (0 = client default); VolumeTierMoveDatToRemote/FromRemote requests gain a concurrency field (0 = backend default). * remote storage: honor RemoteConf upload/download concurrency in s3 and azure clients s3 client: ReadFile passes conf download_concurrency to the downloader, WriteFile uses upload_concurrency for the uploader; previously hard-coded 1 upload / 5 download parts. 0 keeps defaults. Same for azure client. * storage: plumb concurrency through backend interface and tier upload/download BackendStorage.CopyFile/DownloadFile take a concurrency hint (<=0 = backend configured default); s3 backend reads upload_concurrency/download_concurrency from scaffold config with parseConcurrency fallback, rclone updated to the new signature. Tier move gRPC handlers forward the request concurrency to the backend. * shell: -upload_concurrency/-download_concurrency for remote.configure, -concurrent for volume.tier remote.configure exposes upload/download concurrency persisted into RemoteConf; volume.tier move/evict commands forward -concurrent to the tier move requests. Documented in master-cloud.toml scaffold. * test: cover concurrency propagation in remote tier integration test * remote.configure: merge existing config on partial update Load the stored RemoteConf before saving so a partial update (e.g. only -upload_concurrency) preserves credentials, endpoints, and type instead of replacing them with new-config defaults. Only treat a confirmed ErrNotFound as a new configuration; propagate all other load errors so a transient filer failure does not overwrite stored settings. On a type transition, reset backend-specific fields to the destination type's new-config defaults rather than inheriting the old backend's empty values. Bound configured concurrency to a sane maximum. * remote storage: honor configured download concurrency in S3 and Azure ReadFileWithConcurrency now resolves a zero request override against the client's configured download_concurrency (new downloadConcurrency() helpers), so the remote-mount/cache read path honors RemoteConf.DownloadConcurrency instead of the hard-coded default. Azure also clamps the resolved value to math.MaxUint16 regardless of whether the fallback was used, preventing uint16 wraparound when a configured value exceeds 65535. * shell: rename -concurrent to -concurrency and validate tier transfer bounds Rename the -concurrent flag to -concurrency across volume.tier.upload, volume.tier.download, and volume.tier.compact to match the proto field and RemoteConf field names. Add validateTierConcurrency to reject values that would wrap int32 or exceed a 1024 cap before constructing the request. * server: clamp tier move concurrency in gRPC handlers Add clampTierConcurrency to both VolumeTierMoveDatToRemote and VolumeTierMoveDatFromRemote handlers so a direct gRPC caller cannot spawn an unbounded number of network workers. * trim verbose comments added with concurrency feature Remove redundant doc comments on the backend interface, rclone backend, s3_backend parseConcurrency, and test helpers that restated the obvious. * remote.configure: apply type defaults before re-parse so explicit flags win applyTypeDefaults ran after the second flag parse, overwriting explicit destination flags (e.g. -s3.region=eu-west-1) with new-config defaults. Move the type-transition default reset before the re-parse so user-supplied flags override the destination defaults. * remote.configure: only treat explicit -type as a type transition The first parse defaults -type to s3, so a concurrency-only update on an existing non-S3 config captured requestedType=s3 and wrongly triggered a type transition, resetting the stored backend to S3. Use fs.Visit to detect whether -type was explicitly supplied; an omitted -type keeps the stored backend. --------- Co-authored-by: Jack Meredith <9480542+jackusm@users.noreply.github.com> Co-authored-by: Chris Lu --- docker/compose/master-cloud.toml | 2 + weed/pb/remote.proto | 3 + weed/pb/remote_pb/remote.pb.go | 22 +- weed/pb/volume_server.proto | 2 + weed/pb/volume_server_pb/volume_server.pb.go | 26 +- .../azure/azure_storage_client.go | 24 +- weed/remote_storage/s3/s3_storage_client.go | 25 +- weed/server/volume_grpc_tier_download.go | 2 +- weed/server/volume_grpc_tier_download_test.go | 62 ++++- weed/server/volume_grpc_tier_test.go | 4 +- weed/server/volume_grpc_tier_upload.go | 16 +- weed/shell/command_remote_configure.go | 241 +++++++++++++----- weed/shell/command_volume_tier_compact.go | 19 +- weed/shell/command_volume_tier_download.go | 19 +- weed/shell/command_volume_tier_upload.go | 35 ++- weed/storage/backend/backend.go | 5 +- .../backend/rclone_backend/rclone_backend.go | 4 +- weed/storage/backend/s3_backend/s3_backend.go | 38 ++- .../backend/s3_backend/s3_backend_test.go | 53 ++++ .../storage/backend/s3_backend/s3_download.go | 7 +- weed/storage/backend/s3_backend/s3_upload.go | 7 +- weed/storage/remote_tier_integration_test.go | 8 +- 22 files changed, 506 insertions(+), 118 deletions(-) diff --git a/docker/compose/master-cloud.toml b/docker/compose/master-cloud.toml index 29324772c..27bf3f0fe 100644 --- a/docker/compose/master-cloud.toml +++ b/docker/compose/master-cloud.toml @@ -30,3 +30,5 @@ sleep_minutes = 17 # sleep minutes between each script execution bucket = "volume_bucket" # an existing bucket endpoint = "http://server2:8333" storage_class = "STANDARD_IA" + # upload_concurrency = 5 # concurrent multipart part uploads per volume (volume.tier.upload -concurrent overrides) + # download_concurrency = 5 # concurrent multipart part downloads per volume (volume.tier.download -concurrent overrides) diff --git a/weed/pb/remote.proto b/weed/pb/remote.proto index c5d81eda5..2f50a3a73 100644 --- a/weed/pb/remote.proto +++ b/weed/pb/remote.proto @@ -65,6 +65,9 @@ message RemoteConf { string contabo_secret_key = 69; string contabo_endpoint = 70; string contabo_region = 71; + + uint32 upload_concurrency = 72; // multipart upload concurrency per file, 0 = client default (1 for S3, 16 for Azure) + uint32 download_concurrency = 73; // multipart download concurrency per read, 0 = client default (5 for S3, 16 for Azure) } message RemoteStorageMapping { diff --git a/weed/pb/remote_pb/remote.pb.go b/weed/pb/remote_pb/remote.pb.go index e246d2744..daf6cb789 100644 --- a/weed/pb/remote_pb/remote.pb.go +++ b/weed/pb/remote_pb/remote.pb.go @@ -71,6 +71,8 @@ type RemoteConf struct { ContaboSecretKey string `protobuf:"bytes,69,opt,name=contabo_secret_key,json=contaboSecretKey,proto3" json:"contabo_secret_key,omitempty"` ContaboEndpoint string `protobuf:"bytes,70,opt,name=contabo_endpoint,json=contaboEndpoint,proto3" json:"contabo_endpoint,omitempty"` ContaboRegion string `protobuf:"bytes,71,opt,name=contabo_region,json=contaboRegion,proto3" json:"contabo_region,omitempty"` + UploadConcurrency uint32 `protobuf:"varint,72,opt,name=upload_concurrency,json=uploadConcurrency,proto3" json:"upload_concurrency,omitempty"` // multipart upload concurrency per file, 0 = client default (1 for S3, 16 for Azure) + DownloadConcurrency uint32 `protobuf:"varint,73,opt,name=download_concurrency,json=downloadConcurrency,proto3" json:"download_concurrency,omitempty"` // multipart download concurrency per read, 0 = client default (5 for S3, 16 for Azure) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -420,6 +422,20 @@ func (x *RemoteConf) GetContaboRegion() string { return "" } +func (x *RemoteConf) GetUploadConcurrency() uint32 { + if x != nil { + return x.UploadConcurrency + } + return 0 +} + +func (x *RemoteConf) GetDownloadConcurrency() uint32 { + if x != nil { + return x.DownloadConcurrency + } + return 0 +} + type RemoteStorageMapping struct { state protoimpl.MessageState `protogen:"open.v1"` Mappings map[string]*RemoteStorageLocation `protobuf:"bytes,1,rep,name=mappings,proto3" json:"mappings,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` @@ -552,7 +568,7 @@ var File_remote_proto protoreflect.FileDescriptor const file_remote_proto_rawDesc = "" + "\n" + - "\fremote.proto\x12\tremote_pb\"\xea\x0e\n" + + "\fremote.proto\x12\tremote_pb\"\xcc\x0f\n" + "\n" + "RemoteConf\x12\x12\n" + "\x04type\x18\x01 \x01(\tR\x04type\x12\x12\n" + @@ -601,7 +617,9 @@ const file_remote_proto_rawDesc = "" + "\x12contabo_access_key\x18D \x01(\tR\x10contaboAccessKey\x12,\n" + "\x12contabo_secret_key\x18E \x01(\tR\x10contaboSecretKey\x12)\n" + "\x10contabo_endpoint\x18F \x01(\tR\x0fcontaboEndpoint\x12%\n" + - "\x0econtabo_region\x18G \x01(\tR\rcontaboRegion\"\xff\x01\n" + + "\x0econtabo_region\x18G \x01(\tR\rcontaboRegion\x12-\n" + + "\x12upload_concurrency\x18H \x01(\rR\x11uploadConcurrency\x121\n" + + "\x14download_concurrency\x18I \x01(\rR\x13downloadConcurrency\"\xff\x01\n" + "\x14RemoteStorageMapping\x12I\n" + "\bmappings\x18\x01 \x03(\v2-.remote_pb.RemoteStorageMapping.MappingsEntryR\bmappings\x12=\n" + "\x1bprimary_bucket_storage_name\x18\x02 \x01(\tR\x18primaryBucketStorageName\x1a]\n" + diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index de1bb235b..4fddb2f4e 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -659,6 +659,7 @@ message VolumeTierMoveDatToRemoteRequest { string collection = 2; string destination_backend_name = 3; bool keep_local_dat_file = 4; + int32 concurrency = 5; // multipart upload concurrency, 0 = backend default (5 for S3) } message VolumeTierMoveDatToRemoteResponse { int64 processed = 1; @@ -669,6 +670,7 @@ message VolumeTierMoveDatFromRemoteRequest { uint32 volume_id = 1; string collection = 2; bool keep_remote_dat_file = 3; + int32 concurrency = 4; // multipart download concurrency, 0 = backend default (5 for S3) } message VolumeTierMoveDatFromRemoteResponse { int64 processed = 1; diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 30f0e4258..6e7c220eb 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -5471,6 +5471,7 @@ type VolumeTierMoveDatToRemoteRequest struct { Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"` DestinationBackendName string `protobuf:"bytes,3,opt,name=destination_backend_name,json=destinationBackendName,proto3" json:"destination_backend_name,omitempty"` KeepLocalDatFile bool `protobuf:"varint,4,opt,name=keep_local_dat_file,json=keepLocalDatFile,proto3" json:"keep_local_dat_file,omitempty"` + Concurrency int32 `protobuf:"varint,5,opt,name=concurrency,proto3" json:"concurrency,omitempty"` // multipart upload concurrency, 0 = backend default (5 for S3) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -5533,6 +5534,13 @@ func (x *VolumeTierMoveDatToRemoteRequest) GetKeepLocalDatFile() bool { return false } +func (x *VolumeTierMoveDatToRemoteRequest) GetConcurrency() int32 { + if x != nil { + return x.Concurrency + } + return 0 +} + type VolumeTierMoveDatToRemoteResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Processed int64 `protobuf:"varint,1,opt,name=processed,proto3" json:"processed,omitempty"` @@ -5590,6 +5598,7 @@ type VolumeTierMoveDatFromRemoteRequest struct { VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"` KeepRemoteDatFile bool `protobuf:"varint,3,opt,name=keep_remote_dat_file,json=keepRemoteDatFile,proto3" json:"keep_remote_dat_file,omitempty"` + Concurrency int32 `protobuf:"varint,4,opt,name=concurrency,proto3" json:"concurrency,omitempty"` // multipart download concurrency, 0 = backend default (5 for S3) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -5645,6 +5654,13 @@ func (x *VolumeTierMoveDatFromRemoteRequest) GetKeepRemoteDatFile() bool { return false } +func (x *VolumeTierMoveDatFromRemoteRequest) GetConcurrency() int32 { + if x != nil { + return x.Concurrency + } + return 0 +} + type VolumeTierMoveDatFromRemoteResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Processed int64 `protobuf:"varint,1,opt,name=processed,proto3" json:"processed,omitempty"` @@ -7644,23 +7660,25 @@ const file_volume_server_proto_rawDesc = "" + "\vBytesOffset\x18\x04 \x01(\rR\vBytesOffset\x12\"\n" + "\rdat_file_size\x18\x05 \x01(\x03R\vdatFileSize\x12 \n" + "\vDestroyTime\x18\x06 \x01(\x04R\vDestroyTime\x12\x1b\n" + - "\tread_only\x18\a \x01(\bR\breadOnly\"\xc8\x01\n" + + "\tread_only\x18\a \x01(\bR\breadOnly\"\xea\x01\n" + " VolumeTierMoveDatToRemoteRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" + "\n" + "collection\x18\x02 \x01(\tR\n" + "collection\x128\n" + "\x18destination_backend_name\x18\x03 \x01(\tR\x16destinationBackendName\x12-\n" + - "\x13keep_local_dat_file\x18\x04 \x01(\bR\x10keepLocalDatFile\"s\n" + + "\x13keep_local_dat_file\x18\x04 \x01(\bR\x10keepLocalDatFile\x12 \n" + + "\vconcurrency\x18\x05 \x01(\x05R\vconcurrency\"s\n" + "!VolumeTierMoveDatToRemoteResponse\x12\x1c\n" + "\tprocessed\x18\x01 \x01(\x03R\tprocessed\x120\n" + - "\x13processedPercentage\x18\x02 \x01(\x02R\x13processedPercentage\"\x92\x01\n" + + "\x13processedPercentage\x18\x02 \x01(\x02R\x13processedPercentage\"\xb4\x01\n" + "\"VolumeTierMoveDatFromRemoteRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" + "\n" + "collection\x18\x02 \x01(\tR\n" + "collection\x12/\n" + - "\x14keep_remote_dat_file\x18\x03 \x01(\bR\x11keepRemoteDatFile\"u\n" + + "\x14keep_remote_dat_file\x18\x03 \x01(\bR\x11keepRemoteDatFile\x12 \n" + + "\vconcurrency\x18\x04 \x01(\x05R\vconcurrency\"u\n" + "#VolumeTierMoveDatFromRemoteResponse\x12\x1c\n" + "\tprocessed\x18\x01 \x01(\x03R\tprocessed\x120\n" + "\x13processedPercentage\x18\x02 \x01(\x02R\x13processedPercentage\"\x1b\n" + diff --git a/weed/remote_storage/azure/azure_storage_client.go b/weed/remote_storage/azure/azure_storage_client.go index f72ffc8c2..b1446acb9 100644 --- a/weed/remote_storage/azure/azure_storage_client.go +++ b/weed/remote_storage/azure/azure_storage_client.go @@ -153,6 +153,20 @@ type azureRemoteStorageClient struct { var _ = remote_storage.RemoteStorageClient(&azureRemoteStorageClient{}) var _ = remote_storage.RemoteStorageConcurrentReader(&azureRemoteStorageClient{}) +func (az *azureRemoteStorageClient) uploadConcurrency() int { + if n := int(az.conf.GetUploadConcurrency()); n > 0 { + return n + } + return defaultConcurrency +} + +func (az *azureRemoteStorageClient) downloadConcurrency() int { + if n := int(az.conf.GetDownloadConcurrency()); n > 0 { + return n + } + return defaultReadConcurrency +} + func (az *azureRemoteStorageClient) ListDirectory(ctx context.Context, loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) { pathKey := loc.Path[1:] if pathKey != "" && !strings.HasSuffix(pathKey, "/") { @@ -303,7 +317,7 @@ func (az *azureRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocatio } func (az *azureRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (data []byte, err error) { - return az.ReadFileWithConcurrency(loc, offset, size, defaultReadConcurrency) + return az.ReadFileWithConcurrency(loc, offset, size, 0) } // ReadFileWithConcurrency fetches a byte range of a blob using the Azure SDK's @@ -322,9 +336,9 @@ func (az *azureRemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.Remot } if concurrency <= 0 { - concurrency = defaultReadConcurrency - } else if concurrency > math.MaxUint16 { - // DownloadBufferOptions.Concurrency is uint16; clamp to avoid wraparound. + concurrency = az.downloadConcurrency() + } + if concurrency > math.MaxUint16 { concurrency = math.MaxUint16 } @@ -448,7 +462,7 @@ func (az *azureRemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocati _, err = blobClient.UploadStream(context.Background(), reader, &blockblob.UploadStreamOptions{ BlockSize: defaultBlockSize, - Concurrency: defaultConcurrency, + Concurrency: az.uploadConcurrency(), HTTPHeaders: httpHeaders, Metadata: metadata, }) diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index 5aa4f2d69..1f126f7c5 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -229,6 +229,25 @@ type s3RemoteStorageClient struct { conn s3iface.S3API } +const ( + defaultUploadConcurrency = 1 + defaultReadConcurrency = 5 +) + +func (s *s3RemoteStorageClient) uploadConcurrency() int { + if n := int(s.conf.GetUploadConcurrency()); n > 0 { + return n + } + return defaultUploadConcurrency +} + +func (s *s3RemoteStorageClient) downloadConcurrency() int { + if n := int(s.conf.GetDownloadConcurrency()); n > 0 { + return n + } + return defaultReadConcurrency +} + var _ = remote_storage.RemoteStorageClient(&s3RemoteStorageClient{}) func (s *s3RemoteStorageClient) Traverse(remote *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) { @@ -374,12 +393,12 @@ func (s *s3RemoteStorageClient) StatFile(loc *remote_pb.RemoteStorageLocation) ( } func (s *s3RemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (data []byte, err error) { - return s.ReadFileWithConcurrency(loc, offset, size, 5) + return s.ReadFileWithConcurrency(loc, offset, size, 0) } func (s *s3RemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.RemoteStorageLocation, offset int64, size int64, concurrency int) (data []byte, err error) { if concurrency <= 0 { - concurrency = 5 + concurrency = s.downloadConcurrency() } downloader := s3manager.NewDownloaderWithClient(s.conn, func(u *s3manager.Downloader) { u.PartSize = int64(4 * 1024 * 1024) @@ -493,7 +512,7 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation, // Create an uploader with the session and custom options uploader := s3manager.NewUploaderWithClient(s.conn, func(u *s3manager.Uploader) { u.PartSize = partSize - u.Concurrency = 1 + u.Concurrency = s.uploadConcurrency() }) // process tagging diff --git a/weed/server/volume_grpc_tier_download.go b/weed/server/volume_grpc_tier_download.go index 32e65979b..16461b757 100644 --- a/weed/server/volume_grpc_tier_download.go +++ b/weed/server/volume_grpc_tier_download.go @@ -70,7 +70,7 @@ func (vs *VolumeServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.Volume } // copy the data file (DownloadFile opens, fsyncs, and closes the .dat internally) datFileName := v.FileName(".dat") - _, err := backendStorage.DownloadFile(datFileName, storageKey, fn) + _, err := backendStorage.DownloadFile(datFileName, storageKey, fn, clampTierConcurrency(int(req.Concurrency))) if err != nil { return fmt.Errorf("backend %s copy file %s: %v", storageName, datFileName, err) } diff --git a/weed/server/volume_grpc_tier_download_test.go b/weed/server/volume_grpc_tier_download_test.go index fdca2a8aa..29aee51e6 100644 --- a/weed/server/volume_grpc_tier_download_test.go +++ b/weed/server/volume_grpc_tier_download_test.go @@ -32,6 +32,9 @@ type tierTestBackend struct { mu sync.Mutex deletes []string + + lastUploadConcurrency int + lastDownloadConcurrency int } func (b *tierTestBackend) ToProperties() map[string]string { return map[string]string{"root": b.root} } @@ -40,7 +43,10 @@ func (b *tierTestBackend) NewStorageFile(key string, tierInfo *volume_server_pb. return &tierTestBackendFile{backend: b, key: key, tierInfo: tierInfo} } -func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) { +func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) { + b.mu.Lock() + b.lastUploadConcurrency = concurrency + b.mu.Unlock() key = fmt.Sprintf("obj-%d", time.Now().UnixNano()) dst := filepath.Join(b.root, key) out, err := os.Create(dst) @@ -58,7 +64,10 @@ func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percent return key, written, nil } -func (b *tierTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) { +func (b *tierTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) { + b.mu.Lock() + b.lastDownloadConcurrency = concurrency + b.mu.Unlock() in, err := os.Open(filepath.Join(b.root, key)) if err != nil { return 0, err @@ -105,6 +114,12 @@ func (b *tierTestBackend) objectExists(key string) bool { return err == nil } +func (b *tierTestBackend) concurrencySeen() (upload, download int) { + b.mu.Lock() + defer b.mu.Unlock() + return b.lastUploadConcurrency, b.lastDownloadConcurrency +} + type tierTestBackendFile struct { backend *tierTestBackend key string @@ -201,7 +216,7 @@ func tierUpVolumeOnDisk(t *testing.T, dir string, vid needle.VolumeId, b *tierTe t.Fatalf("expected on-disk backend before tier-up, got %T", v.DataBackend) } datPath := v.FileName(".dat") - key, size, err := b.CopyFile(diskFile.File, nil) + key, size, err := b.CopyFile(diskFile.File, nil, 0) if err != nil { t.Fatalf("upload to fake backend: %v", err) } @@ -326,3 +341,44 @@ func TestTierMoveDatFromRemote_KeepRemote_LeavesReplicaLocal(t *testing.T) { t.Fatalf("local .dat content mismatch: got %d bytes, want %d", len(gotDat), len(localDat)) } } + +// TestTierMoveConcurrencyPlumbing verifies the -concurrency value travels the +// whole tier-move path to CopyFile and DownloadFile. +func TestTierMoveConcurrencyPlumbing(t *testing.T) { + b := &tierTestBackend{root: t.TempDir()} + backend.BackendStorages[tierTestBackendName] = b + t.Cleanup(func() { delete(backend.BackendStorages, tierTestBackendName) }) + + dir := t.TempDir() + const vid = needle.VolumeId(72) + tierUpVolumeOnDisk(t, dir, vid, b) + + // download path: request concurrency must reach DownloadFile + store := newTierTestStore(t, dir) + v := store.GetVolume(vid) + if v == nil { + t.Fatal("tiered volume not loaded by store") + } + vs := &VolumeServer{store: store} + if err := vs.VolumeTierMoveDatFromRemote(&volume_server_pb.VolumeTierMoveDatFromRemoteRequest{ + VolumeId: uint32(vid), + Concurrency: 4, + }, &fakeTierStream{}); err != nil { + t.Fatalf("VolumeTierMoveDatFromRemote: %v", err) + } + if up, down := b.concurrencySeen(); down != 4 { + t.Fatalf("DownloadFile concurrency = %d (upload=%d), want 4", down, up) + } + + // upload path: the volume is local again after the download + if err := vs.VolumeTierMoveDatToRemote(&volume_server_pb.VolumeTierMoveDatToRemoteRequest{ + VolumeId: uint32(vid), + DestinationBackendName: tierTestBackendName, + Concurrency: 3, + }, &discardServerStream[volume_server_pb.VolumeTierMoveDatToRemoteResponse]{}); err != nil { + t.Fatalf("VolumeTierMoveDatToRemote: %v", err) + } + if up, down := b.concurrencySeen(); up != 3 || down != 4 { + t.Fatalf("CopyFile concurrency = %d (download=%d), want 3/4", up, down) + } +} diff --git a/weed/server/volume_grpc_tier_test.go b/weed/server/volume_grpc_tier_test.go index f042948de..ee41d619e 100644 --- a/weed/server/volume_grpc_tier_test.go +++ b/weed/server/volume_grpc_tier_test.go @@ -52,7 +52,7 @@ func (b *tierTimestampTestBackend) NewStorageFile(key string, volumeInfo *volume } } -func (b *tierTimestampTestBackend) CopyFile(file *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) { +func (b *tierTimestampTestBackend) CopyFile(file *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) { key = "remote.dat" fileInfo, err := file.Stat() if err != nil { @@ -72,7 +72,7 @@ func (b *tierTimestampTestBackend) CopyFile(file *os.File, fn func(progressed in return key, size, err } -func (b *tierTimestampTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) { +func (b *tierTimestampTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) { input, err := os.Open(filepath.Join(b.root, key)) if err != nil { return 0, err diff --git a/weed/server/volume_grpc_tier_upload.go b/weed/server/volume_grpc_tier_upload.go index c6355c8c1..f4af5f0e1 100644 --- a/weed/server/volume_grpc_tier_upload.go +++ b/weed/server/volume_grpc_tier_upload.go @@ -10,6 +10,20 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" ) +// clampTierConcurrency bounds a per-request transfer concurrency so a direct +// gRPC caller cannot spawn an unbounded number of network workers. +const maxTierConcurrency = 1024 + +func clampTierConcurrency(n int) int { + if n < 0 { + return 0 + } + if n > maxTierConcurrency { + return maxTierConcurrency + } + return n +} + // VolumeTierMoveDatToRemote copy dat file to a remote tier func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTierMoveDatToRemoteRequest, stream volume_server_pb.VolumeServer_VolumeTierMoveDatToRemoteServer) error { if err := vs.checkGrpcAdminAuth(stream.Context()); err != nil { @@ -73,7 +87,7 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi } // copy the data file - key, size, err := backendStorage.CopyFile(diskFile.File, fn) + key, size, err := backendStorage.CopyFile(diskFile.File, fn, clampTierConcurrency(int(req.Concurrency))) if err != nil { return fmt.Errorf("backend %s copy file %s: %v", req.DestinationBackendName, diskFile.Name(), err) } diff --git a/weed/shell/command_remote_configure.go b/weed/shell/command_remote_configure.go index 8f754f0e6..0abceb155 100644 --- a/weed/shell/command_remote_configure.go +++ b/weed/shell/command_remote_configure.go @@ -2,6 +2,7 @@ package shell import ( "context" + "errors" "flag" "fmt" "io" @@ -45,6 +46,9 @@ func (c *commandRemoteConfigure) Help() string { remote.configure -name=cloud7 -type=storj -storj.access_key=xxx -storj.secret_key=yyy -storj.endpoint=https://gateway.us1.storjshare.io remote.configure -name=cloud8 -type=filebase -filebase.access_key=xxx -filebase.secret_key=yyy -filebase.endpoint=https://s3.filebase.com + # tune transfer concurrency (applies to s3-compatible and azure storage) + remote.configure -name=cloud1 -upload_concurrency=4 -download_concurrency=8 + # delete one configuration remote.configure -delete -name=cloud1 @@ -63,71 +67,11 @@ func (c *commandRemoteConfigure) Do(args []string, commandEnv *CommandEnv, write conf := &remote_pb.RemoteConf{} - remoteConfigureCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) - isDelete := remoteConfigureCommand.Bool("delete", false, "delete one remote storage by its name") - - remoteConfigureCommand.StringVar(&conf.Name, "name", "", "a short name to identify the remote storage") - remoteConfigureCommand.StringVar(&conf.Type, "type", "s3", fmt.Sprintf("[%s] storage type", remote_storage.GetAllRemoteStorageNames())) - - remoteConfigureCommand.StringVar(&conf.S3AccessKey, "s3.access_key", "", "s3 access key") - remoteConfigureCommand.StringVar(&conf.S3SecretKey, "s3.secret_key", "", "s3 secret key") - remoteConfigureCommand.StringVar(&conf.S3Region, "s3.region", "us-east-2", "s3 region") - remoteConfigureCommand.StringVar(&conf.S3Endpoint, "s3.endpoint", "", "endpoint for s3-compatible local object store") - remoteConfigureCommand.StringVar(&conf.S3StorageClass, "s3.storage_class", "", "s3 storage class") - remoteConfigureCommand.BoolVar(&conf.S3ForcePathStyle, "s3.force_path_style", true, "s3 force path style") - remoteConfigureCommand.BoolVar(&conf.S3V4Signature, "s3.v4_signature", false, "s3 V4 signature") - remoteConfigureCommand.BoolVar(&conf.S3SupportTagging, "s3.support_tagging", true, "s3 supportTagging") - - remoteConfigureCommand.StringVar(&conf.GcsGoogleApplicationCredentials, "gcs.appCredentialsFile", "", "google cloud storage credentials file, default to use env GOOGLE_APPLICATION_CREDENTIALS") - remoteConfigureCommand.StringVar(&conf.GcsProjectId, "gcs.projectId", "", "google cloud storage project id, default to use env GOOGLE_CLOUD_PROJECT") - - remoteConfigureCommand.StringVar(&conf.AzureAccountName, "azure.account_name", "", "azure account name, default to use env AZURE_STORAGE_ACCOUNT") - remoteConfigureCommand.StringVar(&conf.AzureAccountKey, "azure.account_key", "", "azure account key, default to use env AZURE_STORAGE_ACCESS_KEY. Leave empty to authenticate with Entra ID") - remoteConfigureCommand.StringVar(&conf.AzureClientId, "azure.client_id", "", "azure user-assigned identity to authenticate, when no account key is given. Workload identity also reads env AZURE_TENANT_ID and AZURE_FEDERATED_TOKEN_FILE") - remoteConfigureCommand.StringVar(&conf.AzureEndpoint, "azure.endpoint", "", "azure blob service url, for accounts outside the public cloud, e.g. https://xxx.blob.core.usgovcloudapi.net/") - - remoteConfigureCommand.StringVar(&conf.BackblazeKeyId, "b2.key_id", "", "backblaze keyID") - remoteConfigureCommand.StringVar(&conf.BackblazeApplicationKey, "b2.application_key", "", "backblaze applicationKey. Note that your Master Application Key will not work with the S3 Compatible API. You must create a new key that is eligible for use. For more information: https://help.backblaze.com/hc/en-us/articles/360047425453") - remoteConfigureCommand.StringVar(&conf.BackblazeEndpoint, "b2.endpoint", "", "backblaze endpoint") - remoteConfigureCommand.StringVar(&conf.BackblazeRegion, "b2.region", "us-west-002", "backblaze region") - - remoteConfigureCommand.StringVar(&conf.AliyunAccessKey, "aliyun.access_key", "", "Aliyun access key, default to use env ALICLOUD_ACCESS_KEY_ID") - remoteConfigureCommand.StringVar(&conf.AliyunSecretKey, "aliyun.secret_key", "", "Aliyun secret key, default to use env ALICLOUD_ACCESS_KEY_SECRET") - remoteConfigureCommand.StringVar(&conf.AliyunEndpoint, "aliyun.endpoint", "", "Aliyun endpoint") - remoteConfigureCommand.StringVar(&conf.AliyunRegion, "aliyun.region", "", "Aliyun region") - - remoteConfigureCommand.StringVar(&conf.TencentSecretId, "tencent.secret_id", "", "Tencent Secret Id, default to use env COS_SECRETID") - remoteConfigureCommand.StringVar(&conf.TencentSecretKey, "tencent.secret_key", "", "Tencent secret key, default to use env COS_SECRETKEY") - remoteConfigureCommand.StringVar(&conf.TencentEndpoint, "tencent.endpoint", "", "Tencent endpoint") - - remoteConfigureCommand.StringVar(&conf.BaiduAccessKey, "baidu.access_key", "", "Baidu access key, default to use env BDCLOUD_ACCESS_KEY") - remoteConfigureCommand.StringVar(&conf.BaiduSecretKey, "baidu.secret_key", "", "Baidu secret key, default to use env BDCLOUD_SECRET_KEY") - remoteConfigureCommand.StringVar(&conf.BaiduEndpoint, "baidu.endpoint", "", "Baidu endpoint") - remoteConfigureCommand.StringVar(&conf.BaiduRegion, "baidu.region", "", "Baidu region") - - remoteConfigureCommand.StringVar(&conf.WasabiAccessKey, "wasabi.access_key", "", "Wasabi access key") - remoteConfigureCommand.StringVar(&conf.WasabiSecretKey, "wasabi.secret_key", "", "Wasabi secret key") - remoteConfigureCommand.StringVar(&conf.WasabiEndpoint, "wasabi.endpoint", "", "Wasabi endpoint, see https://wasabi.com/wp-content/themes/wasabi/docs/API_Guide/index.html#t=topics%2Fapidiff-intro.htm") - remoteConfigureCommand.StringVar(&conf.WasabiRegion, "wasabi.region", "", "Wasabi region") - - remoteConfigureCommand.StringVar(&conf.FilebaseAccessKey, "filebase.access_key", "", "Filebase access key") - remoteConfigureCommand.StringVar(&conf.FilebaseSecretKey, "filebase.secret_key", "", "Filebase secret key") - remoteConfigureCommand.StringVar(&conf.FilebaseEndpoint, "filebase.endpoint", "", "Filebase endpoint, https://s3.filebase.com") - - remoteConfigureCommand.StringVar(&conf.StorjAccessKey, "storj.access_key", "", "Storj access key") - remoteConfigureCommand.StringVar(&conf.StorjSecretKey, "storj.secret_key", "", "Storj secret key") - remoteConfigureCommand.StringVar(&conf.StorjEndpoint, "storj.endpoint", "", "Storj endpoint") - - if err = remoteConfigureCommand.Parse(args); err != nil { + fs, isDelete, uploadConcurrency, downloadConcurrency := c.configureFlagSet(conf, false) + if err = fs.Parse(args); err != nil { return nil } - if conf.Type != "s3" { - // clear out the default values - conf.S3Region = "" - conf.S3ForcePathStyle = false - } - if conf.Name == "" { return c.listExistingRemoteStorages(commandEnv, writer) } @@ -140,10 +84,183 @@ func (c *commandRemoteConfigure) Do(args []string, commandEnv *CommandEnv, write return c.deleteRemoteStorage(commandEnv, writer, conf.Name) } + // Merge with an existing configuration so a partial update preserves + // previously stored credentials and endpoints. Only treat a confirmed + // missing entry as a new configuration; propagate all other load errors + // so a transient filer failure does not overwrite stored credentials. + typeExplicit := false + fs.Visit(func(f *flag.Flag) { + if f.Name == "type" { + typeExplicit = true + } + }) + requestedType := conf.Type + existing, loadErr := c.loadRemoteStorageConf(commandEnv, conf.Name) + if loadErr != nil && !errors.Is(loadErr, filer_pb.ErrNotFound) { + return fmt.Errorf("load existing configuration %s: %v", conf.Name, loadErr) + } + if existing != nil { + conf = existing + // On an explicit type transition, reset backend-specific fields to + // the destination defaults before re-parsing so explicit flags + // override. An omitted -type keeps the stored backend. + if typeExplicit && requestedType != existing.Type { + conf.Type = requestedType + c.applyTypeDefaults(conf) + } + fs, isDelete, uploadConcurrency, downloadConcurrency = c.configureFlagSet(conf, true) + if err = fs.Parse(args); err != nil { + return nil + } + } + + if err = applyConcurrency(conf, *uploadConcurrency, *downloadConcurrency); err != nil { + return err + } + + if conf.Type != "s3" { + conf.S3Region = "" + conf.S3ForcePathStyle = false + } + return c.saveRemoteStorage(commandEnv, writer, conf) } +// configureFlagSet builds the remote.configure flag set bound to conf. When +// existing is true, flag defaults are taken from conf so omitted flags preserve +// prior values instead of being reset to the hard-coded new-config defaults. +func (c *commandRemoteConfigure) configureFlagSet(conf *remote_pb.RemoteConf, existing bool) (fs *flag.FlagSet, isDelete *bool, uploadConcurrency *int, downloadConcurrency *int) { + fs = flag.NewFlagSet(c.Name(), flag.ContinueOnError) + isDelete = fs.Bool("delete", false, "delete one remote storage by its name") + + fs.StringVar(&conf.Name, "name", conf.Name, "a short name to identify the remote storage") + typeDefault := "s3" + if existing { + typeDefault = conf.Type + } + fs.StringVar(&conf.Type, "type", typeDefault, fmt.Sprintf("[%s] storage type", remote_storage.GetAllRemoteStorageNames())) + + uploadConcurrency = fs.Int("upload_concurrency", int(conf.UploadConcurrency), "concurrent part uploads per file (0 = client default: s3 1, azure 16)") + downloadConcurrency = fs.Int("download_concurrency", int(conf.DownloadConcurrency), "concurrent part downloads per read (0 = client default: s3 5, azure 16)") + + fs.StringVar(&conf.S3AccessKey, "s3.access_key", conf.S3AccessKey, "s3 access key") + fs.StringVar(&conf.S3SecretKey, "s3.secret_key", conf.S3SecretKey, "s3 secret key") + s3RegionDefault := "us-east-2" + if existing { + s3RegionDefault = conf.S3Region + } + fs.StringVar(&conf.S3Region, "s3.region", s3RegionDefault, "s3 region") + fs.StringVar(&conf.S3Endpoint, "s3.endpoint", conf.S3Endpoint, "endpoint for s3-compatible local object store") + fs.StringVar(&conf.S3StorageClass, "s3.storage_class", conf.S3StorageClass, "s3 storage class") + s3ForcePathStyleDefault := true + if existing { + s3ForcePathStyleDefault = conf.S3ForcePathStyle + } + fs.BoolVar(&conf.S3ForcePathStyle, "s3.force_path_style", s3ForcePathStyleDefault, "s3 force path style") + fs.BoolVar(&conf.S3V4Signature, "s3.v4_signature", conf.S3V4Signature, "s3 V4 signature") + s3SupportTaggingDefault := true + if existing { + s3SupportTaggingDefault = conf.S3SupportTagging + } + fs.BoolVar(&conf.S3SupportTagging, "s3.support_tagging", s3SupportTaggingDefault, "s3 supportTagging") + + fs.StringVar(&conf.GcsGoogleApplicationCredentials, "gcs.appCredentialsFile", conf.GcsGoogleApplicationCredentials, "google cloud storage credentials file, default to use env GOOGLE_APPLICATION_CREDENTIALS") + fs.StringVar(&conf.GcsProjectId, "gcs.projectId", conf.GcsProjectId, "google cloud storage project id, default to use env GOOGLE_CLOUD_PROJECT") + + fs.StringVar(&conf.AzureAccountName, "azure.account_name", conf.AzureAccountName, "azure account name, default to use env AZURE_STORAGE_ACCOUNT") + fs.StringVar(&conf.AzureAccountKey, "azure.account_key", conf.AzureAccountKey, "azure account key, default to use env AZURE_STORAGE_ACCESS_KEY. Leave empty to authenticate with Entra ID") + fs.StringVar(&conf.AzureClientId, "azure.client_id", conf.AzureClientId, "azure user-assigned identity to authenticate, when no account key is given. Workload identity also reads env AZURE_TENANT_ID and AZURE_FEDERATED_TOKEN_FILE") + fs.StringVar(&conf.AzureEndpoint, "azure.endpoint", conf.AzureEndpoint, "azure blob service url, for accounts outside the public cloud, e.g. https://xxx.blob.core.usgovcloudapi.net/") + + fs.StringVar(&conf.BackblazeKeyId, "b2.key_id", conf.BackblazeKeyId, "backblaze keyID") + fs.StringVar(&conf.BackblazeApplicationKey, "b2.application_key", conf.BackblazeApplicationKey, "backblaze applicationKey. Note that your Master Application Key will not work with the S3 Compatible API. You must create a new key that is eligible for use. For more information: https://help.backblaze.com/hc/en-us/articles/360047425453") + fs.StringVar(&conf.BackblazeEndpoint, "b2.endpoint", conf.BackblazeEndpoint, "backblaze endpoint") + b2RegionDefault := "us-west-002" + if existing { + b2RegionDefault = conf.BackblazeRegion + } + fs.StringVar(&conf.BackblazeRegion, "b2.region", b2RegionDefault, "backblaze region") + + fs.StringVar(&conf.AliyunAccessKey, "aliyun.access_key", conf.AliyunAccessKey, "Aliyun access key, default to use env ALICLOUD_ACCESS_KEY_ID") + fs.StringVar(&conf.AliyunSecretKey, "aliyun.secret_key", conf.AliyunSecretKey, "Aliyun secret key, default to use env ALICLOUD_ACCESS_KEY_SECRET") + fs.StringVar(&conf.AliyunEndpoint, "aliyun.endpoint", conf.AliyunEndpoint, "Aliyun endpoint") + fs.StringVar(&conf.AliyunRegion, "aliyun.region", conf.AliyunRegion, "Aliyun region") + + fs.StringVar(&conf.TencentSecretId, "tencent.secret_id", conf.TencentSecretId, "Tencent Secret Id, default to use env COS_SECRETID") + fs.StringVar(&conf.TencentSecretKey, "tencent.secret_key", conf.TencentSecretKey, "Tencent secret key, default to use env COS_SECRETKEY") + fs.StringVar(&conf.TencentEndpoint, "tencent.endpoint", conf.TencentEndpoint, "Tencent endpoint") + + fs.StringVar(&conf.BaiduAccessKey, "baidu.access_key", conf.BaiduAccessKey, "Baidu access key, default to use env BDCLOUD_ACCESS_KEY") + fs.StringVar(&conf.BaiduSecretKey, "baidu.secret_key", conf.BaiduSecretKey, "Baidu secret key, default to use env BDCLOUD_SECRET_KEY") + fs.StringVar(&conf.BaiduEndpoint, "baidu.endpoint", conf.BaiduEndpoint, "Baidu endpoint") + fs.StringVar(&conf.BaiduRegion, "baidu.region", conf.BaiduRegion, "Baidu region") + + fs.StringVar(&conf.WasabiAccessKey, "wasabi.access_key", conf.WasabiAccessKey, "Wasabi access key") + fs.StringVar(&conf.WasabiSecretKey, "wasabi.secret_key", conf.WasabiSecretKey, "Wasabi secret key") + fs.StringVar(&conf.WasabiEndpoint, "wasabi.endpoint", conf.WasabiEndpoint, "Wasabi endpoint, see https://wasabi.com/wp-content/themes/wasabi/docs/API_Guide/index.html#t=topics%2Fapidiff-intro.htm") + fs.StringVar(&conf.WasabiRegion, "wasabi.region", conf.WasabiRegion, "Wasabi region") + + fs.StringVar(&conf.FilebaseAccessKey, "filebase.access_key", conf.FilebaseAccessKey, "Filebase access key") + fs.StringVar(&conf.FilebaseSecretKey, "filebase.secret_key", conf.FilebaseSecretKey, "Filebase secret key") + fs.StringVar(&conf.FilebaseEndpoint, "filebase.endpoint", conf.FilebaseEndpoint, "Filebase endpoint, https://s3.filebase.com") + + fs.StringVar(&conf.StorjAccessKey, "storj.access_key", conf.StorjAccessKey, "Storj access key") + fs.StringVar(&conf.StorjSecretKey, "storj.secret_key", conf.StorjSecretKey, "Storj secret key") + fs.StringVar(&conf.StorjEndpoint, "storj.endpoint", conf.StorjEndpoint, "Storj endpoint") + + return +} + +// maxRemoteConcurrency caps configured remote transfer concurrency so one +// configuration cannot request an unbounded number of network workers. +const maxRemoteConcurrency = 1024 + +// applyTypeDefaults resets backend-specific fields to their new-config defaults +// for conf.Type, used when the storage type changes on update. +func (c *commandRemoteConfigure) applyTypeDefaults(conf *remote_pb.RemoteConf) { + conf.S3Region = "" + conf.S3ForcePathStyle = false + conf.S3SupportTagging = false + conf.S3V4Signature = false + conf.BackblazeRegion = "" + if conf.Type == "s3" { + conf.S3Region = "us-east-2" + conf.S3ForcePathStyle = true + conf.S3SupportTagging = true + } else if conf.Type == "b2" { + conf.BackblazeRegion = "us-west-002" + } +} + +func applyConcurrency(conf *remote_pb.RemoteConf, upload, download int) error { + if upload < 0 || upload > maxRemoteConcurrency { + return fmt.Errorf("upload_concurrency must be between 0 and %d", maxRemoteConcurrency) + } + if download < 0 || download > maxRemoteConcurrency { + return fmt.Errorf("download_concurrency must be between 0 and %d", maxRemoteConcurrency) + } + conf.UploadConcurrency = uint32(upload) + conf.DownloadConcurrency = uint32(download) + return nil +} + +func (c *commandRemoteConfigure) loadRemoteStorageConf(commandEnv *CommandEnv, name string) (*remote_pb.RemoteConf, error) { + var conf *remote_pb.RemoteConf + err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + content, readErr := filer.ReadInsideFiler(context.Background(), client, filer.DirectoryEtcRemote, name+filer.REMOTE_STORAGE_CONF_SUFFIX) + if readErr != nil { + return readErr + } + conf = &remote_pb.RemoteConf{} + if unmarshalErr := proto.Unmarshal(content, conf); unmarshalErr != nil { + return fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, name, unmarshalErr) + } + return nil + }) + return conf, err +} + func (c *commandRemoteConfigure) listExistingRemoteStorages(commandEnv *CommandEnv, writer io.Writer) error { return filer_pb.ReadDirAllEntries(context.Background(), commandEnv, util.FullPath(filer.DirectoryEtcRemote), "", func(entry *filer_pb.Entry, isLast bool) error { diff --git a/weed/shell/command_volume_tier_compact.go b/weed/shell/command_volume_tier_compact.go index c73313d52..fd55958b5 100644 --- a/weed/shell/command_volume_tier_compact.go +++ b/weed/shell/command_volume_tier_compact.go @@ -29,8 +29,8 @@ func (c *commandVolumeTierCompact) Name() string { func (c *commandVolumeTierCompact) Help() string { return `compact remote volumes to reclaim space on cloud storage - volume.tier.compact [-volumeId=] - volume.tier.compact [-collection=""] [-garbageThreshold=0.3] + volume.tier.compact [-volumeId=] [-concurrency=] + volume.tier.compact [-collection=""] [-garbageThreshold=0.3] [-concurrency=] e.g.: volume.tier.compact -volumeId=7 @@ -64,10 +64,15 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri volumeId := tierCommand.Int("volumeId", 0, "the volume id") collection := tierCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name") garbageThreshold := tierCommand.Float64("garbageThreshold", 0.3, "compact when garbage ratio exceeds this value") + concurrency := tierCommand.Int("concurrency", 0, "multipart transfer concurrency (0 = backend default)") if err = tierCommand.Parse(args); err != nil { return nil } + if err = validateTierConcurrency(*concurrency); err != nil { + return err + } + if err = commandEnv.confirmIsLocked(args); err != nil { return } @@ -107,7 +112,7 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri var failedCount int for _, rv := range remoteVolumes { - if err = doVolumeTierCompact(commandEnv, writer, rv, *garbageThreshold); err != nil { + if err = doVolumeTierCompact(commandEnv, writer, rv, *garbageThreshold, *concurrency); err != nil { fmt.Fprintf(writer, "error compacting volume %d: %v\n", rv.vid, err) failedCount++ } @@ -194,7 +199,7 @@ func collectRemoteVolumesWithInfo(topoInfo *master_pb.TopologyInfo, collectionPa return result, nil } -func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolumeInfo, garbageThreshold float64) error { +func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolumeInfo, garbageThreshold float64, concurrency int) error { grpcDialOption := commandEnv.option.GrpcDialOption // step 1: check garbage level @@ -214,7 +219,7 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu // step 2: download .dat from remote to local // this deletes the remote file and reloads the volume as local, then re-uploads below fmt.Fprintf(writer, " downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName) - err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false) + err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false, concurrency) if err != nil { return fmt.Errorf("download volume %d from remote: %v", rv.vid, err) } @@ -227,7 +232,7 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu // upload the uncompacted volume back to restore cloud tier state fmt.Fprintf(writer, " compaction failed: %v\n", err) fmt.Fprintf(writer, " re-uploading volume %d to %s without compaction...\n", rv.vid, rv.remoteStorageName) - uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false) + uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency) if uploadErr != nil { return fmt.Errorf("compaction failed (%v) and re-upload also failed (%v), volume %d remains local", err, uploadErr, rv.vid) @@ -238,7 +243,7 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu // step 4: upload compacted volume back to remote fmt.Fprintf(writer, " uploading compacted volume %d to %s...\n", rv.vid, rv.remoteStorageName) - err = uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false) + err = uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency) if err != nil { return fmt.Errorf("upload compacted volume %d to %s: %v (volume remains local with compacted data)", rv.vid, rv.remoteStorageName, err) diff --git a/weed/shell/command_volume_tier_download.go b/weed/shell/command_volume_tier_download.go index 43ddda76d..79668744c 100644 --- a/weed/shell/command_volume_tier_download.go +++ b/weed/shell/command_volume_tier_download.go @@ -32,7 +32,7 @@ func (c *commandVolumeTierDownload) Help() string { return `download the dat file of a volume from a remote tier volume.tier.download [-collection=""] - volume.tier.download [-collection=""] -volumeId= + volume.tier.download [-collection=""] -volumeId= [-concurrency=] The -collection parameter supports regular expressions for pattern matching: - Use exact match: volume.tier.download -collection="^mybucket$" @@ -41,6 +41,7 @@ func (c *commandVolumeTierDownload) Help() string { e.g.: volume.tier.download -volumeId=7 + volume.tier.download -volumeId=7 -concurrency=1 This command will download the dat file of a volume from a remote tier to a volume server in local cluster. @@ -56,10 +57,15 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) volumeId := tierCommand.Int("volumeId", 0, "the volume id") collection := tierCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name") + concurrency := tierCommand.Int("concurrency", 0, "multipart download concurrency (0 = backend default)") if err = tierCommand.Parse(args); err != nil { return nil } + if err = validateTierConcurrency(*concurrency); err != nil { + return err + } + if err = commandEnv.confirmIsLocked(args); err != nil { return } @@ -74,7 +80,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr // volumeId is provided if vid != 0 { - return doVolumeTierDownload(commandEnv, writer, *collection, vid) + return doVolumeTierDownload(commandEnv, writer, *collection, vid, *concurrency) } // apply to all volumes in the collection @@ -85,7 +91,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr } fmt.Printf("tier download volumes: %v\n", volumeIds) for _, vid := range volumeIds { - if err = doVolumeTierDownload(commandEnv, writer, *collection, vid); err != nil { + if err = doVolumeTierDownload(commandEnv, writer, *collection, vid, *concurrency); err != nil { return err } } @@ -118,7 +124,7 @@ func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, collectionPattern st return } -func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId) (err error) { +func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, concurrency int) (err error) { // find volume location locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(vid)) if !found { @@ -131,7 +137,7 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s for i, loc := range locations { keepRemote := i < len(locations)-1 // copy the .dat file from remote tier to local - err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote) + err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote, concurrency) if err != nil { // A replica already made local by a prior interrupted run is not a // failure; skip it so the remaining remote replicas still download. @@ -146,13 +152,14 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s return nil } -func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool) error { +func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool, concurrency int) error { err := operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { stream, downloadErr := volumeServerClient.VolumeTierMoveDatFromRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatFromRemoteRequest{ VolumeId: uint32(volumeId), Collection: collection, KeepRemoteDatFile: keepRemote, + Concurrency: int32(concurrency), }) var lastProcessed int64 diff --git a/weed/shell/command_volume_tier_upload.go b/weed/shell/command_volume_tier_upload.go index 4bf83e124..df21d99dc 100644 --- a/weed/shell/command_volume_tier_upload.go +++ b/weed/shell/command_volume_tier_upload.go @@ -5,6 +5,7 @@ import ( "flag" "fmt" "io" + "math" "time" "github.com/seaweedfs/seaweedfs/weed/storage/types" @@ -21,6 +22,21 @@ import ( "github.com/seaweedfs/seaweedfs/weed/wdclient" ) +// maxTierConcurrency caps per-request multipart transfer concurrency. +const maxTierConcurrency = 1024 + +// validateTierConcurrency rejects values that would wrap when narrowed to the +// int32 proto field or exhaust volume-server resources. 0 means backend default. +func validateTierConcurrency(n int) error { + if n < 0 || n > math.MaxInt32 { + return fmt.Errorf("concurrency must be between 0 and %d, got %d", math.MaxInt32, n) + } + if n > maxTierConcurrency { + return fmt.Errorf("concurrency must be at most %d, got %d", maxTierConcurrency, n) + } + return nil +} + func init() { Commands = append(Commands, &commandVolumeTierUpload{}) } @@ -36,11 +52,12 @@ func (c *commandVolumeTierUpload) Help() string { return `upload the dat file of a volume to a remote tier volume.tier.upload [-collection=""] [-fullPercent=95] [-quietFor=1h] - volume.tier.upload [-collection=""] -volumeId= -dest= [-keepLocalDatFile] + volume.tier.upload [-collection=""] -volumeId= -dest= [-keepLocalDatFile] [-concurrency=] e.g.: volume.tier.upload -volumeId=7 -dest=s3 volume.tier.upload -volumeId=7 -dest=s3.default + volume.tier.upload -volumeId=7 -dest=s3.telegram -concurrency=1 The is defined in master.toml. For example, "s3.default" in [storage.backend.s3.default] @@ -78,10 +95,15 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ dest := tierCommand.String("dest", "", "the target tier name") keepLocalDatFile := tierCommand.Bool("keepLocalDatFile", false, "whether keep local dat file") disk := tierCommand.String("disk", "", "[hdd|ssd|] hard drive or solid state drive or any tag") + concurrency := tierCommand.Int("concurrency", 0, "multipart upload concurrency (0 = backend default)") if err = tierCommand.Parse(args); err != nil { return nil } + if err = validateTierConcurrency(*concurrency); err != nil { + return err + } + if err = commandEnv.confirmIsLocked(args); err != nil { return } @@ -90,7 +112,7 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ // volumeId is provided if vid != 0 { - return doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile) + return doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile, *concurrency) } var diskType *types.DiskType @@ -107,7 +129,7 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ } fmt.Printf("tier upload volumes: %v\n", volumeIds) for _, vid := range volumeIds { - if err = doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile); err != nil { + if err = doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile, *concurrency); err != nil { return err } } @@ -115,7 +137,7 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ return nil } -func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool) (err error) { +func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool, concurrency int) (err error) { // find volume location topoInfo, _, err := collectTopologyInfo(commandEnv, 0) if err != nil { @@ -137,7 +159,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str } // copy the .dat file to remote tier - err = uploadDatToRemoteTier(commandEnv.option.GrpcDialOption, writer, vid, collection, existingLocations[0].ServerAddress(), dest, keepLocalDatFile) + err = uploadDatToRemoteTier(commandEnv.option.GrpcDialOption, writer, vid, collection, existingLocations[0].ServerAddress(), dest, keepLocalDatFile, concurrency) if err != nil { return fmt.Errorf("copy dat file for volume %d on %s to %s: %v", vid, existingLocations[0].Url, dest, err) } @@ -190,7 +212,7 @@ func collectVolumeTierUploadLocations(topoInfo *master_pb.TopologyInfo, vid need return append(tiered, local...) } -func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, sourceVolumeServer pb.ServerAddress, dest string, keepLocalDatFile bool) error { +func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, sourceVolumeServer pb.ServerAddress, dest string, keepLocalDatFile bool, concurrency int) error { err := operation.WithVolumeServerClient(true, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { stream, copyErr := volumeServerClient.VolumeTierMoveDatToRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatToRemoteRequest{ @@ -198,6 +220,7 @@ func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, vol Collection: collection, DestinationBackendName: dest, KeepLocalDatFile: keepLocalDatFile, + Concurrency: int32(concurrency), }) if stream == nil { diff --git a/weed/storage/backend/backend.go b/weed/storage/backend/backend.go index 539867f8f..632adb6c3 100644 --- a/weed/storage/backend/backend.go +++ b/weed/storage/backend/backend.go @@ -25,8 +25,9 @@ type BackendStorageFile interface { type BackendStorage interface { ToProperties() map[string]string NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) BackendStorageFile - CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) - DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) + // concurrency > 0 caps concurrent network transfers; <= 0 uses the backend default. + CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) + DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) DeleteFile(key string) (err error) } diff --git a/weed/storage/backend/rclone_backend/rclone_backend.go b/weed/storage/backend/rclone_backend/rclone_backend.go index 1fe7aca76..daeaac0ea 100644 --- a/weed/storage/backend/rclone_backend/rclone_backend.go +++ b/weed/storage/backend/rclone_backend/rclone_backend.go @@ -103,7 +103,7 @@ func (s *RcloneBackendStorage) NewStorageFile(key string, tierInfo *volume_serve return f } -func (s *RcloneBackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) { +func (s *RcloneBackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) { randomUuid, err := uuid.NewRandom() if err != nil { return key, 0, err @@ -154,7 +154,7 @@ func uploadViaRclone(rfs fs.Fs, filename string, key string, fn func(progressed return obj.Size(), err } -func (s *RcloneBackendStorage) DownloadFile(filename string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) { +func (s *RcloneBackendStorage) DownloadFile(filename string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) { glog.V(1).Infof("download dat file of %s from remote rclone.%s as %s", filename, s.id, key) util.Retry("download via Rclone", func() error { diff --git a/weed/storage/backend/s3_backend/s3_backend.go b/weed/storage/backend/s3_backend/s3_backend.go index 1e0b21700..882f21e99 100644 --- a/weed/storage/backend/s3_backend/s3_backend.go +++ b/weed/storage/backend/s3_backend/s3_backend.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "os" + "strconv" "strings" "time" @@ -41,6 +42,8 @@ type S3BackendStorage struct { endpoint string storageClass string forcePathStyle bool + uploadConcurrency int + downloadConcurrency int conn s3iface.S3API } @@ -54,6 +57,8 @@ func newS3BackendStorage(configuration backend.StringProperties, configPrefix st s.endpoint = configuration.GetString(configPrefix + "endpoint") s.storageClass = configuration.GetString(configPrefix + "storage_class") s.forcePathStyle = util.ParseBool(configuration.GetString(configPrefix+"force_path_style"), true) + s.uploadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"upload_concurrency"), defaultUploadConcurrency) + s.downloadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"download_concurrency"), defaultDownloadConcurrency) if s.storageClass == "" { s.storageClass = "STANDARD_IA" } @@ -73,9 +78,27 @@ func (s *S3BackendStorage) ToProperties() map[string]string { m["endpoint"] = s.endpoint m["storage_class"] = s.storageClass m["force_path_style"] = util.BoolToString(s.forcePathStyle) + m["upload_concurrency"] = strconv.Itoa(s.uploadConcurrency) + m["download_concurrency"] = strconv.Itoa(s.downloadConcurrency) return m } +const ( + defaultUploadConcurrency = 5 + defaultDownloadConcurrency = 5 +) + +func parseConcurrency(value string, def int) int { + if value == "" { + return def + } + if n, err := strconv.Atoi(value); err == nil && n > 0 { + return n + } + glog.Warningf("invalid concurrency value %q, using default %d", value, def) + return def +} + func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile { if strings.HasPrefix(key, "/") { key = key[1:] @@ -90,25 +113,32 @@ func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb return f } -func (s *S3BackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) { +func (s *S3BackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) { randomUuid, _ := uuid.NewRandom() key = randomUuid.String() + if concurrency <= 0 { + concurrency = s.uploadConcurrency + } + glog.V(1).Infof("copying dat file of %s to remote s3.%s as %s", f.Name(), s.id, key) util.Retry("upload to S3", func() error { - size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, s.storageClass, fn) + size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, s.storageClass, fn, concurrency) return err }) return } -func (s *S3BackendStorage) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) { +func (s *S3BackendStorage) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) { + if concurrency <= 0 { + concurrency = s.downloadConcurrency + } glog.V(1).Infof("download dat file of %s from remote s3.%s as %s", fileName, s.id, key) - size, err = downloadFromS3(s.conn, fileName, s.bucket, key, fn) + size, err = downloadFromS3(s.conn, fileName, s.bucket, key, fn, concurrency) return } diff --git a/weed/storage/backend/s3_backend/s3_backend_test.go b/weed/storage/backend/s3_backend/s3_backend_test.go index 35537a6d3..347c37982 100644 --- a/weed/storage/backend/s3_backend/s3_backend_test.go +++ b/weed/storage/backend/s3_backend/s3_backend_test.go @@ -12,6 +12,12 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" ) +type testProperties map[string]string + +func (m testProperties) GetString(key string) string { + return m[key] +} + type stubS3Client struct { s3iface.S3API getObject func(*s3.GetObjectInput) (*s3.GetObjectOutput, error) @@ -101,3 +107,50 @@ func TestReadAtRejectsInvalidRequestsLocally(t *testing.T) { }) } } + +func TestParseConcurrency(t *testing.T) { + cases := []struct { + value string + def int + expected int + }{ + {"", 5, 5}, // unset -> default + {"0", 5, 5}, // explicit zero means default + {"-3", 5, 5}, // invalid -> default + {"abc", 5, 5}, // invalid -> default + {"1", 5, 1}, // override + {"64", 5, 64}, // override + } + for _, tt := range cases { + if got := parseConcurrency(tt.value, tt.def); got != tt.expected { + t.Errorf("parseConcurrency(%q, %d) = %d, want %d", tt.value, tt.def, got, tt.expected) + } + } +} + +func TestS3BackendStorageConcurrencyConfigRoundTrip(t *testing.T) { + s, err := newS3BackendStorage(testProperties{}, "", "test") + if err != nil { + t.Fatal(err) + } + if s.uploadConcurrency != defaultUploadConcurrency || s.downloadConcurrency != defaultDownloadConcurrency { + t.Fatalf("defaults: upload=%d download=%d, want %d/%d", + s.uploadConcurrency, s.downloadConcurrency, defaultUploadConcurrency, defaultDownloadConcurrency) + } + + s, err = newS3BackendStorage(testProperties{ + "upload_concurrency": "1", + "download_concurrency": "17", + }, "", "test") + if err != nil { + t.Fatal(err) + } + if s.uploadConcurrency != 1 || s.downloadConcurrency != 17 { + t.Fatalf("configured: upload=%d download=%d, want 1/17", s.uploadConcurrency, s.downloadConcurrency) + } + + props := s.ToProperties() + if props["upload_concurrency"] != "1" || props["download_concurrency"] != "17" { + t.Fatalf("ToProperties: %v", props) + } +} diff --git a/weed/storage/backend/s3_backend/s3_download.go b/weed/storage/backend/s3_backend/s3_download.go index 82d73d71b..b0928e9be 100644 --- a/weed/storage/backend/s3_backend/s3_download.go +++ b/weed/storage/backend/s3_backend/s3_download.go @@ -14,7 +14,10 @@ import ( ) func downloadFromS3(sess s3iface.S3API, destFileName string, sourceBucket string, sourceKey string, - fn func(progressed int64, percentage float32) error) (fileSize int64, err error) { + fn func(progressed int64, percentage float32) error, concurrency int) (fileSize int64, err error) { + if concurrency <= 0 { + concurrency = defaultDownloadConcurrency + } fileSize, err = getFileSize(sess, sourceBucket, sourceKey) if err != nil { @@ -31,7 +34,7 @@ func downloadFromS3(sess s3iface.S3API, destFileName string, sourceBucket string // Create a downloader with the session and custom options downloader := s3manager.NewDownloaderWithClient(sess, func(u *s3manager.Downloader) { u.PartSize = int64(64 * 1024 * 1024) - u.Concurrency = 5 + u.Concurrency = concurrency }) fileWriter := &s3DownloadProgressedWriter{ diff --git a/weed/storage/backend/s3_backend/s3_upload.go b/weed/storage/backend/s3_backend/s3_upload.go index 1392ad23a..61bcbe5a8 100644 --- a/weed/storage/backend/s3_backend/s3_upload.go +++ b/weed/storage/backend/s3_backend/s3_upload.go @@ -12,7 +12,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" ) -func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string, storageClass string, fn func(progressed int64, percentage float32) error) (fileSize int64, err error) { +func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string, storageClass string, fn func(progressed int64, percentage float32) error, concurrency int) (fileSize int64, err error) { + if concurrency <= 0 { + concurrency = defaultUploadConcurrency + } //open the file f, err := os.Open(filename) @@ -36,7 +39,7 @@ func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey // Create an uploader with the session and custom options uploader := s3manager.NewUploaderWithClient(sess, func(u *s3manager.Uploader) { u.PartSize = partSize - u.Concurrency = 5 + u.Concurrency = concurrency }) fileReader := &s3UploadProgressedReader{ diff --git a/weed/storage/remote_tier_integration_test.go b/weed/storage/remote_tier_integration_test.go index 27f79bf37..fc59258e5 100644 --- a/weed/storage/remote_tier_integration_test.go +++ b/weed/storage/remote_tier_integration_test.go @@ -52,7 +52,7 @@ func (b *localDirBackend) NewStorageFile(key string, tierInfo *volume_server_pb. return &localDirBackendFile{backend: b, key: key, tierInfo: tierInfo} } -func (b *localDirBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) { +func (b *localDirBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) { key = fmt.Sprintf("obj-%d-%d", time.Now().UnixNano(), os.Getpid()) dst := filepath.Join(b.root, key) out, err := os.Create(dst) @@ -73,7 +73,7 @@ func (b *localDirBackend) CopyFile(f *os.File, fn func(progressed int64, percent return key, written, nil } -func (b *localDirBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) { +func (b *localDirBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) { src := filepath.Join(b.root, key) in, err := os.Open(src) if err != nil { @@ -193,7 +193,7 @@ func tierUpVolumeLive(t *testing.T, dir string, vid needle.VolumeId, b *localDir diskFile, ok := v.DataBackend.(*backend.DiskFile) require.True(t, ok, "expected on-disk backend before tier-up") - uploadKey, size, err := b.CopyFile(diskFile.File, nil) + uploadKey, size, err := b.CopyFile(diskFile.File, nil, 0) require.NoError(t, err) bType, bId := backend.BackendNameToTypeId(testBackendName) @@ -422,7 +422,7 @@ func TestRemoteTier_ECEncodeDecode_AfterDownload(t *testing.T) { baseFileName := filepath.Join(dir, fmt.Sprintf("%d", uint32(vid))) datPath := baseFileName + ".dat" - _, err := b.DownloadFile(datPath, key, nil) + _, err := b.DownloadFile(datPath, key, nil, 0) require.NoError(t, err) require.NoError(t, erasure_coding.WriteSortedFileFromIdx(baseFileName, ".ecx"))