Add TUS protocol support for resumable uploads (#7592)
* Add TUS protocol integration tests
This commit adds integration tests for the TUS (resumable upload) protocol
in preparation for implementing TUS support in the filer.
Test coverage includes:
- OPTIONS handler for capability discovery
- Basic single-request upload
- Chunked/resumable uploads
- HEAD requests for offset tracking
- DELETE for upload cancellation
- Error handling (invalid offsets, missing uploads)
- Creation-with-upload extension
- Resume after interruption simulation
Tests are skipped in short mode and require a running SeaweedFS cluster.
* Add TUS session storage types and utilities
Implements TUS upload session management:
- TusSession struct for tracking upload state
- Session creation with directory-based storage
- Session persistence using filer entries
- Session retrieval and offset updates
- Session deletion with chunk cleanup
- Upload completion with chunk assembly into final file
Session data is stored in /.uploads.tus/{upload-id}/ directory,
following the pattern used by S3 multipart uploads.
* Add TUS HTTP handlers
Implements TUS protocol HTTP handlers:
- tusHandler: Main entry point routing requests
- tusOptionsHandler: Capability discovery (OPTIONS)
- tusCreateHandler: Create new upload (POST)
- tusHeadHandler: Get upload offset (HEAD)
- tusPatchHandler: Upload data at offset (PATCH)
- tusDeleteHandler: Cancel upload (DELETE)
- tusWriteData: Upload data to volume servers
Features:
- Supports creation-with-upload extension
- Validates TUS protocol headers
- Offset conflict detection
- Automatic upload completion when size is reached
- Metadata parsing from Upload-Metadata header
* Wire up TUS protocol routes in filer server
Add TUS handler route (/.tus/) to the filer HTTP server.
The TUS route is registered before the catch-all route to ensure
proper routing of TUS protocol requests.
TUS protocol is now accessible at:
- OPTIONS /.tus/ - Capability discovery
- POST /.tus/{path} - Create upload
- HEAD /.tus/.uploads/{id} - Get offset
- PATCH /.tus/.uploads/{id} - Upload data
- DELETE /.tus/.uploads/{id} - Cancel upload
* Improve TUS integration test setup
Add comprehensive Makefile for TUS tests with targets:
- test-with-server: Run tests with automatic server management
- test-basic/chunked/resume/errors: Specific test categories
- manual-start/stop: For development testing
- debug-logs/status: For debugging
- ci-test: For CI/CD pipelines
Update README.md with:
- Detailed TUS protocol documentation
- All endpoint descriptions with headers
- Usage examples with curl commands
- Architecture diagram
- Comparison with S3 multipart uploads
Follows the pattern established by other tests in test/ folder.
* Fix TUS integration tests and creation-with-upload
- Fix test URLs to use full URLs instead of relative paths
- Fix creation-with-upload to refresh session before completing
- Fix Makefile to properly handle test cleanup
- Add FullURL helper function to TestCluster
* Add TUS protocol tests to GitHub Actions CI
- Add tus-tests.yml workflow that runs on PRs and pushes
- Runs when TUS-related files are modified
- Automatic server management for integration testing
- Upload logs on failure for debugging
* Make TUS base path configurable via CLI
- Add -tus.path CLI flag to filer command
- TUS is disabled by default (empty path)
- Example: -tus.path=/.tus to enable at /.tus endpoint
- Update test Makefile to use -tus.path flag
- Update README with TUS enabling instructions
* Rename -tus.path to -tusBasePath with default .tus
- Rename CLI flag from -tus.path to -tusBasePath
- Default to .tus (TUS enabled by default)
- Add -filer.tusBasePath option to weed server command
- Properly handle path prefix (prepend / if missing)
* Address code review comments
- Sort chunks by offset before assembling final file
- Use chunk.Offset directly instead of recalculating
- Return error on invalid file ID instead of skipping
- Require Content-Length header for PATCH requests
- Use fs.option.Cipher for encryption setting
- Detect MIME type from data using http.DetectContentType
- Fix concurrency group for push events in workflow
- Use os.Interrupt instead of Kill for graceful shutdown in tests
* fmt
* Address remaining code review comments
- Fix potential open redirect vulnerability by sanitizing uploadLocation path
- Add language specifier to README code block
- Handle os.Create errors in test setup
- Use waitForHTTPServer instead of time.Sleep for master/volume readiness
- Improve test reliability and debugging
* Address critical and high-priority review comments
- Add per-session locking to prevent race conditions in updateTusSessionOffset
- Stream data directly to volume server instead of buffering entire chunk
- Only buffer 512 bytes for MIME type detection, then stream remaining data
- Clean up session locks when session is deleted
* Fix race condition to work across multiple filer instances
- Store each chunk as a separate file entry instead of updating session JSON
- Chunk file names encode offset, size, and fileId for atomic storage
- getTusSession loads chunks from directory listing (atomic read)
- Eliminates read-modify-write race condition across multiple filers
- Remove in-memory mutex that only worked for single filer instance
* Address code review comments: fix variable shadowing, sniff size, and test stability
- Rename path variable to reqPath to avoid shadowing path package
- Make sniff buffer size respect contentLength (read at most contentLength bytes)
- Handle Content-Length < 0 in creation-with-upload (return error for chunked encoding)
- Fix test cluster: use temp directory for filer store, add startup delay
* Fix test stability: increase cluster stabilization delay to 5 seconds
The tests were intermittently failing because the volume server needed more
time to create volumes and register with the master. Increasing the delay
from 2 to 5 seconds fixes the flaky test behavior.
* Address PR review comments for TUS protocol support
- Fix strconv.Atoi error handling in test file (lines 386, 747)
- Fix lossy fileId encoding: use base64 instead of underscore replacement
- Add pagination support for ListDirectoryEntries in getTusSession
- Batch delete chunks instead of one-by-one in deleteTusSession
* Address additional PR review comments for TUS protocol
- Fix UploadAt timestamp: use entry.Crtime instead of time.Now()
- Remove redundant JSON content in chunk entry (metadata in filename)
- Refactor tusWriteData to stream in 4MB chunks to avoid OOM on large uploads
- Pass filer.Entry to parseTusChunkPath to preserve actual upload time
* Address more PR review comments for TUS protocol
- Normalize TUS path once in filer_server.go, store in option.TusPath
- Remove redundant path normalization from TUS handlers
- Remove goto statement in tusCreateHandler, simplify control flow
* Remove unnecessary mutexes in tusWriteData
The upload loop is sequential, so uploadErrLock and chunksLock are not needed.
* Rename updateTusSessionOffset to saveTusChunk
Remove unused newOffset parameter and rename function to better reflect its purpose.
* Improve TUS upload performance and add path validation
- Reuse operation.Uploader across sub-chunks for better connection reuse
- Guard against TusPath='/' to prevent hijacking all filer routes
* Address PR review comments for TUS protocol
- Fix critical chunk filename parsing: use strings.Cut instead of SplitN
to correctly handle base64-encoded fileIds that may contain underscores
- Rename tusPath to tusBasePath for naming consistency across codebase
- Add background garbage collection for expired TUS sessions (runs hourly)
- Improve error messages with %w wrapping for better debuggability
* Address additional TUS PR review comments
- Fix tusBasePath default to use leading slash (/.tus) for consistency
- Add chunk contiguity validation in completeTusUpload to detect gaps/overlaps
- Fix offset calculation to find maximum contiguous range from 0, not just last chunk
- Return 413 Request Entity Too Large instead of silently truncating content
- Document tusChunkSize rationale (4MB balances memory vs request overhead)
- Fix Makefile xargs portability by removing GNU-specific -r flag
- Add explicit -tusBasePath flag to integration test for robustness
- Fix README example to use /.uploads/tus path format
* Revert log_buffer changes (moved to separate PR)
* Minor style fixes from PR review
- Simplify tusBasePath flag description to use example format
- Add 'TUS upload' prefix to session not found error message
- Remove duplicate tusChunkSize comment
- Capitalize warning message for consistency
- Add grep filter to Makefile xargs for better empty input handling
This commit is contained in:
Chris Lu
@@ -74,6 +74,7 @@ type FilerOptions struct {
|
||||
diskType *string
|
||||
allowedOrigins *string
|
||||
exposeDirectoryData *bool
|
||||
tusBasePath *string
|
||||
certProvider certprovider.Provider
|
||||
}
|
||||
|
||||
@@ -109,6 +110,7 @@ func init() {
|
||||
f.diskType = cmdFiler.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
|
||||
f.allowedOrigins = cmdFiler.Flag.String("allowedOrigins", "*", "comma separated list of allowed origins")
|
||||
f.exposeDirectoryData = cmdFiler.Flag.Bool("exposeDirectoryData", true, "whether to return directory metadata and content in Filer UI")
|
||||
f.tusBasePath = cmdFiler.Flag.String("tusBasePath", "/.tus", "TUS resumable upload endpoint base path (e.g., /.tus)")
|
||||
|
||||
// start s3 on filer
|
||||
filerStartS3 = cmdFiler.Flag.Bool("s3", false, "whether to start S3 gateway")
|
||||
@@ -342,6 +344,7 @@ func (fo *FilerOptions) startFiler() {
|
||||
DownloadMaxBytesPs: int64(*fo.downloadMaxMBps) * 1024 * 1024,
|
||||
DiskType: *fo.diskType,
|
||||
AllowedOrigins: strings.Split(*fo.allowedOrigins, ","),
|
||||
TusBasePath: *fo.tusBasePath,
|
||||
})
|
||||
if nfs_err != nil {
|
||||
glog.Fatalf("Filer startup error: %v", nfs_err)
|
||||
|
||||
@@ -129,6 +129,7 @@ func init() {
|
||||
filerOptions.downloadMaxMBps = cmdServer.Flag.Int("filer.downloadMaxMBps", 0, "download max speed for each download request, in MB per second")
|
||||
filerOptions.diskType = cmdServer.Flag.String("filer.disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
|
||||
filerOptions.exposeDirectoryData = cmdServer.Flag.Bool("filer.exposeDirectoryData", true, "expose directory data via filer. If false, filer UI will be innaccessible.")
|
||||
filerOptions.tusBasePath = cmdServer.Flag.String("filer.tusBasePath", "/.tus", "TUS resumable upload endpoint base path (e.g., /.tus)")
|
||||
|
||||
serverOptions.v.port = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port")
|
||||
serverOptions.v.portGrpc = cmdServer.Flag.Int("volume.port.grpc", 0, "volume server grpc listen port")
|
||||
|
||||
@@ -167,6 +167,14 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
|
||||
// Copy extended attributes from source, filtering out conflicting encryption metadata
|
||||
// Pre-compute encryption state once for efficiency
|
||||
srcHasSSEC := IsSSECEncrypted(entry.Extended)
|
||||
srcHasSSEKMS := IsSSEKMSEncrypted(entry.Extended)
|
||||
srcHasSSES3 := IsSSES3EncryptedInternal(entry.Extended)
|
||||
dstWantsSSEC := IsSSECRequest(r)
|
||||
dstWantsSSEKMS := IsSSEKMSRequest(r)
|
||||
dstWantsSSES3 := IsSSES3RequestInternal(r)
|
||||
|
||||
for k, v := range entry.Extended {
|
||||
// Skip encryption-specific headers that might conflict with destination encryption type
|
||||
skipHeader := false
|
||||
@@ -177,17 +185,9 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
skipHeader = true
|
||||
}
|
||||
|
||||
// If we're doing cross-encryption, skip conflicting headers
|
||||
if !skipHeader && len(entry.GetChunks()) > 0 {
|
||||
// Detect source and destination encryption types
|
||||
srcHasSSEC := IsSSECEncrypted(entry.Extended)
|
||||
srcHasSSEKMS := IsSSEKMSEncrypted(entry.Extended)
|
||||
srcHasSSES3 := IsSSES3EncryptedInternal(entry.Extended)
|
||||
dstWantsSSEC := IsSSECRequest(r)
|
||||
dstWantsSSEKMS := IsSSEKMSRequest(r)
|
||||
dstWantsSSES3 := IsSSES3RequestInternal(r)
|
||||
|
||||
// Use helper function to determine if header should be skipped
|
||||
// Filter conflicting headers for cross-encryption or encrypted→unencrypted copies
|
||||
// This applies to both inline files (no chunks) and chunked files - fixes GitHub #7562
|
||||
if !skipHeader {
|
||||
skipHeader = shouldSkipEncryptionHeader(k,
|
||||
srcHasSSEC, srcHasSSEKMS, srcHasSSES3,
|
||||
dstWantsSSEC, dstWantsSSEKMS, dstWantsSSES3)
|
||||
@@ -212,10 +212,31 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
dstEntry.Extended[k] = v
|
||||
}
|
||||
|
||||
// For zero-size files or files without chunks, use the original approach
|
||||
// For zero-size files or files without chunks, handle inline content
|
||||
// This includes encrypted inline files that need decryption/re-encryption
|
||||
if entry.Attributes.FileSize == 0 || len(entry.GetChunks()) == 0 {
|
||||
// Just copy the entry structure without chunks for zero-size files
|
||||
dstEntry.Chunks = nil
|
||||
|
||||
// Handle inline encrypted content - fixes GitHub #7562
|
||||
if len(entry.Content) > 0 {
|
||||
inlineContent, inlineMetadata, inlineErr := s3a.processInlineContentForCopy(
|
||||
entry, r, dstBucket, dstObject,
|
||||
srcHasSSEC, srcHasSSEKMS, srcHasSSES3,
|
||||
dstWantsSSEC, dstWantsSSEKMS, dstWantsSSES3)
|
||||
if inlineErr != nil {
|
||||
glog.Errorf("CopyObjectHandler inline content error: %v", inlineErr)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
dstEntry.Content = inlineContent
|
||||
|
||||
// Apply inline destination metadata
|
||||
if inlineMetadata != nil {
|
||||
for k, v := range inlineMetadata {
|
||||
dstEntry.Extended[k] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Use unified copy strategy approach
|
||||
dstChunks, dstMetadata, copyErr := s3a.executeUnifiedCopyStrategy(entry, r, dstBucket, srcObject, dstObject)
|
||||
@@ -2508,3 +2529,233 @@ func shouldSkipEncryptionHeader(headerKey string,
|
||||
// Default: don't skip the header
|
||||
return false
|
||||
}
|
||||
|
||||
// processInlineContentForCopy handles encryption/decryption for inline content during copy
|
||||
// This fixes GitHub #7562 where small files stored inline weren't properly decrypted/re-encrypted
|
||||
func (s3a *S3ApiServer) processInlineContentForCopy(
|
||||
entry *filer_pb.Entry, r *http.Request, dstBucket, dstObject string,
|
||||
srcSSEC, srcSSEKMS, srcSSES3 bool,
|
||||
dstSSEC, dstSSEKMS, dstSSES3 bool) ([]byte, map[string][]byte, error) {
|
||||
|
||||
content := entry.Content
|
||||
var dstMetadata map[string][]byte
|
||||
|
||||
// Check if source is encrypted and needs decryption
|
||||
srcEncrypted := srcSSEC || srcSSEKMS || srcSSES3
|
||||
|
||||
// Check if destination needs encryption (explicit request or bucket default)
|
||||
dstNeedsEncryption := dstSSEC || dstSSEKMS || dstSSES3
|
||||
if !dstNeedsEncryption {
|
||||
// Check bucket default encryption
|
||||
bucketMetadata, err := s3a.getBucketMetadata(dstBucket)
|
||||
if err == nil && bucketMetadata != nil && bucketMetadata.Encryption != nil {
|
||||
switch bucketMetadata.Encryption.SseAlgorithm {
|
||||
case "aws:kms":
|
||||
dstSSEKMS = true
|
||||
dstNeedsEncryption = true
|
||||
case "AES256":
|
||||
dstSSES3 = true
|
||||
dstNeedsEncryption = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Decrypt source content if encrypted
|
||||
if srcEncrypted {
|
||||
decryptedContent, decErr := s3a.decryptInlineContent(entry, srcSSEC, srcSSEKMS, srcSSES3, r)
|
||||
if decErr != nil {
|
||||
return nil, nil, fmt.Errorf("failed to decrypt inline content: %w", decErr)
|
||||
}
|
||||
content = decryptedContent
|
||||
glog.V(3).Infof("Decrypted inline content: %d bytes", len(content))
|
||||
}
|
||||
|
||||
// Re-encrypt if destination needs encryption
|
||||
if dstNeedsEncryption {
|
||||
encryptedContent, encMetadata, encErr := s3a.encryptInlineContent(content, dstBucket, dstObject, dstSSEC, dstSSEKMS, dstSSES3, r)
|
||||
if encErr != nil {
|
||||
return nil, nil, fmt.Errorf("failed to encrypt inline content: %w", encErr)
|
||||
}
|
||||
content = encryptedContent
|
||||
dstMetadata = encMetadata
|
||||
glog.V(3).Infof("Encrypted inline content: %d bytes", len(content))
|
||||
}
|
||||
|
||||
return content, dstMetadata, nil
|
||||
}
|
||||
|
||||
// decryptInlineContent decrypts inline content from an encrypted source
|
||||
func (s3a *S3ApiServer) decryptInlineContent(entry *filer_pb.Entry, srcSSEC, srcSSEKMS, srcSSES3 bool, r *http.Request) ([]byte, error) {
|
||||
content := entry.Content
|
||||
|
||||
if srcSSES3 {
|
||||
// Get SSE-S3 key from metadata
|
||||
keyData, exists := entry.Extended[s3_constants.SeaweedFSSSES3Key]
|
||||
if !exists {
|
||||
return nil, fmt.Errorf("SSE-S3 key not found in metadata")
|
||||
}
|
||||
|
||||
keyManager := GetSSES3KeyManager()
|
||||
sseKey, err := DeserializeSSES3Metadata(keyData, keyManager)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to deserialize SSE-S3 key: %w", err)
|
||||
}
|
||||
|
||||
// Get IV
|
||||
iv := sseKey.IV
|
||||
if len(iv) == 0 {
|
||||
return nil, fmt.Errorf("SSE-S3 IV not found")
|
||||
}
|
||||
|
||||
// Decrypt content
|
||||
decryptedReader, err := CreateSSES3DecryptedReader(bytes.NewReader(content), sseKey, iv)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create SSE-S3 decrypted reader: %w", err)
|
||||
}
|
||||
return io.ReadAll(decryptedReader)
|
||||
|
||||
} else if srcSSEKMS {
|
||||
// Get SSE-KMS key from metadata
|
||||
keyData, exists := entry.Extended[s3_constants.SeaweedFSSSEKMSKey]
|
||||
if !exists {
|
||||
return nil, fmt.Errorf("SSE-KMS key not found in metadata")
|
||||
}
|
||||
|
||||
sseKey, err := DeserializeSSEKMSMetadata(keyData)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to deserialize SSE-KMS key: %w", err)
|
||||
}
|
||||
|
||||
// Decrypt content
|
||||
decryptedReader, err := CreateSSEKMSDecryptedReader(bytes.NewReader(content), sseKey)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create SSE-KMS decrypted reader: %w", err)
|
||||
}
|
||||
return io.ReadAll(decryptedReader)
|
||||
|
||||
} else if srcSSEC {
|
||||
// Get SSE-C key from request headers
|
||||
sourceKey, err := ParseSSECCopySourceHeaders(r)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse SSE-C copy source headers: %w", err)
|
||||
}
|
||||
|
||||
// Get IV from metadata
|
||||
iv, err := GetSSECIVFromMetadata(entry.Extended)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get SSE-C IV: %w", err)
|
||||
}
|
||||
|
||||
// Decrypt content
|
||||
decryptedReader, err := CreateSSECDecryptedReader(bytes.NewReader(content), sourceKey, iv)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create SSE-C decrypted reader: %w", err)
|
||||
}
|
||||
return io.ReadAll(decryptedReader)
|
||||
}
|
||||
|
||||
// Source not encrypted, return as-is
|
||||
return content, nil
|
||||
}
|
||||
|
||||
// encryptInlineContent encrypts inline content for the destination
|
||||
func (s3a *S3ApiServer) encryptInlineContent(content []byte, dstBucket, dstObject string,
|
||||
dstSSEC, dstSSEKMS, dstSSES3 bool, r *http.Request) ([]byte, map[string][]byte, error) {
|
||||
|
||||
dstMetadata := make(map[string][]byte)
|
||||
|
||||
if dstSSES3 {
|
||||
// Generate SSE-S3 key
|
||||
keyManager := GetSSES3KeyManager()
|
||||
key, err := keyManager.GetOrCreateKey("")
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to generate SSE-S3 key: %w", err)
|
||||
}
|
||||
|
||||
// Encrypt content
|
||||
encryptedReader, iv, err := CreateSSES3EncryptedReader(bytes.NewReader(content), key)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to create SSE-S3 encrypted reader: %w", err)
|
||||
}
|
||||
|
||||
encryptedContent, err := io.ReadAll(encryptedReader)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to read encrypted content: %w", err)
|
||||
}
|
||||
|
||||
// Store IV on key and serialize metadata
|
||||
key.IV = iv
|
||||
keyData, err := SerializeSSES3Metadata(key)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to serialize SSE-S3 metadata: %w", err)
|
||||
}
|
||||
|
||||
dstMetadata[s3_constants.SeaweedFSSSES3Key] = keyData
|
||||
dstMetadata[s3_constants.AmzServerSideEncryption] = []byte("AES256")
|
||||
|
||||
return encryptedContent, dstMetadata, nil
|
||||
|
||||
} else if dstSSEKMS {
|
||||
// Parse SSE-KMS headers
|
||||
keyID, encryptionContext, bucketKeyEnabled, err := ParseSSEKMSCopyHeaders(r)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to parse SSE-KMS headers: %w", err)
|
||||
}
|
||||
|
||||
// Build encryption context if needed
|
||||
if encryptionContext == nil {
|
||||
encryptionContext = BuildEncryptionContext(dstBucket, dstObject, bucketKeyEnabled)
|
||||
}
|
||||
|
||||
// Encrypt content
|
||||
encryptedReader, sseKey, err := CreateSSEKMSEncryptedReaderWithBucketKey(
|
||||
bytes.NewReader(content), keyID, encryptionContext, bucketKeyEnabled)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to create SSE-KMS encrypted reader: %w", err)
|
||||
}
|
||||
|
||||
encryptedContent, err := io.ReadAll(encryptedReader)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to read encrypted content: %w", err)
|
||||
}
|
||||
|
||||
// Serialize metadata
|
||||
keyData, err := SerializeSSEKMSMetadata(sseKey)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to serialize SSE-KMS metadata: %w", err)
|
||||
}
|
||||
|
||||
dstMetadata[s3_constants.SeaweedFSSSEKMSKey] = keyData
|
||||
dstMetadata[s3_constants.AmzServerSideEncryption] = []byte("aws:kms")
|
||||
|
||||
return encryptedContent, dstMetadata, nil
|
||||
|
||||
} else if dstSSEC {
|
||||
// Parse SSE-C headers
|
||||
destKey, err := ParseSSECHeaders(r)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to parse SSE-C headers: %w", err)
|
||||
}
|
||||
|
||||
// Encrypt content
|
||||
encryptedReader, iv, err := CreateSSECEncryptedReader(bytes.NewReader(content), destKey)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to create SSE-C encrypted reader: %w", err)
|
||||
}
|
||||
|
||||
encryptedContent, err := io.ReadAll(encryptedReader)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to read encrypted content: %w", err)
|
||||
}
|
||||
|
||||
// Store IV in metadata
|
||||
StoreSSECIVInMetadata(dstMetadata, iv)
|
||||
dstMetadata[s3_constants.AmzServerSideEncryptionCustomerAlgorithm] = []byte("AES256")
|
||||
dstMetadata[s3_constants.AmzServerSideEncryptionCustomerKeyMD5] = []byte(destKey.KeyMD5)
|
||||
|
||||
return encryptedContent, dstMetadata, nil
|
||||
}
|
||||
|
||||
// No encryption needed
|
||||
return content, nil, nil
|
||||
}
|
||||
|
||||
@@ -79,6 +79,7 @@ type FilerOption struct {
|
||||
DiskType string
|
||||
AllowedOrigins []string
|
||||
ExposeDirectoryData bool
|
||||
TusBasePath string
|
||||
}
|
||||
|
||||
type FilerServer struct {
|
||||
@@ -198,6 +199,24 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
|
||||
handleStaticResources(defaultMux)
|
||||
if !option.DisableHttp {
|
||||
defaultMux.HandleFunc("/healthz", requestIDMiddleware(fs.filerHealthzHandler))
|
||||
// TUS resumable upload protocol handler
|
||||
if option.TusBasePath != "" {
|
||||
// Normalize TusPath to always have a leading slash and no trailing slash
|
||||
if !strings.HasPrefix(option.TusBasePath, "/") {
|
||||
option.TusBasePath = "/" + option.TusBasePath
|
||||
}
|
||||
option.TusBasePath = strings.TrimRight(option.TusBasePath, "/")
|
||||
|
||||
// Disallow using "/" as TUS base to avoid hijacking all filer routes
|
||||
if option.TusBasePath == "" {
|
||||
glog.Warningf("Invalid TUS base path; TUS disabled (must not be root '/')")
|
||||
} else {
|
||||
handlePath := option.TusBasePath + "/"
|
||||
defaultMux.HandleFunc(handlePath, fs.filerGuard.WhiteList(requestIDMiddleware(fs.tusHandler)))
|
||||
// Start background cleanup of expired TUS sessions (every hour)
|
||||
fs.StartTusSessionCleanup(1 * time.Hour)
|
||||
}
|
||||
}
|
||||
defaultMux.HandleFunc("/", fs.filerGuard.WhiteList(requestIDMiddleware(fs.filerHandler)))
|
||||
}
|
||||
if defaultMux != readonlyMux {
|
||||
|
||||
461
weed/server/filer_server_tus_handlers.go
Normal file
461
weed/server/filer_server_tus_handlers.go
Normal file
@@ -0,0 +1,461 @@
|
||||
package weed_server
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"path"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/operation"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/stats"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
// tusHandler is the main entry point for TUS protocol requests
|
||||
func (fs *FilerServer) tusHandler(w http.ResponseWriter, r *http.Request) {
|
||||
// Set common TUS response headers
|
||||
w.Header().Set("Tus-Resumable", TusVersion)
|
||||
|
||||
// Check Tus-Resumable header for non-OPTIONS requests
|
||||
if r.Method != http.MethodOptions {
|
||||
tusVersion := r.Header.Get("Tus-Resumable")
|
||||
if tusVersion != TusVersion {
|
||||
http.Error(w, "Unsupported TUS version", http.StatusPreconditionFailed)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Route based on method and path
|
||||
reqPath := r.URL.Path
|
||||
// TusBasePath is pre-normalized in filer_server.go (leading slash, no trailing slash)
|
||||
tusPrefix := fs.option.TusBasePath
|
||||
|
||||
// Check if this is an upload location (contains upload ID after {tusPrefix}/.uploads/)
|
||||
uploadsPrefix := tusPrefix + "/.uploads/"
|
||||
if strings.HasPrefix(reqPath, uploadsPrefix) {
|
||||
uploadID := strings.TrimPrefix(reqPath, uploadsPrefix)
|
||||
uploadID = strings.Split(uploadID, "/")[0] // Get just the ID, not any trailing path
|
||||
|
||||
switch r.Method {
|
||||
case http.MethodHead:
|
||||
fs.tusHeadHandler(w, r, uploadID)
|
||||
case http.MethodPatch:
|
||||
fs.tusPatchHandler(w, r, uploadID)
|
||||
case http.MethodDelete:
|
||||
fs.tusDeleteHandler(w, r, uploadID)
|
||||
default:
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Handle creation endpoints (POST to /.tus/{path})
|
||||
switch r.Method {
|
||||
case http.MethodOptions:
|
||||
fs.tusOptionsHandler(w, r)
|
||||
case http.MethodPost:
|
||||
fs.tusCreateHandler(w, r)
|
||||
default:
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
}
|
||||
}
|
||||
|
||||
// tusOptionsHandler handles OPTIONS requests for capability discovery
|
||||
func (fs *FilerServer) tusOptionsHandler(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Tus-Version", TusVersion)
|
||||
w.Header().Set("Tus-Extension", TusExtensions)
|
||||
w.Header().Set("Tus-Max-Size", strconv.FormatInt(TusMaxSize, 10))
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
// tusCreateHandler handles POST requests to create new uploads
|
||||
func (fs *FilerServer) tusCreateHandler(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := r.Context()
|
||||
|
||||
// Parse Upload-Length header (required)
|
||||
uploadLengthStr := r.Header.Get("Upload-Length")
|
||||
if uploadLengthStr == "" {
|
||||
http.Error(w, "Upload-Length header required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
uploadLength, err := strconv.ParseInt(uploadLengthStr, 10, 64)
|
||||
if err != nil || uploadLength < 0 {
|
||||
http.Error(w, "Invalid Upload-Length", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if uploadLength > TusMaxSize {
|
||||
http.Error(w, "Upload-Length exceeds maximum", http.StatusRequestEntityTooLarge)
|
||||
return
|
||||
}
|
||||
|
||||
// Parse Upload-Metadata header (optional)
|
||||
metadata := parseTusMetadata(r.Header.Get("Upload-Metadata"))
|
||||
|
||||
// TusBasePath is pre-normalized in filer_server.go (leading slash, no trailing slash)
|
||||
tusPrefix := fs.option.TusBasePath
|
||||
|
||||
// Determine target path from request URL
|
||||
targetPath := strings.TrimPrefix(r.URL.Path, tusPrefix)
|
||||
if targetPath == "" || targetPath == "/" {
|
||||
http.Error(w, "Target path required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Generate upload ID
|
||||
uploadID := uuid.New().String()
|
||||
|
||||
// Create upload session
|
||||
session, err := fs.createTusSession(ctx, uploadID, targetPath, uploadLength, metadata)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to create TUS session: %v", err)
|
||||
http.Error(w, "Failed to create upload", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Build upload location URL (ensure it starts with single /)
|
||||
uploadLocation := path.Clean(fmt.Sprintf("%s/.uploads/%s", tusPrefix, uploadID))
|
||||
if !strings.HasPrefix(uploadLocation, "/") {
|
||||
uploadLocation = "/" + uploadLocation
|
||||
}
|
||||
|
||||
// Handle creation-with-upload extension
|
||||
// TUS requires Content-Length for uploads; reject chunked encoding
|
||||
if r.Header.Get("Content-Type") == "application/offset+octet-stream" {
|
||||
if r.ContentLength < 0 {
|
||||
fs.deleteTusSession(ctx, uploadID)
|
||||
http.Error(w, "Content-Length header required for creation-with-upload", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if r.ContentLength > 0 {
|
||||
// Upload data in the creation request
|
||||
bytesWritten, uploadErr := fs.tusWriteData(ctx, session, 0, r.Body, r.ContentLength)
|
||||
if uploadErr != nil {
|
||||
// Cleanup session on failure
|
||||
fs.deleteTusSession(ctx, uploadID)
|
||||
if errors.Is(uploadErr, ErrContentTooLarge) {
|
||||
http.Error(w, "Content-Length exceeds declared upload size", http.StatusRequestEntityTooLarge)
|
||||
return
|
||||
}
|
||||
glog.Errorf("Failed to write initial TUS data: %v", uploadErr)
|
||||
http.Error(w, "Failed to write data", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Update offset in response header
|
||||
w.Header().Set("Upload-Offset", strconv.FormatInt(bytesWritten, 10))
|
||||
|
||||
// Check if upload is complete
|
||||
if bytesWritten == session.Size {
|
||||
// Refresh session to get updated chunks
|
||||
session, err = fs.getTusSession(ctx, uploadID)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get updated TUS session: %v", err)
|
||||
http.Error(w, "Failed to complete upload", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if err := fs.completeTusUpload(ctx, session); err != nil {
|
||||
glog.Errorf("Failed to complete TUS upload: %v", err)
|
||||
http.Error(w, "Failed to complete upload", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
// ContentLength == 0 is allowed, just proceed to respond
|
||||
}
|
||||
|
||||
w.Header().Set("Location", uploadLocation)
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
}
|
||||
|
||||
// tusHeadHandler handles HEAD requests to get current upload offset
|
||||
func (fs *FilerServer) tusHeadHandler(w http.ResponseWriter, r *http.Request, uploadID string) {
|
||||
ctx := r.Context()
|
||||
|
||||
session, err := fs.getTusSession(ctx, uploadID)
|
||||
if err != nil {
|
||||
http.Error(w, "Upload not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Upload-Offset", strconv.FormatInt(session.Offset, 10))
|
||||
w.Header().Set("Upload-Length", strconv.FormatInt(session.Size, 10))
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
// tusPatchHandler handles PATCH requests to upload data
|
||||
func (fs *FilerServer) tusPatchHandler(w http.ResponseWriter, r *http.Request, uploadID string) {
|
||||
ctx := r.Context()
|
||||
|
||||
// Validate Content-Type
|
||||
contentType := r.Header.Get("Content-Type")
|
||||
if contentType != "application/offset+octet-stream" {
|
||||
http.Error(w, "Content-Type must be application/offset+octet-stream", http.StatusUnsupportedMediaType)
|
||||
return
|
||||
}
|
||||
|
||||
// Get current session
|
||||
session, err := fs.getTusSession(ctx, uploadID)
|
||||
if err != nil {
|
||||
http.Error(w, "Upload not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate Upload-Offset header
|
||||
uploadOffsetStr := r.Header.Get("Upload-Offset")
|
||||
if uploadOffsetStr == "" {
|
||||
http.Error(w, "Upload-Offset header required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
uploadOffset, err := strconv.ParseInt(uploadOffsetStr, 10, 64)
|
||||
if err != nil || uploadOffset < 0 {
|
||||
http.Error(w, "Invalid Upload-Offset", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Check offset matches current position
|
||||
if uploadOffset != session.Offset {
|
||||
http.Error(w, fmt.Sprintf("Offset mismatch: expected %d, got %d", session.Offset, uploadOffset), http.StatusConflict)
|
||||
return
|
||||
}
|
||||
|
||||
// TUS requires Content-Length header for PATCH requests
|
||||
if r.ContentLength < 0 {
|
||||
http.Error(w, "Content-Length header required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Write data
|
||||
bytesWritten, err := fs.tusWriteData(ctx, session, uploadOffset, r.Body, r.ContentLength)
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrContentTooLarge) {
|
||||
http.Error(w, "Content-Length exceeds remaining upload size", http.StatusRequestEntityTooLarge)
|
||||
return
|
||||
}
|
||||
glog.Errorf("Failed to write TUS data: %v", err)
|
||||
http.Error(w, "Failed to write data", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
newOffset := uploadOffset + bytesWritten
|
||||
|
||||
// Check if upload is complete
|
||||
if newOffset == session.Size {
|
||||
// Refresh session to get updated chunks
|
||||
session, err = fs.getTusSession(ctx, uploadID)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to get updated TUS session: %v", err)
|
||||
http.Error(w, "Failed to complete upload", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
if err := fs.completeTusUpload(ctx, session); err != nil {
|
||||
glog.Errorf("Failed to complete TUS upload: %v", err)
|
||||
http.Error(w, "Failed to complete upload", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
w.Header().Set("Upload-Offset", strconv.FormatInt(newOffset, 10))
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// tusDeleteHandler handles DELETE requests to cancel uploads
|
||||
func (fs *FilerServer) tusDeleteHandler(w http.ResponseWriter, r *http.Request, uploadID string) {
|
||||
ctx := r.Context()
|
||||
|
||||
if err := fs.deleteTusSession(ctx, uploadID); err != nil {
|
||||
glog.Errorf("Failed to delete TUS session: %v", err)
|
||||
http.Error(w, "Failed to delete upload", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// tusChunkSize is the size of sub-chunks used when streaming uploads to volume servers.
|
||||
// 4MB balances memory usage (avoiding buffering large TUS chunks) with upload efficiency
|
||||
// (minimizing the number of volume server requests). Smaller values reduce memory but
|
||||
// increase request overhead; larger values do the opposite.
|
||||
const tusChunkSize = 4 * 1024 * 1024 // 4MB
|
||||
|
||||
// ErrContentTooLarge is returned when Content-Length exceeds remaining upload space
|
||||
var ErrContentTooLarge = fmt.Errorf("content length exceeds remaining upload size")
|
||||
|
||||
// tusWriteData uploads data to volume servers in streaming chunks and updates session
|
||||
// It reads data in fixed-size sub-chunks to avoid buffering large TUS chunks entirely in memory
|
||||
func (fs *FilerServer) tusWriteData(ctx context.Context, session *TusSession, offset int64, reader io.Reader, contentLength int64) (int64, error) {
|
||||
if contentLength == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Check if content length exceeds remaining size - return error instead of silently truncating
|
||||
remaining := session.Size - offset
|
||||
if contentLength > remaining {
|
||||
return 0, ErrContentTooLarge
|
||||
}
|
||||
if remaining <= 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Determine storage options based on target path
|
||||
so, err := fs.detectStorageOption0(ctx, session.TargetPath, "", "", "", "", "", "", "", "", "")
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("detect storage option: %w", err)
|
||||
}
|
||||
|
||||
// Read first bytes for MIME type detection
|
||||
sniffSize := int64(512)
|
||||
if contentLength < sniffSize {
|
||||
sniffSize = contentLength
|
||||
}
|
||||
sniffBuf := make([]byte, sniffSize)
|
||||
sniffN, sniffErr := io.ReadFull(reader, sniffBuf)
|
||||
if sniffErr != nil && sniffErr != io.EOF && sniffErr != io.ErrUnexpectedEOF {
|
||||
return 0, fmt.Errorf("read data for mime detection: %w", sniffErr)
|
||||
}
|
||||
if sniffN == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
sniffBuf = sniffBuf[:sniffN]
|
||||
mimeType := http.DetectContentType(sniffBuf)
|
||||
|
||||
// Create a combined reader with sniffed bytes prepended
|
||||
var dataReader io.Reader
|
||||
if int64(sniffN) >= contentLength {
|
||||
dataReader = bytes.NewReader(sniffBuf)
|
||||
} else {
|
||||
dataReader = io.MultiReader(bytes.NewReader(sniffBuf), io.LimitReader(reader, contentLength-int64(sniffN)))
|
||||
}
|
||||
|
||||
// Upload in streaming chunks to avoid buffering entire content in memory
|
||||
var totalWritten int64
|
||||
var uploadErr error
|
||||
var uploadedChunks []*TusChunkInfo
|
||||
|
||||
// Create one uploader for all sub-chunks to reuse HTTP client connections
|
||||
uploader, uploaderErr := operation.NewUploader()
|
||||
if uploaderErr != nil {
|
||||
return 0, fmt.Errorf("create uploader: %w", uploaderErr)
|
||||
}
|
||||
|
||||
chunkBuf := make([]byte, tusChunkSize)
|
||||
currentOffset := offset
|
||||
|
||||
for totalWritten < contentLength {
|
||||
// Read up to tusChunkSize bytes
|
||||
readSize := int64(tusChunkSize)
|
||||
if contentLength-totalWritten < readSize {
|
||||
readSize = contentLength - totalWritten
|
||||
}
|
||||
|
||||
n, readErr := io.ReadFull(dataReader, chunkBuf[:readSize])
|
||||
if readErr != nil && readErr != io.EOF && readErr != io.ErrUnexpectedEOF {
|
||||
uploadErr = fmt.Errorf("read chunk data: %w", readErr)
|
||||
break
|
||||
}
|
||||
if n == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
chunkData := chunkBuf[:n]
|
||||
|
||||
// Assign file ID from master for this sub-chunk
|
||||
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(ctx, so)
|
||||
if assignErr != nil {
|
||||
uploadErr = fmt.Errorf("assign volume: %w", assignErr)
|
||||
break
|
||||
}
|
||||
|
||||
// Upload to volume server using BytesReader (avoids double buffering in uploader)
|
||||
uploadResult, uploadResultErr, _ := uploader.Upload(ctx, util.NewBytesReader(chunkData), &operation.UploadOption{
|
||||
UploadUrl: urlLocation,
|
||||
Filename: "",
|
||||
Cipher: fs.option.Cipher,
|
||||
IsInputCompressed: false,
|
||||
MimeType: mimeType,
|
||||
PairMap: nil,
|
||||
Jwt: auth,
|
||||
})
|
||||
if uploadResultErr != nil {
|
||||
uploadErr = fmt.Errorf("upload data: %w", uploadResultErr)
|
||||
break
|
||||
}
|
||||
|
||||
// Create chunk info and save it
|
||||
chunk := &TusChunkInfo{
|
||||
Offset: currentOffset,
|
||||
Size: int64(uploadResult.Size),
|
||||
FileId: fileId,
|
||||
UploadAt: time.Now().UnixNano(),
|
||||
}
|
||||
|
||||
if saveErr := fs.saveTusChunk(ctx, session.ID, chunk); saveErr != nil {
|
||||
// Cleanup this chunk on failure
|
||||
fs.filer.DeleteChunks(ctx, util.FullPath(session.TargetPath), []*filer_pb.FileChunk{
|
||||
{FileId: fileId},
|
||||
})
|
||||
uploadErr = fmt.Errorf("update session: %w", saveErr)
|
||||
break
|
||||
}
|
||||
|
||||
uploadedChunks = append(uploadedChunks, chunk)
|
||||
|
||||
totalWritten += int64(uploadResult.Size)
|
||||
currentOffset += int64(uploadResult.Size)
|
||||
stats.FilerHandlerCounter.WithLabelValues("tusUploadChunk").Inc()
|
||||
}
|
||||
|
||||
if uploadErr != nil {
|
||||
// Cleanup all uploaded chunks on error
|
||||
if len(uploadedChunks) > 0 {
|
||||
var chunksToDelete []*filer_pb.FileChunk
|
||||
for _, c := range uploadedChunks {
|
||||
chunksToDelete = append(chunksToDelete, &filer_pb.FileChunk{FileId: c.FileId})
|
||||
}
|
||||
fs.filer.DeleteChunks(ctx, util.FullPath(session.TargetPath), chunksToDelete)
|
||||
}
|
||||
return 0, uploadErr
|
||||
}
|
||||
|
||||
return totalWritten, nil
|
||||
}
|
||||
|
||||
// parseTusMetadata parses the Upload-Metadata header
|
||||
// Format: key1 base64value1,key2 base64value2,...
|
||||
func parseTusMetadata(header string) map[string]string {
|
||||
metadata := make(map[string]string)
|
||||
if header == "" {
|
||||
return metadata
|
||||
}
|
||||
|
||||
pairs := strings.Split(header, ",")
|
||||
for _, pair := range pairs {
|
||||
pair = strings.TrimSpace(pair)
|
||||
parts := strings.SplitN(pair, " ", 2)
|
||||
if len(parts) != 2 {
|
||||
continue
|
||||
}
|
||||
key := strings.TrimSpace(parts[0])
|
||||
encodedValue := strings.TrimSpace(parts[1])
|
||||
|
||||
value, err := base64.StdEncoding.DecodeString(encodedValue)
|
||||
if err != nil {
|
||||
glog.V(1).Infof("Failed to decode TUS metadata value for key %s: %v", key, err)
|
||||
continue
|
||||
}
|
||||
metadata[key] = string(value)
|
||||
}
|
||||
|
||||
return metadata
|
||||
}
|
||||
448
weed/server/filer_server_tus_session.go
Normal file
448
weed/server/filer_server_tus_session.go
Normal file
@@ -0,0 +1,448 @@
|
||||
package weed_server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
const (
|
||||
TusVersion = "1.0.0"
|
||||
TusMaxSize = int64(5 * 1024 * 1024 * 1024) // 5GB default max size
|
||||
TusUploadsFolder = ".uploads.tus"
|
||||
TusInfoFileName = ".info"
|
||||
TusChunkExt = ".chunk"
|
||||
TusExtensions = "creation,creation-with-upload,termination"
|
||||
)
|
||||
|
||||
// TusSession represents an in-progress TUS upload session
|
||||
type TusSession struct {
|
||||
ID string `json:"id"`
|
||||
TargetPath string `json:"target_path"`
|
||||
Size int64 `json:"size"`
|
||||
Offset int64 `json:"offset"`
|
||||
Metadata map[string]string `json:"metadata,omitempty"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
ExpiresAt time.Time `json:"expires_at,omitempty"`
|
||||
Chunks []*TusChunkInfo `json:"chunks,omitempty"`
|
||||
}
|
||||
|
||||
// TusChunkInfo tracks individual chunk uploads within a session
|
||||
type TusChunkInfo struct {
|
||||
Offset int64 `json:"offset"`
|
||||
Size int64 `json:"size"`
|
||||
FileId string `json:"file_id"`
|
||||
UploadAt int64 `json:"upload_at"`
|
||||
}
|
||||
|
||||
// tusSessionDir returns the directory path for storing TUS upload sessions
|
||||
func (fs *FilerServer) tusSessionDir() string {
|
||||
return "/" + TusUploadsFolder
|
||||
}
|
||||
|
||||
// tusSessionPath returns the path to a specific upload session directory
|
||||
func (fs *FilerServer) tusSessionPath(uploadID string) string {
|
||||
return fmt.Sprintf("/%s/%s", TusUploadsFolder, uploadID)
|
||||
}
|
||||
|
||||
// tusSessionInfoPath returns the path to the session info file
|
||||
func (fs *FilerServer) tusSessionInfoPath(uploadID string) string {
|
||||
return fmt.Sprintf("/%s/%s/%s", TusUploadsFolder, uploadID, TusInfoFileName)
|
||||
}
|
||||
|
||||
// tusChunkPath returns the path to store a chunk info file
|
||||
// Format: /{TusUploadsFolder}/{uploadID}/chunk_{offset}_{size}_{encodedFileId}
|
||||
func (fs *FilerServer) tusChunkPath(uploadID string, offset, size int64, fileId string) string {
|
||||
// Use URL-safe base64 encoding to safely encode fileId (handles both / and _ in fileId)
|
||||
encodedFileId := base64.RawURLEncoding.EncodeToString([]byte(fileId))
|
||||
return fmt.Sprintf("/%s/%s/chunk_%016d_%016d_%s", TusUploadsFolder, uploadID, offset, size, encodedFileId)
|
||||
}
|
||||
|
||||
// parseTusChunkPath parses chunk info from a chunk entry
|
||||
// The entry's Crtime is used for the UploadAt timestamp to preserve the actual upload time
|
||||
func parseTusChunkPath(entry *filer.Entry) (*TusChunkInfo, error) {
|
||||
name := entry.Name()
|
||||
if !strings.HasPrefix(name, "chunk_") {
|
||||
return nil, fmt.Errorf("not a chunk file: %s", name)
|
||||
}
|
||||
// Use strings.Cut to correctly handle base64-encoded fileId which may contain underscores
|
||||
s := name[6:] // Skip "chunk_" prefix
|
||||
offsetStr, rest, found := strings.Cut(s, "_")
|
||||
if !found {
|
||||
return nil, fmt.Errorf("invalid chunk file name format (missing offset): %s", name)
|
||||
}
|
||||
sizeStr, encodedFileId, found := strings.Cut(rest, "_")
|
||||
if !found {
|
||||
return nil, fmt.Errorf("invalid chunk file name format (missing size): %s", name)
|
||||
}
|
||||
|
||||
offset, err := strconv.ParseInt(offsetStr, 10, 64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid offset in chunk file %q: %w", name, err)
|
||||
}
|
||||
size, err := strconv.ParseInt(sizeStr, 10, 64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid size in chunk file %q: %w", name, err)
|
||||
}
|
||||
// Decode fileId from URL-safe base64
|
||||
fileIdBytes, err := base64.RawURLEncoding.DecodeString(encodedFileId)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid fileId encoding in chunk file %q: %w", name, err)
|
||||
}
|
||||
return &TusChunkInfo{
|
||||
Offset: offset,
|
||||
Size: size,
|
||||
FileId: string(fileIdBytes),
|
||||
UploadAt: entry.Crtime.UnixNano(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// createTusSession creates a new TUS upload session
|
||||
func (fs *FilerServer) createTusSession(ctx context.Context, uploadID, targetPath string, size int64, metadata map[string]string) (*TusSession, error) {
|
||||
session := &TusSession{
|
||||
ID: uploadID,
|
||||
TargetPath: targetPath,
|
||||
Size: size,
|
||||
Offset: 0,
|
||||
Metadata: metadata,
|
||||
CreatedAt: time.Now(),
|
||||
ExpiresAt: time.Now().Add(7 * 24 * time.Hour), // 7 days default expiration
|
||||
Chunks: []*TusChunkInfo{},
|
||||
}
|
||||
|
||||
// Create session directory
|
||||
sessionDirPath := util.FullPath(fs.tusSessionPath(uploadID))
|
||||
if err := fs.filer.CreateEntry(ctx, &filer.Entry{
|
||||
FullPath: sessionDirPath,
|
||||
Attr: filer.Attr{
|
||||
Mode: os.ModeDir | 0755,
|
||||
Crtime: time.Now(),
|
||||
Mtime: time.Now(),
|
||||
Uid: OS_UID,
|
||||
Gid: OS_GID,
|
||||
},
|
||||
}, false, false, nil, false, fs.filer.MaxFilenameLength); err != nil {
|
||||
return nil, fmt.Errorf("create session directory: %w", err)
|
||||
}
|
||||
|
||||
// Save session info
|
||||
if err := fs.saveTusSession(ctx, session); err != nil {
|
||||
// Cleanup the directory on failure
|
||||
fs.filer.DeleteEntryMetaAndData(ctx, sessionDirPath, true, true, false, false, nil, 0)
|
||||
return nil, fmt.Errorf("save session info: %w", err)
|
||||
}
|
||||
|
||||
glog.V(2).Infof("Created TUS session %s for %s, size=%d", uploadID, targetPath, size)
|
||||
return session, nil
|
||||
}
|
||||
|
||||
// saveTusSession saves the session info to the filer
|
||||
func (fs *FilerServer) saveTusSession(ctx context.Context, session *TusSession) error {
|
||||
sessionData, err := json.Marshal(session)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal session: %w", err)
|
||||
}
|
||||
|
||||
infoPath := util.FullPath(fs.tusSessionInfoPath(session.ID))
|
||||
entry := &filer.Entry{
|
||||
FullPath: infoPath,
|
||||
Attr: filer.Attr{
|
||||
Mode: 0644,
|
||||
Crtime: session.CreatedAt,
|
||||
Mtime: time.Now(),
|
||||
Uid: OS_UID,
|
||||
Gid: OS_GID,
|
||||
},
|
||||
Content: sessionData,
|
||||
}
|
||||
|
||||
if err := fs.filer.CreateEntry(ctx, entry, false, false, nil, false, fs.filer.MaxFilenameLength); err != nil {
|
||||
return fmt.Errorf("save session info entry: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// getTusSession retrieves a TUS session by upload ID, including chunks from directory listing
|
||||
func (fs *FilerServer) getTusSession(ctx context.Context, uploadID string) (*TusSession, error) {
|
||||
infoPath := util.FullPath(fs.tusSessionInfoPath(uploadID))
|
||||
entry, err := fs.filer.FindEntry(ctx, infoPath)
|
||||
if err != nil {
|
||||
if err == filer_pb.ErrNotFound {
|
||||
return nil, fmt.Errorf("TUS upload session not found: %s", uploadID)
|
||||
}
|
||||
return nil, fmt.Errorf("find session: %w", err)
|
||||
}
|
||||
|
||||
var session TusSession
|
||||
if err := json.Unmarshal(entry.Content, &session); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal session: %w", err)
|
||||
}
|
||||
|
||||
// Load chunks from directory listing with pagination (atomic read, no race condition)
|
||||
sessionDirPath := util.FullPath(fs.tusSessionPath(uploadID))
|
||||
session.Chunks = nil
|
||||
session.Offset = 0
|
||||
|
||||
lastFileName := ""
|
||||
pageSize := 1000
|
||||
for {
|
||||
entries, hasMore, err := fs.filer.ListDirectoryEntries(ctx, sessionDirPath, lastFileName, false, int64(pageSize), "", "", "")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list session directory: %w", err)
|
||||
}
|
||||
|
||||
for _, e := range entries {
|
||||
if strings.HasPrefix(e.Name(), "chunk_") {
|
||||
chunk, parseErr := parseTusChunkPath(e)
|
||||
if parseErr != nil {
|
||||
glog.V(1).Infof("Skipping invalid chunk file %s: %v", e.Name(), parseErr)
|
||||
continue
|
||||
}
|
||||
session.Chunks = append(session.Chunks, chunk)
|
||||
}
|
||||
lastFileName = e.Name()
|
||||
}
|
||||
|
||||
if !hasMore || len(entries) < pageSize {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Sort chunks by offset and compute current offset as maximum contiguous range from 0
|
||||
if len(session.Chunks) > 0 {
|
||||
sort.Slice(session.Chunks, func(i, j int) bool {
|
||||
return session.Chunks[i].Offset < session.Chunks[j].Offset
|
||||
})
|
||||
// Compute the maximum contiguous offset from 0
|
||||
// This correctly handles gaps in the upload sequence
|
||||
contiguousEnd := int64(0)
|
||||
for _, chunk := range session.Chunks {
|
||||
if chunk.Offset > contiguousEnd {
|
||||
// Gap detected, stop at the first gap
|
||||
break
|
||||
}
|
||||
chunkEnd := chunk.Offset + chunk.Size
|
||||
if chunkEnd > contiguousEnd {
|
||||
contiguousEnd = chunkEnd
|
||||
}
|
||||
}
|
||||
session.Offset = contiguousEnd
|
||||
}
|
||||
|
||||
return &session, nil
|
||||
}
|
||||
|
||||
// saveTusChunk stores the chunk info as a separate file entry
|
||||
// This avoids read-modify-write race conditions across multiple filer instances
|
||||
// The chunk metadata is encoded in the filename; the entry's Crtime preserves upload time
|
||||
func (fs *FilerServer) saveTusChunk(ctx context.Context, uploadID string, chunk *TusChunkInfo) error {
|
||||
if chunk == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Store chunk info as a separate file entry (atomic operation)
|
||||
// Chunk metadata is encoded in the filename; Crtime is used for UploadAt when reading back
|
||||
chunkPath := util.FullPath(fs.tusChunkPath(uploadID, chunk.Offset, chunk.Size, chunk.FileId))
|
||||
|
||||
if err := fs.filer.CreateEntry(ctx, &filer.Entry{
|
||||
FullPath: chunkPath,
|
||||
Attr: filer.Attr{
|
||||
Mode: 0644,
|
||||
Crtime: time.Now(),
|
||||
Mtime: time.Now(),
|
||||
Uid: OS_UID,
|
||||
Gid: OS_GID,
|
||||
},
|
||||
}, false, false, nil, false, fs.filer.MaxFilenameLength); err != nil {
|
||||
return fmt.Errorf("save chunk info: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// deleteTusSession removes a TUS upload session and all its data
|
||||
func (fs *FilerServer) deleteTusSession(ctx context.Context, uploadID string) error {
|
||||
|
||||
session, err := fs.getTusSession(ctx, uploadID)
|
||||
if err != nil {
|
||||
// Session might already be deleted or never existed
|
||||
glog.V(1).Infof("TUS session %s not found for deletion: %v", uploadID, err)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Batch delete all uploaded chunks from volume servers
|
||||
if len(session.Chunks) > 0 {
|
||||
var chunksToDelete []*filer_pb.FileChunk
|
||||
for _, chunk := range session.Chunks {
|
||||
if chunk.FileId != "" {
|
||||
chunksToDelete = append(chunksToDelete, &filer_pb.FileChunk{FileId: chunk.FileId})
|
||||
}
|
||||
}
|
||||
if len(chunksToDelete) > 0 {
|
||||
fs.filer.DeleteChunks(ctx, util.FullPath(session.TargetPath), chunksToDelete)
|
||||
}
|
||||
}
|
||||
|
||||
// Delete the session directory
|
||||
sessionDirPath := util.FullPath(fs.tusSessionPath(uploadID))
|
||||
if err := fs.filer.DeleteEntryMetaAndData(ctx, sessionDirPath, true, true, false, false, nil, 0); err != nil {
|
||||
return fmt.Errorf("delete session directory: %w", err)
|
||||
}
|
||||
|
||||
glog.V(2).Infof("Deleted TUS session %s", uploadID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// completeTusUpload assembles all chunks and creates the final file
|
||||
func (fs *FilerServer) completeTusUpload(ctx context.Context, session *TusSession) error {
|
||||
if session.Offset != session.Size {
|
||||
return fmt.Errorf("upload incomplete: offset=%d, expected=%d", session.Offset, session.Size)
|
||||
}
|
||||
|
||||
// Sort chunks by offset to ensure correct order
|
||||
sort.Slice(session.Chunks, func(i, j int) bool {
|
||||
return session.Chunks[i].Offset < session.Chunks[j].Offset
|
||||
})
|
||||
|
||||
// Validate chunks are contiguous with no gaps or overlaps
|
||||
expectedOffset := int64(0)
|
||||
for _, chunk := range session.Chunks {
|
||||
if chunk.Offset != expectedOffset {
|
||||
return fmt.Errorf("chunk gap or overlap detected: expected offset %d, got %d", expectedOffset, chunk.Offset)
|
||||
}
|
||||
expectedOffset = chunk.Offset + chunk.Size
|
||||
}
|
||||
if expectedOffset != session.Size {
|
||||
return fmt.Errorf("chunks do not cover full file: chunks end at %d, expected %d", expectedOffset, session.Size)
|
||||
}
|
||||
|
||||
// Assemble file chunks in order
|
||||
var fileChunks []*filer_pb.FileChunk
|
||||
|
||||
for _, chunk := range session.Chunks {
|
||||
fid, fidErr := filer_pb.ToFileIdObject(chunk.FileId)
|
||||
if fidErr != nil {
|
||||
return fmt.Errorf("invalid file ID %s at offset %d: %w", chunk.FileId, chunk.Offset, fidErr)
|
||||
}
|
||||
|
||||
fileChunk := &filer_pb.FileChunk{
|
||||
FileId: chunk.FileId,
|
||||
Offset: chunk.Offset,
|
||||
Size: uint64(chunk.Size),
|
||||
ModifiedTsNs: chunk.UploadAt,
|
||||
Fid: fid,
|
||||
}
|
||||
fileChunks = append(fileChunks, fileChunk)
|
||||
}
|
||||
|
||||
// Determine content type from metadata
|
||||
contentType := ""
|
||||
if session.Metadata != nil {
|
||||
if ct, ok := session.Metadata["content-type"]; ok {
|
||||
contentType = ct
|
||||
}
|
||||
}
|
||||
|
||||
// Create the final file entry
|
||||
targetPath := util.FullPath(session.TargetPath)
|
||||
entry := &filer.Entry{
|
||||
FullPath: targetPath,
|
||||
Attr: filer.Attr{
|
||||
Mode: 0644,
|
||||
Crtime: session.CreatedAt,
|
||||
Mtime: time.Now(),
|
||||
Uid: OS_UID,
|
||||
Gid: OS_GID,
|
||||
Mime: contentType,
|
||||
},
|
||||
Chunks: fileChunks,
|
||||
}
|
||||
|
||||
// Ensure parent directory exists
|
||||
if err := fs.filer.CreateEntry(ctx, entry, false, false, nil, false, fs.filer.MaxFilenameLength); err != nil {
|
||||
return fmt.Errorf("create final file entry: %w", err)
|
||||
}
|
||||
|
||||
// Delete the session (but keep the chunks since they're now part of the final file)
|
||||
sessionDirPath := util.FullPath(fs.tusSessionPath(session.ID))
|
||||
if err := fs.filer.DeleteEntryMetaAndData(ctx, sessionDirPath, true, false, false, false, nil, 0); err != nil {
|
||||
glog.V(1).Infof("Failed to cleanup TUS session directory %s: %v", session.ID, err)
|
||||
}
|
||||
|
||||
glog.V(2).Infof("Completed TUS upload %s -> %s, size=%d, chunks=%d",
|
||||
session.ID, session.TargetPath, session.Size, len(fileChunks))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// StartTusSessionCleanup starts a background goroutine that periodically cleans up expired TUS sessions
|
||||
func (fs *FilerServer) StartTusSessionCleanup(interval time.Duration) {
|
||||
go func() {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for range ticker.C {
|
||||
fs.cleanupExpiredTusSessions()
|
||||
}
|
||||
}()
|
||||
glog.V(0).Infof("TUS session cleanup started with interval %v", interval)
|
||||
}
|
||||
|
||||
// cleanupExpiredTusSessions scans for and removes expired TUS upload sessions
|
||||
func (fs *FilerServer) cleanupExpiredTusSessions() {
|
||||
ctx := context.Background()
|
||||
uploadsDir := util.FullPath(fs.tusSessionDir())
|
||||
|
||||
// List all session directories under the TUS uploads folder
|
||||
var lastFileName string
|
||||
const pageSize = 100
|
||||
|
||||
for {
|
||||
entries, hasMore, err := fs.filer.ListDirectoryEntries(ctx, uploadsDir, lastFileName, false, int64(pageSize), "", "", "")
|
||||
if err != nil {
|
||||
glog.V(1).Infof("TUS cleanup: failed to list sessions: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
for _, entry := range entries {
|
||||
if !entry.IsDirectory() {
|
||||
lastFileName = entry.Name()
|
||||
continue
|
||||
}
|
||||
|
||||
uploadID := entry.Name()
|
||||
session, err := fs.getTusSession(ctx, uploadID)
|
||||
if err != nil {
|
||||
glog.V(2).Infof("TUS cleanup: skipping session %s: %v", uploadID, err)
|
||||
lastFileName = uploadID
|
||||
continue
|
||||
}
|
||||
|
||||
if !session.ExpiresAt.IsZero() && now.After(session.ExpiresAt) {
|
||||
glog.V(1).Infof("TUS cleanup: removing expired session %s (expired at %v)", uploadID, session.ExpiresAt)
|
||||
if err := fs.deleteTusSession(ctx, uploadID); err != nil {
|
||||
glog.V(1).Infof("TUS cleanup: failed to delete session %s: %v", uploadID, err)
|
||||
}
|
||||
}
|
||||
|
||||
lastFileName = uploadID
|
||||
}
|
||||
|
||||
if !hasMore || len(entries) < pageSize {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user