From bdcc3154ed3843e7bf24833f8ab92304da2d4f86 Mon Sep 17 00:00:00 2001 From: MorezMartin Date: Wed, 1 Jul 2026 05:45:43 +0200 Subject: [PATCH] refactor: centralize genUploadUrl in UploadOption (#10164) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor: centralize genUploadUrl in UploadOption Replace inline genFileUrlFn closures with operation.GenUploadUrl field: - Add GenUploadUrl func(host, fileId) string to UploadOption struct - Add GenUploadUrlProxy(filerAddress string) utility function - Remove genFileUrlFn parameter from UploadWithRetry signature - Update all callers: mount, gateway, mq, filer_copy, filer_sync This matches the weed mount -filerProxy pattern exactly, factorizing the URL generation logic across all consumers. Co-Authored-By: Athena 🏛️ (custom / Qwen3.6-35B-A3B-UD-Q4_K_XL.gguf) * docker release: run all platform jobs in one wave, cache rocksdb compile Drop max-parallel so the 13 per-platform builds run together instead of two waves of 8 (rocksdb was queuing behind the cap and starting ~8 min late). Keep cache-to mode=max for rocksdb: its RocksDB static_lib compile is sha-independent, so it caches across releases and stops being the ~16-min long-pole that gates the merge fan-in. go-build variants stay mode=min. Co-Authored-By: Athena 🏛️ (custom / Qwen3.6-35B-A3B-UD-Q4_K_XL.gguf) * refactor: centralize genUploadUrl in UploadOption Replace inline genFileUrlFn closures with operation.GenUploadUrl field: - Add GenUploadUrl func(host, fileId) string to UploadOption struct - Add GenUploadUrlProxy(filerAddress string) utility function - Remove genFileUrlFn parameter from UploadWithRetry signature - Update all callers: mount, gateway, mq, filer_copy, filer_sync This matches the weed mount -filerProxy pattern exactly, factorizing the URL generation logic across all consumers. Co-Authored-By: Athena 🏛️ (custom / Qwen3.6-35B-A3B-UD-Q4_K_XL.gguf) * Remove accidental ROCmFPX submodule reference * gofmt chunk upload option block * Preserve broker cipher and re-read proxy filer per upload attempt Chunk uploads must keep the configured Cipher, and both the mount and broker current filer can change on failover, so build the proxy upload URL inside the closure instead of capturing the address once. --------- Co-authored-by: Chris Lu --- weed/command/filer_copy.go | 9 --- weed/filer/gateway_upload.go | 9 +-- weed/mount/weedfs_write.go | 11 ++-- weed/mq/broker/broker_write.go | 27 +++++---- weed/mq/logstore/log_to_parquet.go | 3 - weed/operation/upload_content.go | 32 +++++++++-- weed/operation/upload_content_test.go | 5 +- .../replication/sink/filersink/fetch_write.go | 57 ++++++++----------- weed/server/filer_server_handlers_read.go | 13 ++++- 9 files changed, 84 insertions(+), 82 deletions(-) diff --git a/weed/command/filer_copy.go b/weed/command/filer_copy.go index f4a712f8f..4b814116d 100644 --- a/weed/command/filer_copy.go +++ b/weed/command/filer_copy.go @@ -371,9 +371,6 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err MimeType: mimeType, PairMap: nil, }, - func(host, fileId string) string { - return fmt.Sprintf("http://%s/%s", host, fileId) - }, util.NewBytesReader(data), ) if flushErr != nil { @@ -456,9 +453,6 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File, MimeType: "", PairMap: nil, }, - func(host, fileId string) string { - return fmt.Sprintf("http://%s/%s", host, fileId) - }, io.NewSectionReader(f, i*chunkSize, chunkSize), ) @@ -576,9 +570,6 @@ func (worker *FileCopyWorker) saveDataAsChunk(reader io.Reader, name string, off MimeType: "", PairMap: nil, }, - func(host, fileId string) string { - return fmt.Sprintf("http://%s/%s", host, fileId) - }, reader, ) diff --git a/weed/filer/gateway_upload.go b/weed/filer/gateway_upload.go index 507b5efa6..7b6da4a25 100644 --- a/weed/filer/gateway_upload.go +++ b/weed/filer/gateway_upload.go @@ -17,7 +17,6 @@ type GatewayChunkUploader interface { filerClient filer_pb.FilerClient, assignRequest *filer_pb.AssignVolumeRequest, uploadOption *operation.UploadOption, - genFileUrlFn func(host, fileId string) string, reader io.Reader, ) (fileId string, uploadResult *operation.UploadResult, err error, data []byte) } @@ -120,11 +119,8 @@ func SaveGatewayDataAsChunk(req GatewayChunkUploadRequest) (*filer_pb.FileChunk, PairMap: req.PairMap, } - genFileUrlFn := func(host, fileId string) string { - if req.VolumeServerAccess == "filerProxy" && req.FilerHTTPAddress != "" { - return fmt.Sprintf("http://%s/?proxyChunkId=%s", req.FilerHTTPAddress, fileId) - } - return fmt.Sprintf("http://%s/%s", host, fileId) + if req.VolumeServerAccess == "filerProxy" && req.FilerHTTPAddress != "" { + uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(req.FilerHTTPAddress) } assignRequest := &filer_pb.AssignVolumeRequest{ @@ -141,7 +137,6 @@ func SaveGatewayDataAsChunk(req GatewayChunkUploadRequest) (*filer_pb.FileChunk, req.FilerClient, assignRequest, uploadOption, - genFileUrlFn, req.Reader, ) if uploadErr != nil { diff --git a/weed/mount/weedfs_write.go b/weed/mount/weedfs_write.go index dfe5a6e48..8bb6a4157 100644 --- a/weed/mount/weedfs_write.go +++ b/weed/mount/weedfs_write.go @@ -35,12 +35,11 @@ func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath) filer.SaveDataAsChunkFun PairMap: nil, WantMd5: true, } - genFileUrlFn := func(host, fileId string) string { - fileUrl := fmt.Sprintf("http://%s/%s", host, fileId) - if wfs.option.VolumeServerAccess == "filerProxy" { - fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.getCurrentFiler(), fileId) + if wfs.option.VolumeServerAccess == "filerProxy" { + // getCurrentFiler() can change on failover, so read it per attempt. + uploadOption.GenUploadUrl = func(host, fileId string) string { + return fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.getCurrentFiler(), fileId) } - return fileUrl } fileId, uploadResult, err, data := uploader.UploadWithRetry( @@ -54,7 +53,7 @@ func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath) filer.SaveDataAsChunkFun DataCenter: wfs.option.DataCenter, Path: assignPath, }, - uploadOption, genFileUrlFn, reader, + uploadOption, reader, ) if err != nil { diff --git a/weed/mq/broker/broker_write.go b/weed/mq/broker/broker_write.go index bdb72a770..4b91e49e0 100644 --- a/weed/mq/broker/broker_write.go +++ b/weed/mq/broker/broker_write.go @@ -161,27 +161,26 @@ func (b *MessageQueueBroker) assignAndUpload(targetFile string, data []byte) (fi return } + uploadOption := &operation.UploadOption{ + Cipher: b.option.Cipher, + } + if b.option.VolumeServerAccess == "filerProxy" { + // b.currentFiler can change on failover, so read it per attempt. + uploadOption.GenUploadUrl = func(host, fileId string) string { + return fmt.Sprintf("http://%s/?proxyChunkId=%s", b.currentFiler, fileId) + } + } + fileId, uploadResult, err, _ = uploader.UploadWithRetry( b, &filer_pb.AssignVolumeRequest{ Count: 1, Replication: b.option.DefaultReplication, Collection: "topics", - // TtlSec: wfs.option.TtlSec, - // DiskType: string(wfs.option.DiskType), - DataCenter: b.option.DataCenter, - Path: targetFile, - }, - &operation.UploadOption{ - Cipher: b.option.Cipher, - }, - func(host, fileId string) string { - fileUrl := fmt.Sprintf("http://%s/%s", host, fileId) - if b.option.VolumeServerAccess == "filerProxy" { - fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", b.currentFiler, fileId) - } - return fileUrl + DataCenter: b.option.DataCenter, + Path: targetFile, }, + uploadOption, reader, ) return diff --git a/weed/mq/logstore/log_to_parquet.go b/weed/mq/logstore/log_to_parquet.go index bfd5ff10e..067ddf1b0 100644 --- a/weed/mq/logstore/log_to_parquet.go +++ b/weed/mq/logstore/log_to_parquet.go @@ -479,9 +479,6 @@ func saveParquetFileToPartitionDir(filerClient filer_pb.FilerClient, sourceFile MimeType: "application/vnd.apache.parquet", PairMap: nil, }, - func(host, fileId string) string { - return fmt.Sprintf("http://%s/%s", host, fileId) - }, io.NewSectionReader(sourceFile, i*chunkSize, chunkSize), ) if err != nil { diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 493d1c2c9..946149ade 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -30,6 +30,21 @@ import ( util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client" ) +// GenUploadUrlProxy returns a function that builds a chunk upload URL via the +// filer proxy. Identical to the inline logic used by weed mount -filerProxy +// and weed filer gateway: +// +// - without filerProxy: "http://{host}/{fileId}" +// - with filerProxy: "http://{filerAddress}/?proxyChunkId={fileId}" +func GenUploadUrlProxy(filerAddress string) func(host, fileId string) string { + return func(host, fileId string) string { + if filerAddress == "" { + return fmt.Sprintf("http://%s/%s", host, fileId) + } + return fmt.Sprintf("http://%s/?proxyChunkId=%s", filerAddress, fileId) + } +} + type UploadOption struct { UploadUrl string Filename string @@ -43,8 +58,9 @@ type UploadOption struct { Md5 string WantMd5 bool // compute Content-MD5 from the data when Md5 is unset and the upload is not ciphered BytesBuffer *bytes.Buffer - SourceUrl string // optional: for logging when reading from a remote source - MaxAttempts int // <=0 uses the default + SourceUrl string // optional: for logging when reading from a remote source + MaxAttempts int // <=0 uses the default + GenUploadUrl func(host, fileId string) string // if nil → fallback "http://{host}/{fileId}" } type UploadResult struct { @@ -153,7 +169,7 @@ func NewUploaderWithHttpClient(httpClient HTTPClient) *Uploader { } } -func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, host string, auth security.EncodedJwt, err error), uploadOption *UploadOption, genFileUrlFn func(host, fileId string) string, data []byte) (fileId string, uploadResult *UploadResult, err error) { +func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, host string, auth security.EncodedJwt, err error), uploadOption *UploadOption, data []byte) (fileId string, uploadResult *UploadResult, err error) { doUploadFunc := func() error { var host string var auth security.EncodedJwt @@ -162,7 +178,11 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho return err } - uploadOption.UploadUrl = genFileUrlFn(host, fileId) + genUrl := uploadOption.GenUploadUrl + if genUrl == nil { + genUrl = func(host, fileId string) string { return fmt.Sprintf("http://%s/%s", host, fileId) } + } + uploadOption.UploadUrl = genUrl(host, fileId) uploadOption.Jwt = auth uploadResult, err = uploader.retriedUploadData(context.Background(), data, uploadOption) @@ -183,7 +203,7 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho // UploadWithRetry will retry both assigning volume request and uploading content // The option parameter does not need to specify UploadUrl and Jwt, which will come from assigning volume. -func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assignRequest *filer_pb.AssignVolumeRequest, uploadOption *UploadOption, genFileUrlFn func(host, fileId string) string, reader io.Reader) (fileId string, uploadResult *UploadResult, err error, data []byte) { +func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assignRequest *filer_pb.AssignVolumeRequest, uploadOption *UploadOption, reader io.Reader) (fileId string, uploadResult *UploadResult, err error, data []byte) { bytesReader, ok := reader.(*util.BytesReader) if ok { data = bytesReader.Bytes @@ -234,7 +254,7 @@ func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assi err = fmt.Errorf("filerGrpcAddress assign volume: %w", grpcAssignErr) } return - }, uploadOption, genFileUrlFn, data) + }, uploadOption, data) return } diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index bd1e10e50..2b91c0056 100644 --- a/weed/operation/upload_content_test.go +++ b/weed/operation/upload_content_test.go @@ -107,9 +107,7 @@ func TestUploadWithRetryDataReassignsOnVolumeSizeExceeded(t *testing.T) { return "1,first", "volume-a", "", nil } return "2,second", "volume-b", "", nil - }, &UploadOption{Filename: "test.bin"}, func(host, fileId string) string { - return "http://" + host + "/" + fileId - }, []byte("abc")) + }, &UploadOption{Filename: "test.bin"}, []byte("abc")) if err != nil { t.Fatalf("expected success after reassignment, got %v", err) @@ -210,7 +208,6 @@ func TestUploadWithRetryBoundsAssignVolume(t *testing.T) { _, _, err, _ := uploader.UploadWithRetry(client, &filer_pb.AssignVolumeRequest{Count: 1}, &UploadOption{Filename: "test.bin"}, - func(host, fileId string) string { return "http://" + host + "/" + fileId }, bytes.NewReader([]byte("abc")), ) done <- err diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 3f0742fbd..4bfa354d3 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -187,6 +187,17 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc retryName := fmt.Sprintf("replicate manifest chunk %s", sourceFileId) err = util.RetryUntil(retryName, func() error { + uploadOption := &operation.UploadOption{ + Filename: "", + Cipher: false, + IsInputCompressed: false, + MimeType: "application/octet-stream", + PairMap: nil, + RetryForever: false, + } + if fs.writeChunkByFiler { + uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(fs.address) + } currentFileId, uploadResult, uploadErr, _ := uploader.UploadWithRetry( fs, &filer_pb.AssignVolumeRequest{ @@ -198,17 +209,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc DiskType: fs.diskType, Path: path, }, - &operation.UploadOption{ - Filename: "", - Cipher: false, - IsInputCompressed: false, - MimeType: "application/octet-stream", - PairMap: nil, - RetryForever: false, - }, - func(host, fileId string) string { - return fs.buildUploadUrl(host, fileId) - }, + uploadOption, bytes.NewReader(manifestData), ) if uploadErr != nil { @@ -302,6 +303,18 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, transferStatus.Status = "uploading" transferStatus.mu.Unlock() + uploadOption := &operation.UploadOption{ + Filename: savedFilename, + Cipher: false, + IsInputCompressed: "gzip" == savedHeader.Get("Content-Encoding"), + MimeType: savedHeader.Get("Content-Type"), + PairMap: nil, + RetryForever: false, + SourceUrl: savedSourceUrl, + } + if fs.writeChunkByFiler { + uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(fs.address) + } currentFileId, uploadResult, uploadErr, _ := uploader.UploadWithRetry( fs, &filer_pb.AssignVolumeRequest{ @@ -313,20 +326,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, DiskType: fs.diskType, Path: path, }, - &operation.UploadOption{ - Filename: savedFilename, - Cipher: false, - IsInputCompressed: "gzip" == savedHeader.Get("Content-Encoding"), - MimeType: savedHeader.Get("Content-Type"), - PairMap: nil, - RetryForever: false, - SourceUrl: savedSourceUrl, - }, - func(host, fileId string) string { - fileUrl := fs.buildUploadUrl(host, fileId) - glog.V(4).Infof("replicating %s to %s header:%+v", savedFilename, fileUrl, savedHeader) - return fileUrl - }, + uploadOption, util.NewBytesReader(fullData), ) if uploadErr != nil { @@ -435,13 +435,6 @@ func validateReplicatedReadSize(sourceChunk *filer_pb.FileChunk, readSize int) e return nil } -func (fs *FilerSink) buildUploadUrl(host, fileId string) string { - if fs.writeChunkByFiler { - return fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId) - } - return fmt.Sprintf("http://%s/%s", host, fileId) -} - // hasSourceNewerVersion reports whether the source's current entry for targetPath // has moved past sourceMtimeNs — gone, or a strictly-newer mtime — meaning the // version being replayed is stale. The lookup runs regardless of sourceMtimeNs so diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go index 72fbd02be..62e65b0ef 100644 --- a/weed/server/filer_server_handlers_read.go +++ b/weed/server/filer_server_handlers_read.go @@ -254,5 +254,16 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) } func (fs *FilerServer) maybeGetVolumeReadJwtAuthorizationToken(fileId string) string { - return string(security.GenJwtForVolumeServer(fs.volumeGuard.ReadSigningKey(), fs.volumeGuard.ReadExpiresAfterSec(), fileId)) + // Generate a read JWT for volume server access. If the dedicated + // read key (jwt.signing.read.key) is not configured, fall back to the + // general signing key (jwt.signing.key) so the proxy can still + // authenticate to volume servers that require JWT. + key := fs.volumeGuard.ReadSigningKey() + if len(key) == 0 { + key = fs.volumeGuard.SigningKey() + } + if len(key) == 0 { + return "" + } + return string(security.GenJwtForVolumeServer(key, fs.volumeGuard.ReadExpiresAfterSec(), fileId)) }