Files
seaweedFS/weed/s3api/s3api_object_handlers_put.go
Chris Lu 5075381060 Support multiple filers for S3 and IAM servers with automatic failover (#7550)
* Support multiple filers for S3 and IAM servers with automatic failover

This change adds support for multiple filer addresses in the 'weed s3' and 'weed iam' commands, enabling high availability through automatic failover.

Key changes:
- Updated S3ApiServerOption.Filer to Filers ([]pb.ServerAddress)
- Updated IamServerOption.Filer to Filers ([]pb.ServerAddress)
- Modified -filer flag to accept comma-separated addresses
- Added getFilerAddress() helper methods for backward compatibility
- Updated all filer client calls to support multiple addresses
- Uses pb.WithOneOfGrpcFilerClients for automatic failover

Usage:
  weed s3 -filer=localhost:8888,localhost:8889
  weed iam -filer=localhost:8888,localhost:8889

The underlying FilerClient already supported multiple filers with health
tracking and automatic failover - this change exposes that capability
through the command-line interface.

* Add filer discovery: treat initial filers as seeds and discover peers from master

Enhances FilerClient to automatically discover additional filers in the same
filer group by querying the master server. This allows users to specify just
a few seed filers, and the client will discover all other filers in the cluster.

Key changes to wdclient/FilerClient:
- Added MasterClient, FilerGroup, and DiscoveryInterval fields
- Added thread-safe filer list management with RWMutex
- Implemented discoverFilers() background goroutine
- Uses cluster.ListExistingPeerUpdates() to query master for filers
- Automatically adds newly discovered filers to the list
- Added Close() method to clean up discovery goroutine

New FilerClientOption fields:
- MasterClient: enables filer discovery from master
- FilerGroup: specifies which filer group to discover
- DiscoveryInterval: how often to refresh (default 5 minutes)

Usage example:
  masterClient := wdclient.NewMasterClient(...)
  filerClient := wdclient.NewFilerClient(
    []pb.ServerAddress{"localhost:8888"}, // seed filers
    grpcDialOption,
    dataCenter,
    &wdclient.FilerClientOption{
      MasterClient: masterClient,
      FilerGroup: "my-group",
    },
  )
  defer filerClient.Close()

The initial filers act as seeds - the client discovers and adds all other
filers in the same group from the master. Discovered filers are added
dynamically without removing existing ones (relying on health checks for
unavailable filers).

* Address PR review comments: implement full failover for IAM operations

Critical fixes based on code review feedback:

1. **IAM API Failover (Critical)**:
   - Replace pb.WithGrpcFilerClient with pb.WithOneOfGrpcFilerClients in:
     * GetS3ApiConfigurationFromFiler()
     * PutS3ApiConfigurationToFiler()
     * GetPolicies()
     * PutPolicies()
   - Now all IAM operations support automatic failover across multiple filers

2. **Validation Improvements**:
   - Add validation in NewIamApiServerWithStore() to require at least one filer
   - Add validation in NewS3ApiServerWithStore() to require at least one filer
   - Add warning log when no filers configured for credential store

3. **Error Logging**:
   - Circuit breaker now logs when config load fails instead of silently ignoring
   - Helps operators understand why circuit breaker limits aren't applied

4. **Code Quality**:
   - Use ToGrpcAddress() for filer address in credential store setup
   - More consistent with rest of codebase and future-proof

These changes ensure IAM operations have the same high availability guarantees
as S3 operations, completing the multi-filer failover implementation.

* Fix IAM manager initialization: remove code duplication, add TODO for HA

Addresses review comment on s3api_server.go:145

Changes:
- Remove duplicate code for getting first filer address
- Extract filerAddr variable once and reuse
- Add TODO comment documenting the HA limitation for IAM manager
- Document that loadIAMManagerFromConfig and NewS3IAMIntegration need
  updates to support multiple filers for full HA

Note: This is a known limitation when using filer-backed IAM stores.
The interfaces need to be updated to accept multiple filer addresses.
For now, documenting this limitation clearly.

* Document credential store HA limitation with TODO

Addresses review comment on auth_credentials.go:149

Changes:
- Add TODO comment documenting that SetFilerClient interface needs update
  for multi-filer support
- Add informative log message indicating HA limitation
- Document that this is a known limitation for filer-backed credential stores

The SetFilerClient interface currently only accepts a single filer address.
To properly support HA, the credential store interfaces need to be updated
to handle multiple filer addresses.

* Track current active filer in FilerClient for better HA

Add GetCurrentFiler() method to FilerClient that returns the currently
active filer based on the filerIndex which is updated on successful
operations. This provides better availability than always using the
first filer.

Changes:
- Add FilerClient.GetCurrentFiler() method that returns current active filer
- Update S3ApiServer.getFilerAddress() to use FilerClient's current filer
- Add fallback to first filer if FilerClient not yet initialized
- Document IAM limitation (doesn't have FilerClient access)

Benefits:
- Single-filer operations (URLs, ReadFilerConf, etc.) now use the
  currently active/healthy filer
- Better distribution and failover behavior
- FilerClient's round-robin and health tracking automatically
  determines which filer to use

* Document ReadFilerConf HA limitation in lifecycle handlers

Addresses review comment on s3api_bucket_handlers.go:880

Add comment documenting that ReadFilerConf uses the current active filer
from FilerClient (which is better than always using first filer), but
doesn't have built-in multi-filer failover.

Add TODO to update filer.ReadFilerConf to support multiple filers for
complete HA. For now, it uses the currently active/healthy filer tracked
by FilerClient which provides reasonable availability.

* Document multipart upload URL HA limitation

Addresses review comment on s3api_object_handlers_multipart.go:442

Add comment documenting that part upload URLs point to the current
active filer (tracked by FilerClient), which is better than always
using the first filer but still creates a potential point of failure
if that filer becomes unavailable during upload.

Suggest TODO solutions:
- Use virtual hostname/load balancer for filers
- Have S3 server proxy uploads to healthy filers

Current behavior provides reasonable availability by using the
currently active/healthy filer rather than being pinned to first filer.

* Document multipart completion Location URL limitation

Addresses review comment on filer_multipart.go:187

Add comment documenting that the Location URL in CompleteMultipartUpload
response points to the current active filer (tracked by FilerClient).

Note that clients should ideally use the S3 API endpoint rather than
this direct URL. If direct access is attempted and the specific filer
is unavailable, the request will fail.

Current behavior uses the currently active/healthy filer rather than
being pinned to the first filer, providing better availability.

* Make credential store use current active filer for HA

Update FilerEtcStore to use a function that returns the current active
filer instead of a fixed address, enabling high availability.

Changes:
- Add SetFilerAddressFunc() method to FilerEtcStore
- Store uses filerAddressFunc instead of fixed filerGrpcAddress
- withFilerClient() calls the function to get current active filer
- Keep SetFilerClient() for backward compatibility (marked deprecated)
- Update S3ApiServer to pass FilerClient.GetCurrentFiler to store

Benefits:
- Credential store now uses currently active/healthy filer
- Automatic failover when filer becomes unavailable
- True HA for credential operations
- Backward compatible with old SetFilerClient interface

This addresses the credential store limitation - no longer pinned to
first filer, uses FilerClient's tracked current active filer.

* Clarify multipart URL comments: filer address not used for uploads

Update comments to reflect that multipart upload URLs are not actually
used for upload traffic - uploads go directly to volume servers.

Key clarifications:
- genPartUploadUrl: Filer address is parsed out, only path is used
- CompleteMultipartUpload Location: Informational field per AWS S3 spec
- Actual uploads bypass filer proxy and go directly to volume servers

The filer address in these URLs is NOT a HA concern because:
1. Part uploads: URL is parsed for path, upload goes to volume servers
2. Location URL: Informational only, clients use S3 endpoint

This addresses the observation that S3 uploads don't go through filers,
only metadata operations do.

* Remove filer address from upload paths - pass path directly

Eliminate unnecessary filer address from upload URLs by passing file
paths directly instead of full URLs that get immediately parsed.

Changes:
- Rename genPartUploadUrl() → genPartUploadPath() (returns path only)
- Rename toFilerUrl() → toFilerPath() (returns path only)
- Update putToFiler() to accept filePath instead of uploadUrl
- Remove URL parsing code (no longer needed)
- Remove net/url import (no longer used)
- Keep old function names as deprecated wrappers for compatibility

Benefits:
- Cleaner code - no fake URL construction/parsing
- No dependency on filer address for internal operations
- More accurate naming (these are paths, not URLs)
- Eliminates confusion about HA concerns

This completely removes the filer address from upload operations - it was
never actually used for routing, only parsed for the path.

* Remove deprecated functions: use new path-based functions directly

Remove deprecated wrapper functions and update all callers to use the
new function names directly.

Removed:
- genPartUploadUrl() → all callers now use genPartUploadPath()
- toFilerUrl() → all callers now use toFilerPath()
- SetFilerClient() → removed along with fallback code

Updated:
- s3api_object_handlers_multipart.go: uploadUrl → filePath
- s3api_object_handlers_put.go: uploadUrl → filePath, versionUploadUrl → versionFilePath
- s3api_object_versioning.go: toFilerUrl → toFilerPath
- s3api_object_handlers_test.go: toFilerUrl → toFilerPath
- auth_credentials.go: removed SetFilerClient fallback
- filer_etc_store.go: removed deprecated SetFilerClient method

Benefits:
- Cleaner codebase with no deprecated functions
- All variable names accurately reflect that they're paths, not URLs
- Single interface for credential stores (SetFilerAddressFunc only)

All code now consistently uses the new path-based approach.

* Fix toFilerPath: remove URL escaping for raw file paths

The toFilerPath function should return raw file paths, not URL-escaped
paths. URL escaping was needed when the path was embedded in a URL
(old toFilerUrl), but now that we pass paths directly to putToFiler,
they should be unescaped.

This fixes S3 integration test failures:
- test_bucket_listv2_encoding_basic
- test_bucket_list_encoding_basic
- test_bucket_listv2_delimiter_whitespace
- test_bucket_list_delimiter_whitespace

The tests were failing because paths were double-encoded (escaped when
stored, then escaped again when listed), resulting in %252B instead of
%2B for '+' characters.

Root cause: When we removed URL parsing in putToFiler, we should have
also removed URL escaping in toFilerPath since paths are now used
directly without URL encoding/decoding.

* Add thread safety to FilerEtcStore and clarify credential store comments

Address review suggestions for better thread safety and code clarity:

1. **Thread Safety**: Add RWMutex to FilerEtcStore
   - Protects filerAddressFunc and grpcDialOption from concurrent access
   - Initialize() uses write lock when setting function
   - SetFilerAddressFunc() uses write lock
   - withFilerClient() uses read lock to get function and dial option
   - GetPolicies() uses read lock to check if configured

2. **Improved Error Messages**:
   - Prefix errors with "filer_etc:" for easier debugging
   - "filer address not configured" → "filer_etc: filer address function not configured"
   - "filer address is empty" → "filer_etc: filer address is empty"

3. **Clarified Comments**:
   - auth_credentials.go: Clarify that initial setup is temporary
   - Document that it's updated in s3api_server.go after FilerClient creation
   - Remove ambiguity about when FilerClient.GetCurrentFiler is used

Benefits:
- Safe for concurrent credential operations
- Clear error messages for debugging
- Explicit documentation of initialization order

* Enable filer discovery: pass master addresses to FilerClient

Fix two critical issues:

1. **Filer Discovery Not Working**: Master client was not being passed to
   FilerClient, so peer discovery couldn't work

2. **Credential Store Design**: Already uses FilerClient via GetCurrentFiler
   function - this is the correct design for HA

Changes:

**Command (s3.go):**
- Read master addresses from GetFilerConfiguration response
- Pass masterAddresses to S3ApiServerOption
- Log master addresses for visibility

**S3ApiServerOption:**
- Add Masters []pb.ServerAddress field for discovery

**S3ApiServer:**
- Create MasterClient from Masters when available
- Pass MasterClient + FilerGroup to FilerClient via options
- Enable discovery with 5-minute refresh interval
- Log whether discovery is enabled or disabled

**Credential Store:**
- Already correctly uses filerClient.GetCurrentFiler via function
- This provides HA without tight coupling to FilerClient struct
- Function-based design is clean and thread-safe

Discovery Flow:
1. S3 command reads filer config → gets masters + filer group
2. S3ApiServer creates MasterClient from masters
3. FilerClient uses MasterClient to query for peer filers
4. Background goroutine refreshes peer list every 5 minutes
5. Credential store uses GetCurrentFiler to get active filer

Now filer discovery actually works! ��

* Use S3 endpoint in multipart Location instead of filer address

* Add multi-filer failover to ReadFilerConf

* Address CodeRabbit review: fix buffer reuse and improve lock safety

Address two code review suggestions:

1. **Fix buffer reuse in ReadFilerConfFromFilers**:
   - Use local []byte data instead of shared buffer
   - Prevents partial data from failed attempts affecting successful reads
   - Creates fresh buffer inside callback for masterClient path
   - More robust to future changes in read helpers

2. **Improve lock safety in FilerClient**:
   - Add *WithHealth variants that accept health pointer
   - Get health pointer while holding lock, then release before calling
   - Eliminates potential for lock confusion (though no actual deadlock existed)
   - Clearer separation: lock for data access, atomics for health ops

Changes:
- ReadFilerConfFromFilers: var data []byte, create buf inside callback
- shouldSkipUnhealthyFilerWithHealth(health *filerHealth)
- recordFilerSuccessWithHealth(health *filerHealth)
- recordFilerFailureWithHealth(health *filerHealth)
- Keep old functions for backward compatibility (marked deprecated)
- Update LookupVolumeIds to use WithHealth variants

Benefits:
- More robust multi-filer configuration reading
- Clearer lock vs atomic operation boundaries
- No lock held during health checks (even though atomics don't block)
- Better code organization and maintainability

* add constant

* Fix IAM manager and post policy to use current active filer

* Fix critical race condition and goroutine leak

* Update weed/s3api/filer_multipart.go

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Fix compilation error and address code review suggestions

Address remaining unresolved comments:

1. **Fix compilation error**: Add missing net/url import
   - filer_multipart.go used url.PathEscape without import
   - Added "net/url" to imports

2. **Fix Location URL formatting** (all 4 occurrences):
   - Add missing slash between bucket and key
   - Use url.PathEscape for bucket names
   - Use urlPathEscape for object keys
   - Handles special characters in bucket/key names
   - Before: http://host/bucketkey
   - After: http://host/bucket/key (properly escaped)

3. **Optimize discovery loop** (O(N*M) → O(N+M)):
   - Use map for existing filers (O(1) lookup)
   - Reduces time holding write lock
   - Better performance with many filers
   - Before: Nested loop for each discovered filer
   - After: Build map once, then O(1) lookups

Changes:
- filer_multipart.go: Import net/url, fix all Location URLs
- filer_client.go: Use map for efficient filer discovery

Benefits:
- Compiles successfully
- Proper URL encoding (handles spaces, special chars)
- Faster discovery with less lock contention
- Production-ready URL formatting

* Fix race conditions and make Close() idempotent

Address CodeRabbit review #3512078995:

1. **Critical: Fix unsynchronized read in error message**
   - Line 584 read len(fc.filerAddresses) without lock
   - Race with refreshFilerList appending to slice
   - Fixed: Take RLock to read length safely
   - Prevents race detector warnings

2. **Important: Make Close() idempotent**
   - Closing already-closed channel panics
   - Can happen with layered cleanup in shutdown paths
   - Fixed: Use sync.Once to ensure single close
   - Safe to call Close() multiple times now

3. **Nitpick: Add warning for empty filer address**
   - getFilerAddress() can return empty string
   - Helps diagnose unexpected state
   - Added: Warning log when no filers available

4. **Nitpick: Guard deprecated index-based helpers**
   - shouldSkipUnhealthyFiler, recordFilerSuccess/Failure
   - Accessed filerHealth without lock (races with discovery)
   - Fixed: Take RLock and check bounds before array access
   - Prevents index out of bounds and races

Changes:
- filer_client.go:
  - Add closeDiscoveryOnce sync.Once field
  - Use Do() in Close() for idempotent channel close
  - Add RLock guards to deprecated index-based helpers
  - Add bounds checking to prevent panics
  - Synchronized read of filerAddresses length in error

- s3api_server.go:
  - Add warning log when getFilerAddress returns empty

Benefits:
- No race conditions (passes race detector)
- No panic on double-close
- Better error diagnostics
- Safe with discovery enabled
- Production-hardened shutdown logic

* Fix hardcoded http scheme and add panic recovery

Address CodeRabbit review #3512114811:

1. **Major: Fix hardcoded http:// scheme in Location URLs**
   - Location URLs always used http:// regardless of client connection
   - HTTPS clients got http:// URLs (incorrect)
   - Fixed: Detect scheme from request
   - Check X-Forwarded-Proto header (for proxies) first
   - Check r.TLS != nil for direct HTTPS
   - Fallback to http for plain connections
   - Applied to all 4 CompleteMultipartUploadResult locations

2. **Major: Add panic recovery to discovery goroutine**
   - Long-running background goroutine could crash entire process
   - Panic in refreshFilerList would terminate program
   - Fixed: Add defer recover() with error logging
   - Goroutine failures now logged, not fatal

3. **Note: Close() idempotency already implemented**
   - Review flagged as duplicate issue
   - Already fixed in commit 3d7a65c7e
   - sync.Once (closeDiscoveryOnce) prevents double-close panic
   - Safe to call Close() multiple times

Changes:
- filer_multipart.go:
  - Add getRequestScheme() helper function
  - Update all 4 Location URLs to use dynamic scheme
  - Format: scheme://host/bucket/key (was: http://...)

- filer_client.go:
  - Add panic recovery to discoverFilers()
  - Log panics instead of crashing

Benefits:
- Correct scheme (https/http) in Location URLs
- Works behind proxies (X-Forwarded-Proto)
- No process crashes from discovery failures
- Production-hardened background goroutine
- Proper AWS S3 API compliance

* Fix S3 WithFilerClient to use filer failover

Critical fix for multi-filer deployments:

**Problem:**
- S3ApiServer.WithFilerClient() was creating direct connections to ONE filer
- Used pb.WithGrpcClient() with single filer address
- No failover - if that filer failed, ALL operations failed
- Caused test failures: "bucket directory not found"
- IAM Integration Tests failing with 500 Internal Error

**Root Cause:**
- WithFilerClient bypassed filerClient connection management
- Always connected to getFilerAddress() (current filer only)
- Didn't retry other filers on failure
- All getEntry(), updateEntry(), etc. operations failed if current filer down

**Solution:**
1. Added FilerClient.GetAllFilers() method
   - Returns snapshot of all filer addresses
   - Thread-safe copy to avoid races

2. Implemented withFilerClientFailover()
   - Try current filer first (fast path)
   - On failure, try all other filers
   - Log successful failover
   - Return error only if ALL filers fail

3. Updated WithFilerClient()
   - Use filerClient for failover when available
   - Fallback to direct connection for testing/init

**Impact:**
 All S3 operations now support multi-filer failover
 Bucket metadata reads work with any available filer
 Entry operations (getEntry, updateEntry) failover automatically
 IAM tests should pass now
 Production-ready HA support

**Files Changed:**
- wdclient/filer_client.go: Add GetAllFilers() method
- s3api/s3api_handlers.go: Implement failover logic

This fixes the test failure where bucket operations failed when
the primary filer was temporarily unavailable during cleanup.

* Update current filer after successful failover

Address code review: https://github.com/seaweedfs/seaweedfs/pull/7550#pullrequestreview-3512223723

**Issue:**
After successful failover, the current filer index was not updated.
This meant every subsequent request would still try the (potentially
unhealthy) original filer first, then failover again.

**Solution:**

1. Added FilerClient.SetCurrentFiler(addr) method:
   - Finds the index of specified filer address
   - Atomically updates filerIndex to point to it
   - Thread-safe with RLock

2. Call SetCurrentFiler after successful failover:
   - Update happens immediately after successful connection
   - Future requests start with the known-healthy filer
   - Reduces unnecessary failover attempts

**Benefits:**
 Subsequent requests use healthy filer directly
 No repeated failover to same unhealthy filer
 Better performance - fast path hits healthy filer
 Comment now matches actual behavior

* Integrate health tracking with S3 failover

Address code review suggestion to leverage existing health tracking
instead of simple iteration through all filers.

**Changes:**

1. Added address-based health tracking API to FilerClient:
   - ShouldSkipUnhealthyFiler(addr) - check circuit breaker
   - RecordFilerSuccess(addr) - reset failure count
   - RecordFilerFailure(addr) - increment failure count

   These methods find the filer by address and delegate to
   existing *WithHealth methods for actual health management.

2. Updated withFilerClientFailover to use health tracking:
   - Record success/failure for every filer attempt
   - Skip unhealthy filers during failover (circuit breaker)
   - Only try filers that haven't exceeded failure threshold
   - Automatic re-check after reset timeout

**Benefits:**

 Circuit breaker prevents wasting time on known-bad filers
 Health tracking shared across all operations
 Automatic recovery when unhealthy filers come back
 Reduced latency - skip filers in failure state
 Better visibility with health metrics

**Behavior:**

- Try current filer first (fast path)
- If fails, record failure and try other HEALTHY filers
- Skip filers with failureCount >= threshold (default 3)
- Re-check unhealthy filers after resetTimeout (default 30s)
- Record all successes/failures for health tracking

* Update weed/wdclient/filer_client.go

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* Enable filer discovery with empty filerGroup

Empty filerGroup is a valid value representing the default group.
The master client can discover filers even when filerGroup is empty.

**Change:**
- Remove the filerGroup != "" check in NewFilerClient
- Keep only masterClient != nil check
- Empty string will be passed to ListClusterNodes API as-is

This enables filer discovery to work with the default group.

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
2025-11-26 11:29:55 -08:00

1807 lines
74 KiB
Go

package s3api
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/pquerna/cachecontrol/cacheobject"
"github.com/seaweedfs/seaweedfs/weed/filer"
"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/pb/s3_pb"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
"github.com/seaweedfs/seaweedfs/weed/security"
weed_server "github.com/seaweedfs/seaweedfs/weed/server"
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/util/constants"
)
// Object lock validation errors
var (
ErrObjectLockVersioningRequired = errors.New("object lock headers can only be used on versioned buckets")
ErrInvalidObjectLockMode = errors.New("invalid object lock mode")
ErrInvalidLegalHoldStatus = errors.New("invalid legal hold status")
ErrInvalidRetentionDateFormat = errors.New("invalid retention until date format")
ErrRetentionDateMustBeFuture = errors.New("retain until date must be in the future")
ErrObjectLockModeRequiresDate = errors.New("object lock mode requires retention until date")
ErrRetentionDateRequiresMode = errors.New("retention until date requires object lock mode")
ErrGovernanceBypassVersioningRequired = errors.New("governance bypass header can only be used on versioned buckets")
ErrInvalidObjectLockDuration = errors.New("object lock duration must be greater than 0 days")
ErrObjectLockDurationExceeded = errors.New("object lock duration exceeds maximum allowed days")
ErrObjectLockConfigurationMissingEnabled = errors.New("object lock configuration must specify ObjectLockEnabled")
ErrInvalidObjectLockEnabledValue = errors.New("invalid object lock enabled value")
ErrRuleMissingDefaultRetention = errors.New("rule configuration must specify DefaultRetention")
ErrDefaultRetentionMissingMode = errors.New("default retention must specify Mode")
ErrInvalidDefaultRetentionMode = errors.New("invalid default retention mode")
ErrDefaultRetentionMissingPeriod = errors.New("default retention must specify either Days or Years")
ErrDefaultRetentionBothDaysAndYears = errors.New("default retention cannot specify both Days and Years")
ErrDefaultRetentionDaysOutOfRange = errors.New("default retention days must be between 0 and 36500")
ErrDefaultRetentionYearsOutOfRange = errors.New("default retention years must be between 0 and 100")
)
// hasExplicitEncryption checks if any explicit encryption was provided in the request.
// This helper improves readability and makes the encryption check condition more explicit.
func hasExplicitEncryption(customerKey *SSECustomerKey, sseKMSKey *SSEKMSKey, sseS3Key *SSES3Key) bool {
return customerKey != nil || sseKMSKey != nil || sseS3Key != nil
}
// BucketDefaultEncryptionResult holds the result of bucket default encryption processing
type BucketDefaultEncryptionResult struct {
DataReader io.Reader
SSES3Key *SSES3Key
SSEKMSKey *SSEKMSKey
}
// SSEResponseMetadata holds encryption metadata needed for HTTP response headers
type SSEResponseMetadata struct {
SSEType string
KMSKeyID string
BucketKeyEnabled bool
}
func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
// http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
bucket, object := s3_constants.GetBucketAndObject(r)
glog.V(3).Infof("PutObjectHandler %s %s", bucket, object)
_, err := validateContentMd5(r.Header)
if err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidDigest)
return
}
// Check conditional headers
if errCode := s3a.checkConditionalHeaders(r, bucket, object); errCode != s3err.ErrNone {
s3err.WriteErrorResponse(w, r, errCode)
return
}
if r.Header.Get("Cache-Control") != "" {
if _, err = cacheobject.ParseRequestCacheControl(r.Header.Get("Cache-Control")); err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidDigest)
return
}
}
if r.Header.Get("Expires") != "" {
if _, err = time.Parse(http.TimeFormat, r.Header.Get("Expires")); err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrMalformedDate)
return
}
}
dataReader, s3ErrCode := getRequestDataReader(s3a, r)
if s3ErrCode != s3err.ErrNone {
s3err.WriteErrorResponse(w, r, s3ErrCode)
return
}
defer dataReader.Close()
objectContentType := r.Header.Get("Content-Type")
if strings.HasSuffix(object, "/") && r.ContentLength <= 1024 {
if err := s3a.mkdir(
s3a.option.BucketsPath, bucket+strings.TrimSuffix(object, "/"),
func(entry *filer_pb.Entry) {
if objectContentType == "" {
objectContentType = s3_constants.FolderMimeType
}
if r.ContentLength > 0 {
entry.Content, _ = io.ReadAll(r.Body)
}
entry.Attributes.Mime = objectContentType
// Set object owner for directory objects (same as regular objects)
s3a.setObjectOwnerFromRequest(r, entry)
}); err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
return
}
} else {
// Get detailed versioning state for the bucket
versioningState, err := s3a.getVersioningState(bucket)
if err != nil {
if errors.Is(err, filer_pb.ErrNotFound) {
// Auto-create bucket if it doesn't exist (requires Admin permission)
if !s3a.handleAutoCreateBucket(w, r, bucket, "PutObjectHandler") {
return
}
// Re-fetch versioning state to handle race conditions where
// another process might have created the bucket with versioning enabled.
versioningState, err = s3a.getVersioningState(bucket)
if err != nil {
glog.Errorf("Error re-checking versioning status for bucket %s after auto-creation: %v", bucket, err)
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
return
}
} else {
glog.Errorf("Error checking versioning status for bucket %s: %v", bucket, err)
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
return
}
}
versioningEnabled := (versioningState == s3_constants.VersioningEnabled)
versioningConfigured := (versioningState != "")
glog.V(3).Infof("PutObjectHandler: bucket=%s, object=%s, versioningState='%s', versioningEnabled=%v, versioningConfigured=%v", bucket, object, versioningState, versioningEnabled, versioningConfigured)
// Validate object lock headers before processing
if err := s3a.validateObjectLockHeaders(r, versioningEnabled); err != nil {
glog.V(2).Infof("PutObjectHandler: object lock header validation failed for bucket %s, object %s: %v", bucket, object, err)
s3err.WriteErrorResponse(w, r, mapValidationErrorToS3Error(err))
return
}
// For non-versioned buckets, check if existing object has object lock protections
// that would prevent overwrite (PUT operations overwrite existing objects in non-versioned buckets)
if !versioningConfigured {
governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil {
glog.V(2).Infof("PutObjectHandler: object lock permissions check failed for %s/%s: %v", bucket, object, err)
s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
return
}
}
switch versioningState {
case s3_constants.VersioningEnabled:
// Handle enabled versioning - create new versions with real version IDs
glog.V(3).Infof("PutObjectHandler: ENABLED versioning detected for %s/%s, calling putVersionedObject", bucket, object)
versionId, etag, errCode, sseMetadata := s3a.putVersionedObject(r, bucket, object, dataReader, objectContentType)
if errCode != s3err.ErrNone {
glog.Errorf("PutObjectHandler: putVersionedObject failed with errCode=%v for %s/%s", errCode, bucket, object)
s3err.WriteErrorResponse(w, r, errCode)
return
}
glog.V(3).Infof("PutObjectHandler: putVersionedObject returned versionId=%s, etag=%s for %s/%s", versionId, etag, bucket, object)
// Set version ID in response header
if versionId != "" {
w.Header().Set("x-amz-version-id", versionId)
glog.V(3).Infof("PutObjectHandler: set x-amz-version-id header to %s for %s/%s", versionId, bucket, object)
} else {
glog.Errorf("PutObjectHandler: CRITICAL - versionId is EMPTY for versioned bucket %s, object %s", bucket, object)
}
// Set ETag in response
setEtag(w, etag)
// Set SSE response headers for versioned objects
s3a.setSSEResponseHeaders(w, r, sseMetadata)
case s3_constants.VersioningSuspended:
// Handle suspended versioning - overwrite with "null" version ID but preserve existing versions
glog.V(3).Infof("PutObjectHandler: SUSPENDED versioning detected for %s/%s, calling putSuspendedVersioningObject", bucket, object)
etag, errCode, sseMetadata := s3a.putSuspendedVersioningObject(r, bucket, object, dataReader, objectContentType)
if errCode != s3err.ErrNone {
s3err.WriteErrorResponse(w, r, errCode)
return
}
// Note: Suspended versioning should NOT return x-amz-version-id header per AWS S3 spec
// The object is stored with "null" version internally but no version header is returned
// Set ETag in response
setEtag(w, etag)
// Set SSE response headers for suspended versioning
s3a.setSSEResponseHeaders(w, r, sseMetadata)
default:
// Handle regular PUT (never configured versioning)
filePath := s3a.toFilerPath(bucket, object)
if objectContentType == "" {
dataReader = mimeDetect(r, dataReader)
}
etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, 1)
if errCode != s3err.ErrNone {
s3err.WriteErrorResponse(w, r, errCode)
return
}
// No version ID header for never-configured versioning
setEtag(w, etag)
// Set SSE response headers based on encryption type used
s3a.setSSEResponseHeaders(w, r, sseMetadata)
}
}
stats_collect.RecordBucketActiveTime(bucket)
stats_collect.S3UploadedObjectsCounter.WithLabelValues(bucket).Inc()
writeSuccessResponseEmpty(w, r)
}
func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, partNumber int) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
// NEW OPTIMIZATION: Write directly to volume servers, bypassing filer proxy
// This eliminates the filer proxy overhead for PUT operations
// Note: filePath is now passed directly instead of URL (no parsing needed)
// For SSE, encrypt with offset=0 for all parts
// Each part is encrypted independently, then decrypted using metadata during GET
partOffset := int64(0)
// Handle all SSE encryption types in a unified manner
sseResult, sseErrorCode := s3a.handleAllSSEEncryption(r, dataReader, partOffset)
if sseErrorCode != s3err.ErrNone {
return "", sseErrorCode, SSEResponseMetadata{}
}
// Extract results from unified SSE handling
dataReader = sseResult.DataReader
customerKey := sseResult.CustomerKey
sseIV := sseResult.SSEIV
sseKMSKey := sseResult.SSEKMSKey
sseKMSMetadata := sseResult.SSEKMSMetadata
sseS3Key := sseResult.SSES3Key
sseS3Metadata := sseResult.SSES3Metadata
sseType := sseResult.SSEType
// Apply bucket default encryption if no explicit encryption was provided
// This implements AWS S3 behavior where bucket default encryption automatically applies
if !hasExplicitEncryption(customerKey, sseKMSKey, sseS3Key) {
glog.V(4).Infof("putToFiler: no explicit encryption detected, checking for bucket default encryption")
// Apply bucket default encryption and get the result
encryptionResult, applyErr := s3a.applyBucketDefaultEncryption(bucket, r, dataReader)
if applyErr != nil {
glog.Errorf("Failed to apply bucket default encryption: %v", applyErr)
return "", s3err.ErrInternalError, SSEResponseMetadata{}
}
// Update variables based on the result
dataReader = encryptionResult.DataReader
sseS3Key = encryptionResult.SSES3Key
sseKMSKey = encryptionResult.SSEKMSKey
// If bucket-default encryption selected an algorithm, reflect it in SSE type
if sseType == "" {
if sseS3Key != nil {
sseType = s3_constants.SSETypeS3
} else if sseKMSKey != nil {
sseType = s3_constants.SSETypeKMS
}
}
// If SSE-S3 was applied by bucket default, prepare metadata (if not already done)
if sseS3Key != nil && len(sseS3Metadata) == 0 {
var metaErr error
sseS3Metadata, metaErr = SerializeSSES3Metadata(sseS3Key)
if metaErr != nil {
glog.Errorf("Failed to serialize SSE-S3 metadata for bucket default encryption: %v", metaErr)
return "", s3err.ErrInternalError, SSEResponseMetadata{}
}
}
} else {
glog.V(4).Infof("putToFiler: explicit encryption already applied, skipping bucket default encryption")
}
// filePath is already provided directly - no URL parsing needed
// Step 1 & 2: Use auto-chunking to handle large files without OOM
// This splits large uploads into 8MB chunks, preventing memory issues on both S3 API and volume servers
const chunkSize = 8 * 1024 * 1024 // 8MB chunks (S3 standard)
const smallFileLimit = 256 * 1024 // 256KB - store inline in filer
collection := ""
if s3a.option.FilerGroup != "" {
collection = s3a.getCollectionName(bucket)
}
// Create assign function for chunked upload
assignFunc := func(ctx context.Context, count int) (*operation.VolumeAssignRequest, *operation.AssignResult, error) {
var assignResult *filer_pb.AssignVolumeResponse
err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.AssignVolume(ctx, &filer_pb.AssignVolumeRequest{
Count: int32(count),
Replication: "",
Collection: collection,
DiskType: "",
DataCenter: s3a.option.DataCenter,
Path: filePath,
})
if err != nil {
return fmt.Errorf("assign volume: %w", err)
}
if resp.Error != "" {
return fmt.Errorf("assign volume: %v", resp.Error)
}
assignResult = resp
return nil
})
if err != nil {
return nil, nil, err
}
// Convert filer_pb.AssignVolumeResponse to operation.AssignResult
return nil, &operation.AssignResult{
Fid: assignResult.FileId,
Url: assignResult.Location.Url,
PublicUrl: assignResult.Location.PublicUrl,
Count: uint64(count),
Auth: security.EncodedJwt(assignResult.Auth),
}, nil
}
// Upload with auto-chunking
// Use context.Background() to ensure chunk uploads complete even if HTTP request is cancelled
// This prevents partial uploads and data corruption
chunkResult, err := operation.UploadReaderInChunks(context.Background(), dataReader, &operation.ChunkedUploadOption{
ChunkSize: chunkSize,
SmallFileLimit: smallFileLimit,
Collection: collection,
DataCenter: s3a.option.DataCenter,
SaveSmallInline: false, // S3 API always creates chunks, never stores inline
MimeType: r.Header.Get("Content-Type"),
AssignFunc: assignFunc,
})
if err != nil {
glog.Errorf("putToFiler: chunked upload failed: %v", err)
// CRITICAL: Cleanup orphaned chunks before returning error
// UploadReaderInChunks now returns partial results even on error,
// allowing us to cleanup any chunks that were successfully uploaded
// before the failure occurred
if chunkResult != nil && len(chunkResult.FileChunks) > 0 {
glog.Warningf("putToFiler: Upload failed, attempting to cleanup %d orphaned chunks", len(chunkResult.FileChunks))
s3a.deleteOrphanedChunks(chunkResult.FileChunks)
}
if strings.Contains(err.Error(), s3err.ErrMsgPayloadChecksumMismatch) {
return "", s3err.ErrInvalidDigest, SSEResponseMetadata{}
}
return "", s3err.ErrInternalError, SSEResponseMetadata{}
}
// Step 3: Calculate MD5 hash and add SSE metadata to chunks
md5Sum := chunkResult.Md5Hash.Sum(nil)
glog.V(4).Infof("putToFiler: Chunked upload SUCCESS - path=%s, chunks=%d, size=%d",
filePath, len(chunkResult.FileChunks), chunkResult.TotalSize)
// Log chunk details for debugging (verbose only - high frequency)
if glog.V(4) {
for i, chunk := range chunkResult.FileChunks {
glog.Infof(" PUT Chunk[%d]: fid=%s, offset=%d, size=%d", i, chunk.GetFileIdString(), chunk.Offset, chunk.Size)
}
}
// Add SSE metadata to all chunks if present
for _, chunk := range chunkResult.FileChunks {
switch {
case customerKey != nil:
// SSE-C: Create per-chunk metadata (matches filer logic)
chunk.SseType = filer_pb.SSEType_SSE_C
if len(sseIV) > 0 {
// PartOffset tracks position within the encrypted stream
// Since ALL uploads (single-part and multipart parts) encrypt starting from offset 0,
// PartOffset = chunk.Offset represents where this chunk is in that encrypted stream
// - Single-part: chunk.Offset is position in the file's encrypted stream
// - Multipart: chunk.Offset is position in this part's encrypted stream
ssecMetadataStruct := struct {
Algorithm string `json:"algorithm"`
IV string `json:"iv"`
KeyMD5 string `json:"keyMD5"`
PartOffset int64 `json:"partOffset"`
}{
Algorithm: "AES256",
IV: base64.StdEncoding.EncodeToString(sseIV),
KeyMD5: customerKey.KeyMD5,
PartOffset: chunk.Offset, // Position within the encrypted stream (always encrypted from 0)
}
if ssecMetadata, serErr := json.Marshal(ssecMetadataStruct); serErr == nil {
chunk.SseMetadata = ssecMetadata
}
}
case sseKMSKey != nil:
// SSE-KMS: Create per-chunk metadata with chunk-specific offsets
// Each chunk needs its own metadata with ChunkOffset set for proper IV calculation during decryption
chunk.SseType = filer_pb.SSEType_SSE_KMS
// Create a copy of the SSE-KMS key with chunk-specific offset
chunkSSEKey := &SSEKMSKey{
KeyID: sseKMSKey.KeyID,
EncryptedDataKey: sseKMSKey.EncryptedDataKey,
EncryptionContext: sseKMSKey.EncryptionContext,
BucketKeyEnabled: sseKMSKey.BucketKeyEnabled,
IV: sseKMSKey.IV,
ChunkOffset: chunk.Offset, // Set chunk-specific offset for IV calculation
}
// Serialize per-chunk metadata
if chunkMetadata, serErr := SerializeSSEKMSMetadata(chunkSSEKey); serErr == nil {
chunk.SseMetadata = chunkMetadata
} else {
glog.Errorf("Failed to serialize SSE-KMS metadata for chunk at offset %d: %v", chunk.Offset, serErr)
}
case sseS3Key != nil:
// SSE-S3: Create per-chunk metadata with chunk-specific IVs
// Each chunk needs its own IV calculated from the base IV + chunk offset
chunk.SseType = filer_pb.SSEType_SSE_S3
// Calculate chunk-specific IV using base IV and chunk offset
chunkIV, _ := calculateIVWithOffset(sseS3Key.IV, chunk.Offset)
// Create a copy of the SSE-S3 key with chunk-specific IV
chunkSSEKey := &SSES3Key{
Key: sseS3Key.Key,
KeyID: sseS3Key.KeyID,
Algorithm: sseS3Key.Algorithm,
IV: chunkIV, // Use chunk-specific IV
}
// Serialize per-chunk metadata
if chunkMetadata, serErr := SerializeSSES3Metadata(chunkSSEKey); serErr == nil {
chunk.SseMetadata = chunkMetadata
} else {
glog.Errorf("Failed to serialize SSE-S3 metadata for chunk at offset %d: %v", chunk.Offset, serErr)
}
}
}
// Step 4: Create metadata entry
now := time.Now()
mimeType := r.Header.Get("Content-Type")
if mimeType == "" {
mimeType = "application/octet-stream"
}
// Create entry
entry := &filer_pb.Entry{
Name: filepath.Base(filePath),
IsDirectory: false,
Attributes: &filer_pb.FuseAttributes{
Crtime: now.Unix(),
Mtime: now.Unix(),
FileMode: 0660,
Uid: 0,
Gid: 0,
Mime: mimeType,
FileSize: uint64(chunkResult.TotalSize),
},
Chunks: chunkResult.FileChunks, // All chunks from auto-chunking
Extended: make(map[string][]byte),
}
// Set Md5 attribute based on context:
// 1. For multipart upload PARTS (stored in .uploads/ directory): ALWAYS set Md5
// - Parts must use simple MD5 ETags, never composite format
// - Even if a part has multiple chunks internally, its ETag is MD5 of entire part
// 2. For regular object uploads: only set Md5 for single-chunk uploads
// - Multi-chunk regular objects use composite "md5-count" format
isMultipartPart := strings.Contains(filePath, "/"+s3_constants.MultipartUploadsFolder+"/")
if isMultipartPart || len(chunkResult.FileChunks) == 1 {
entry.Attributes.Md5 = md5Sum
}
// Calculate ETag using the same logic as GET to ensure consistency
// For single chunk: uses entry.Attributes.Md5
// For multiple chunks: uses filer.ETagChunks() which returns "<hash>-<count>"
etag = filer.ETag(entry)
glog.V(4).Infof("putToFiler: Calculated ETag=%s for %d chunks", etag, len(chunkResult.FileChunks))
// Set object owner
amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
if amzAccountId != "" {
entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
glog.V(2).Infof("putToFiler: setting owner %s for object %s", amzAccountId, filePath)
}
// Set version ID if present
if versionIdHeader := r.Header.Get(s3_constants.ExtVersionIdKey); versionIdHeader != "" {
entry.Extended[s3_constants.ExtVersionIdKey] = []byte(versionIdHeader)
glog.V(3).Infof("putToFiler: setting version ID %s for object %s", versionIdHeader, filePath)
}
// Set TTL-based S3 expiry flag only if object has a TTL
if entry.Attributes.TtlSec > 0 {
entry.Extended[s3_constants.SeaweedFSExpiresS3] = []byte("true")
}
// Copy user metadata and standard headers
for k, v := range r.Header {
if len(v) > 0 && len(v[0]) > 0 {
if strings.HasPrefix(k, s3_constants.AmzUserMetaPrefix) {
// Go's HTTP server canonicalizes headers (e.g., x-amz-meta-foo → X-Amz-Meta-Foo)
// We store them as they come in (after canonicalization) to preserve the user's intent
entry.Extended[k] = []byte(v[0])
} else if k == "Cache-Control" || k == "Expires" || k == "Content-Disposition" {
entry.Extended[k] = []byte(v[0])
}
if k == "Response-Content-Disposition" {
entry.Extended["Content-Disposition"] = []byte(v[0])
}
}
}
// Set SSE-C metadata
if customerKey != nil && len(sseIV) > 0 {
// Store IV as RAW bytes (matches filer behavior - filer decodes base64 headers and stores raw bytes)
entry.Extended[s3_constants.SeaweedFSSSEIV] = sseIV
entry.Extended[s3_constants.AmzServerSideEncryptionCustomerAlgorithm] = []byte("AES256")
entry.Extended[s3_constants.AmzServerSideEncryptionCustomerKeyMD5] = []byte(customerKey.KeyMD5)
glog.V(3).Infof("putToFiler: storing SSE-C metadata - IV len=%d", len(sseIV))
}
// Set SSE-KMS metadata
if sseKMSKey != nil {
// Store metadata as RAW bytes (matches filer behavior - filer decodes base64 headers and stores raw bytes)
entry.Extended[s3_constants.SeaweedFSSSEKMSKey] = sseKMSMetadata
// Set standard SSE headers for detection
entry.Extended[s3_constants.AmzServerSideEncryption] = []byte("aws:kms")
entry.Extended[s3_constants.AmzServerSideEncryptionAwsKmsKeyId] = []byte(sseKMSKey.KeyID)
glog.V(3).Infof("putToFiler: storing SSE-KMS metadata - keyID=%s, raw len=%d", sseKMSKey.KeyID, len(sseKMSMetadata))
}
// Set SSE-S3 metadata
if sseS3Key != nil && len(sseS3Metadata) > 0 {
// Store metadata as RAW bytes (matches filer behavior - filer decodes base64 headers and stores raw bytes)
entry.Extended[s3_constants.SeaweedFSSSES3Key] = sseS3Metadata
// Set standard SSE header for detection
entry.Extended[s3_constants.AmzServerSideEncryption] = []byte("AES256")
glog.V(3).Infof("putToFiler: storing SSE-S3 metadata - keyID=%s, raw len=%d", sseS3Key.KeyID, len(sseS3Metadata))
}
// Step 4: Save metadata to filer via gRPC
// Use context.Background() to ensure metadata save completes even if HTTP request is cancelled
// This matches the chunk upload behavior and prevents orphaned chunks
glog.V(3).Infof("putToFiler: About to create entry - dir=%s, name=%s, chunks=%d, extended keys=%d",
filepath.Dir(filePath), filepath.Base(filePath), len(entry.Chunks), len(entry.Extended))
createErr := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
req := &filer_pb.CreateEntryRequest{
Directory: filepath.Dir(filePath),
Entry: entry,
}
glog.V(3).Infof("putToFiler: Calling CreateEntry for %s", filePath)
_, err := client.CreateEntry(context.Background(), req)
if err != nil {
glog.Errorf("putToFiler: CreateEntry returned error: %v", err)
}
return err
})
if createErr != nil {
glog.Errorf("putToFiler: failed to create entry for %s: %v", filePath, createErr)
// CRITICAL: Cleanup orphaned chunks before returning error
// If CreateEntry fails, the uploaded chunks are orphaned and must be deleted
// to prevent resource leaks and wasted storage
if len(chunkResult.FileChunks) > 0 {
glog.Warningf("putToFiler: CreateEntry failed, attempting to cleanup %d orphaned chunks", len(chunkResult.FileChunks))
s3a.deleteOrphanedChunks(chunkResult.FileChunks)
}
return "", filerErrorToS3Error(createErr), SSEResponseMetadata{}
}
glog.V(3).Infof("putToFiler: CreateEntry SUCCESS for %s", filePath)
glog.V(2).Infof("putToFiler: Metadata saved SUCCESS - path=%s, etag(hex)=%s, size=%d, partNumber=%d",
filePath, etag, entry.Attributes.FileSize, partNumber)
BucketTrafficReceived(chunkResult.TotalSize, r)
// Build SSE response metadata with encryption details
responseMetadata := SSEResponseMetadata{
SSEType: sseType,
}
// For SSE-KMS, include key ID and bucket-key-enabled flag from stored metadata
if sseKMSKey != nil {
responseMetadata.KMSKeyID = sseKMSKey.KeyID
responseMetadata.BucketKeyEnabled = sseKMSKey.BucketKeyEnabled
glog.V(4).Infof("putToFiler: returning SSE-KMS metadata - keyID=%s, bucketKeyEnabled=%v",
sseKMSKey.KeyID, sseKMSKey.BucketKeyEnabled)
}
return etag, s3err.ErrNone, responseMetadata
}
func setEtag(w http.ResponseWriter, etag string) {
if etag != "" {
if strings.HasPrefix(etag, "\"") {
w.Header()["ETag"] = []string{etag}
} else {
w.Header()["ETag"] = []string{"\"" + etag + "\""}
}
}
}
// setSSEResponseHeaders sets appropriate SSE response headers based on encryption type
func (s3a *S3ApiServer) setSSEResponseHeaders(w http.ResponseWriter, r *http.Request, sseMetadata SSEResponseMetadata) {
switch sseMetadata.SSEType {
case s3_constants.SSETypeS3:
// SSE-S3: Return the encryption algorithm
w.Header().Set(s3_constants.AmzServerSideEncryption, s3_constants.SSEAlgorithmAES256)
case s3_constants.SSETypeC:
// SSE-C: Echo back the customer-provided algorithm and key MD5
if algo := r.Header.Get(s3_constants.AmzServerSideEncryptionCustomerAlgorithm); algo != "" {
w.Header().Set(s3_constants.AmzServerSideEncryptionCustomerAlgorithm, algo)
}
if keyMD5 := r.Header.Get(s3_constants.AmzServerSideEncryptionCustomerKeyMD5); keyMD5 != "" {
w.Header().Set(s3_constants.AmzServerSideEncryptionCustomerKeyMD5, keyMD5)
}
case s3_constants.SSETypeKMS:
// SSE-KMS: Return the KMS key ID and algorithm
w.Header().Set(s3_constants.AmzServerSideEncryption, "aws:kms")
// Use metadata from stored encryption config (for bucket-default encryption)
// or fall back to request headers (for explicit encryption)
if sseMetadata.KMSKeyID != "" {
w.Header().Set(s3_constants.AmzServerSideEncryptionAwsKmsKeyId, sseMetadata.KMSKeyID)
} else if keyID := r.Header.Get(s3_constants.AmzServerSideEncryptionAwsKmsKeyId); keyID != "" {
w.Header().Set(s3_constants.AmzServerSideEncryptionAwsKmsKeyId, keyID)
}
// Set bucket-key-enabled header if it was enabled
if sseMetadata.BucketKeyEnabled {
w.Header().Set(s3_constants.AmzServerSideEncryptionBucketKeyEnabled, "true")
} else if bucketKeyEnabled := r.Header.Get(s3_constants.AmzServerSideEncryptionBucketKeyEnabled); bucketKeyEnabled == "true" {
w.Header().Set(s3_constants.AmzServerSideEncryptionBucketKeyEnabled, "true")
}
}
}
func filerErrorToS3Error(err error) s3err.ErrorCode {
if err == nil {
return s3err.ErrNone
}
errString := err.Error()
switch {
case errString == constants.ErrMsgBadDigest:
return s3err.ErrBadDigest
case errors.Is(err, weed_server.ErrReadOnly):
// Bucket is read-only due to quota enforcement or other configuration
// Return 403 Forbidden per S3 semantics (similar to MinIO's quota enforcement)
// Uses errors.Is() to properly detect wrapped errors
return s3err.ErrAccessDenied
case strings.Contains(errString, "context canceled") || strings.Contains(errString, "code = Canceled"):
// Client canceled the request, return client error not server error
return s3err.ErrInvalidRequest
case strings.HasPrefix(errString, "existing ") && strings.HasSuffix(errString, "is a directory"):
return s3err.ErrExistingObjectIsDirectory
case strings.HasSuffix(errString, "is a file"):
return s3err.ErrExistingObjectIsFile
default:
return s3err.ErrInternalError
}
}
// setObjectOwnerFromRequest sets the object owner metadata based on the authenticated user
func (s3a *S3ApiServer) setObjectOwnerFromRequest(r *http.Request, entry *filer_pb.Entry) {
amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
if amzAccountId != "" {
if entry.Extended == nil {
entry.Extended = make(map[string][]byte)
}
entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
glog.V(2).Infof("setObjectOwnerFromRequest: set object owner to %s", amzAccountId)
}
}
// putSuspendedVersioningObject handles PUT operations for buckets with suspended versioning.
//
// Key architectural approach:
// Instead of creating the file and then updating its metadata (which can cause race conditions and duplicate versions),
// we set all required metadata as HTTP headers BEFORE calling putToFiler. The filer automatically stores any header
// starting with "Seaweed-" in entry.Extended during file creation, ensuring atomic metadata persistence.
//
// This approach eliminates:
// - Race conditions from read-after-write consistency delays
// - Need for retry loops and exponential backoff
// - Duplicate entries from separate create/update operations
//
// For suspended versioning, objects are stored as regular files (version ID "null") in the bucket directory,
// while existing versions from when versioning was enabled remain preserved in the .versions subdirectory.
func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, object string, dataReader io.Reader, objectContentType string) (etag string, errCode s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
// Normalize object path to ensure consistency with toFilerPath behavior
normalizedObject := removeDuplicateSlashes(object)
glog.V(3).Infof("putSuspendedVersioningObject: START bucket=%s, object=%s, normalized=%s",
bucket, object, normalizedObject)
bucketDir := s3a.option.BucketsPath + "/" + bucket
// Check if there's an existing null version in .versions directory and delete it
// This ensures suspended versioning properly overwrites the null version as per S3 spec
// Note: We only delete null versions, NOT regular versions (those should be preserved)
versionsObjectPath := normalizedObject + s3_constants.VersionsFolder
versionsDir := bucketDir + "/" + versionsObjectPath
entries, _, err := s3a.list(versionsDir, "", "", false, 1000)
if err == nil {
// .versions directory exists
glog.V(3).Infof("putSuspendedVersioningObject: found %d entries in .versions for %s/%s", len(entries), bucket, object)
for _, entry := range entries {
if entry.Extended != nil {
if versionIdBytes, ok := entry.Extended[s3_constants.ExtVersionIdKey]; ok {
versionId := string(versionIdBytes)
glog.V(3).Infof("putSuspendedVersioningObject: found version '%s' in .versions", versionId)
if versionId == "null" {
// Only delete null version - preserve real versioned entries
glog.V(3).Infof("putSuspendedVersioningObject: deleting null version from .versions")
err := s3a.rm(versionsDir, entry.Name, true, false)
if err != nil {
glog.Warningf("putSuspendedVersioningObject: failed to delete null version: %v", err)
} else {
glog.V(3).Infof("putSuspendedVersioningObject: successfully deleted null version")
}
break
}
}
}
}
} else {
glog.V(3).Infof("putSuspendedVersioningObject: no .versions directory for %s/%s", bucket, object)
}
filePath := s3a.toFilerPath(bucket, normalizedObject)
body := dataReader
if objectContentType == "" {
body = mimeDetect(r, body)
}
// Set all metadata headers BEFORE calling putToFiler
// This ensures the metadata is set during file creation, not after
// The filer automatically stores any header starting with "Seaweed-" in entry.Extended
// Set version ID to "null" for suspended versioning
r.Header.Set(s3_constants.ExtVersionIdKey, "null")
// Extract and set object lock metadata as headers
// This handles retention mode, retention date, and legal hold
explicitMode := r.Header.Get(s3_constants.AmzObjectLockMode)
explicitRetainUntilDate := r.Header.Get(s3_constants.AmzObjectLockRetainUntilDate)
if explicitMode != "" {
r.Header.Set(s3_constants.ExtObjectLockModeKey, explicitMode)
glog.V(2).Infof("putSuspendedVersioningObject: setting object lock mode header: %s", explicitMode)
}
if explicitRetainUntilDate != "" {
// Parse and convert to Unix timestamp
parsedTime, err := time.Parse(time.RFC3339, explicitRetainUntilDate)
if err != nil {
glog.Errorf("putSuspendedVersioningObject: failed to parse retention until date: %v", err)
return "", s3err.ErrInvalidRequest, SSEResponseMetadata{}
}
r.Header.Set(s3_constants.ExtRetentionUntilDateKey, strconv.FormatInt(parsedTime.Unix(), 10))
glog.V(2).Infof("putSuspendedVersioningObject: setting retention until date header (timestamp: %d)", parsedTime.Unix())
}
if legalHold := r.Header.Get(s3_constants.AmzObjectLockLegalHold); legalHold != "" {
if legalHold == s3_constants.LegalHoldOn || legalHold == s3_constants.LegalHoldOff {
r.Header.Set(s3_constants.ExtLegalHoldKey, legalHold)
glog.V(2).Infof("putSuspendedVersioningObject: setting legal hold header: %s", legalHold)
} else {
glog.Errorf("putSuspendedVersioningObject: invalid legal hold value: %s", legalHold)
return "", s3err.ErrInvalidRequest, SSEResponseMetadata{}
}
}
// Apply bucket default retention if no explicit retention was provided
if explicitMode == "" && explicitRetainUntilDate == "" {
// Create a temporary entry to apply defaults
tempEntry := &filer_pb.Entry{Extended: make(map[string][]byte)}
if err := s3a.applyBucketDefaultRetention(bucket, tempEntry); err == nil {
// Copy default retention headers from temp entry
if modeBytes, ok := tempEntry.Extended[s3_constants.ExtObjectLockModeKey]; ok {
r.Header.Set(s3_constants.ExtObjectLockModeKey, string(modeBytes))
glog.V(2).Infof("putSuspendedVersioningObject: applied bucket default retention mode: %s", string(modeBytes))
}
if dateBytes, ok := tempEntry.Extended[s3_constants.ExtRetentionUntilDateKey]; ok {
r.Header.Set(s3_constants.ExtRetentionUntilDateKey, string(dateBytes))
glog.V(2).Infof("putSuspendedVersioningObject: applied bucket default retention date")
}
}
}
// Upload the file using putToFiler - this will create the file with version metadata
etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, 1)
if errCode != s3err.ErrNone {
glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
return "", errCode, SSEResponseMetadata{}
}
// Update all existing versions/delete markers to set IsLatest=false since "null" is now latest
err = s3a.updateIsLatestFlagsForSuspendedVersioning(bucket, normalizedObject)
if err != nil {
glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err)
// Don't fail the request, but log the warning
}
glog.V(2).Infof("putSuspendedVersioningObject: successfully created null version for %s/%s", bucket, object)
return etag, s3err.ErrNone, sseMetadata
}
// updateIsLatestFlagsForSuspendedVersioning sets IsLatest=false on all existing versions/delete markers
// when a new "null" version becomes the latest during suspended versioning
func (s3a *S3ApiServer) updateIsLatestFlagsForSuspendedVersioning(bucket, object string) error {
bucketDir := s3a.option.BucketsPath + "/" + bucket
versionsObjectPath := object + s3_constants.VersionsFolder
versionsDir := bucketDir + "/" + versionsObjectPath
glog.V(2).Infof("updateIsLatestFlagsForSuspendedVersioning: updating flags for %s%s", bucket, object)
// Check if .versions directory exists
_, err := s3a.getEntry(bucketDir, versionsObjectPath)
if err != nil {
// No .versions directory exists, nothing to update
glog.V(2).Infof("updateIsLatestFlagsForSuspendedVersioning: no .versions directory for %s%s", bucket, object)
return nil
}
// List all entries in .versions directory
entries, _, err := s3a.list(versionsDir, "", "", false, 1000)
if err != nil {
return fmt.Errorf("failed to list versions directory: %v", err)
}
glog.V(2).Infof("updateIsLatestFlagsForSuspendedVersioning: found %d entries to update", len(entries))
// Update each version/delete marker to set IsLatest=false
for _, entry := range entries {
if entry.Extended == nil {
continue
}
// Check if this entry has a version ID (it should be a version or delete marker)
versionIdBytes, hasVersionId := entry.Extended[s3_constants.ExtVersionIdKey]
if !hasVersionId {
continue
}
versionId := string(versionIdBytes)
glog.V(2).Infof("updateIsLatestFlagsForSuspendedVersioning: setting IsLatest=false for version %s", versionId)
// Update the entry to set IsLatest=false (we don't explicitly store this flag,
// it's determined by comparison with latest version metadata)
// We need to clear the latest version metadata from the .versions directory
// so that our getObjectVersionList function will correctly show IsLatest=false
}
// Clear the latest version metadata from .versions directory since "null" is now latest
versionsEntry, err := s3a.getEntry(bucketDir, versionsObjectPath)
if err == nil && versionsEntry.Extended != nil {
// Remove latest version metadata so all versions show IsLatest=false
delete(versionsEntry.Extended, s3_constants.ExtLatestVersionIdKey)
delete(versionsEntry.Extended, s3_constants.ExtLatestVersionFileNameKey)
// Update the .versions directory entry
err = s3a.mkFile(bucketDir, versionsObjectPath, versionsEntry.Chunks, func(updatedEntry *filer_pb.Entry) {
updatedEntry.Extended = versionsEntry.Extended
updatedEntry.Attributes = versionsEntry.Attributes
updatedEntry.Chunks = versionsEntry.Chunks
})
if err != nil {
return fmt.Errorf("failed to update .versions directory metadata: %v", err)
}
glog.V(2).Infof("updateIsLatestFlagsForSuspendedVersioning: cleared latest version metadata for %s%s", bucket, object)
}
return nil
}
func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object string, dataReader io.Reader, objectContentType string) (versionId string, etag string, errCode s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
// Generate version ID
versionId = generateVersionId()
// Normalize object path to ensure consistency with toFilerPath behavior
normalizedObject := removeDuplicateSlashes(object)
glog.V(2).Infof("putVersionedObject: creating version %s for %s/%s (normalized: %s)", versionId, bucket, object, normalizedObject)
// Create the version file name
versionFileName := s3a.getVersionFileName(versionId)
// Upload directly to the versions directory
// We need to construct the object path relative to the bucket
versionObjectPath := normalizedObject + s3_constants.VersionsFolder + "/" + versionFileName
versionFilePath := s3a.toFilerPath(bucket, versionObjectPath)
// Ensure the .versions directory exists before uploading
bucketDir := s3a.option.BucketsPath + "/" + bucket
versionsDir := normalizedObject + s3_constants.VersionsFolder
err := s3a.mkdir(bucketDir, versionsDir, func(entry *filer_pb.Entry) {
entry.Attributes.Mime = s3_constants.FolderMimeType
})
if err != nil {
glog.Errorf("putVersionedObject: failed to create .versions directory: %v", err)
return "", "", s3err.ErrInternalError, SSEResponseMetadata{}
}
body := dataReader
if objectContentType == "" {
body = mimeDetect(r, body)
}
glog.V(2).Infof("putVersionedObject: uploading %s/%s version %s to %s", bucket, object, versionId, versionFilePath)
etag, errCode, sseMetadata = s3a.putToFiler(r, versionFilePath, body, bucket, 1)
if errCode != s3err.ErrNone {
glog.Errorf("putVersionedObject: failed to upload version: %v", errCode)
return "", "", errCode, SSEResponseMetadata{}
}
// Get the uploaded entry to add versioning metadata
// Use retry logic to handle filer consistency delays
var versionEntry *filer_pb.Entry
maxRetries := 8
for attempt := 1; attempt <= maxRetries; attempt++ {
versionEntry, err = s3a.getEntry(bucketDir, versionObjectPath)
if err == nil {
break
}
if attempt < maxRetries {
// Exponential backoff: 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms
delay := time.Millisecond * time.Duration(10*(1<<(attempt-1)))
time.Sleep(delay)
}
}
if err != nil {
glog.Errorf("putVersionedObject: failed to get version entry after %d attempts: %v", maxRetries, err)
return "", "", s3err.ErrInternalError, SSEResponseMetadata{}
}
// Add versioning metadata to this version
if versionEntry.Extended == nil {
versionEntry.Extended = make(map[string][]byte)
}
versionEntry.Extended[s3_constants.ExtVersionIdKey] = []byte(versionId)
// Store ETag with quotes for S3 compatibility
if !strings.HasPrefix(etag, "\"") {
etag = "\"" + etag + "\""
}
versionEntry.Extended[s3_constants.ExtETagKey] = []byte(etag)
// Set object owner for versioned objects
s3a.setObjectOwnerFromRequest(r, versionEntry)
// Extract and store object lock metadata from request headers
if err := s3a.extractObjectLockMetadataFromRequest(r, versionEntry); err != nil {
glog.Errorf("putVersionedObject: failed to extract object lock metadata: %v", err)
return "", "", s3err.ErrInvalidRequest, SSEResponseMetadata{}
}
// Update the version entry with metadata
err = s3a.mkFile(bucketDir, versionObjectPath, versionEntry.Chunks, func(updatedEntry *filer_pb.Entry) {
updatedEntry.Extended = versionEntry.Extended
updatedEntry.Attributes = versionEntry.Attributes
updatedEntry.Chunks = versionEntry.Chunks
})
if err != nil {
glog.Errorf("putVersionedObject: failed to update version metadata: %v", err)
return "", "", s3err.ErrInternalError, SSEResponseMetadata{}
}
// Update the .versions directory metadata to indicate this is the latest version
err = s3a.updateLatestVersionInDirectory(bucket, normalizedObject, versionId, versionFileName)
if err != nil {
glog.Errorf("putVersionedObject: failed to update latest version in directory: %v", err)
return "", "", s3err.ErrInternalError, SSEResponseMetadata{}
}
glog.V(2).Infof("putVersionedObject: successfully created version %s for %s/%s (normalized: %s)", versionId, bucket, object, normalizedObject)
return versionId, etag, s3err.ErrNone, sseMetadata
}
// updateLatestVersionInDirectory updates the .versions directory metadata to indicate the latest version
func (s3a *S3ApiServer) updateLatestVersionInDirectory(bucket, object, versionId, versionFileName string) error {
bucketDir := s3a.option.BucketsPath + "/" + bucket
versionsObjectPath := object + s3_constants.VersionsFolder
// Get the current .versions directory entry with retry logic for filer consistency
var versionsEntry *filer_pb.Entry
var err error
maxRetries := 8
for attempt := 1; attempt <= maxRetries; attempt++ {
versionsEntry, err = s3a.getEntry(bucketDir, versionsObjectPath)
if err == nil {
break
}
if attempt < maxRetries {
// Exponential backoff with higher base: 100ms, 200ms, 400ms, 800ms, 1600ms, 3200ms, 6400ms
delay := time.Millisecond * time.Duration(100*(1<<(attempt-1)))
time.Sleep(delay)
}
}
if err != nil {
glog.Errorf("updateLatestVersionInDirectory: failed to get .versions directory for %s/%s after %d attempts: %v", bucket, object, maxRetries, err)
return fmt.Errorf("failed to get .versions directory after %d attempts: %w", maxRetries, err)
}
// Add or update the latest version metadata
if versionsEntry.Extended == nil {
versionsEntry.Extended = make(map[string][]byte)
}
versionsEntry.Extended[s3_constants.ExtLatestVersionIdKey] = []byte(versionId)
versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey] = []byte(versionFileName)
// Update the .versions directory entry with metadata
err = s3a.mkFile(bucketDir, versionsObjectPath, versionsEntry.Chunks, func(updatedEntry *filer_pb.Entry) {
updatedEntry.Extended = versionsEntry.Extended
updatedEntry.Attributes = versionsEntry.Attributes
updatedEntry.Chunks = versionsEntry.Chunks
})
if err != nil {
glog.Errorf("updateLatestVersionInDirectory: failed to update .versions directory metadata: %v", err)
return fmt.Errorf("failed to update .versions directory metadata: %w", err)
}
return nil
}
// extractObjectLockMetadataFromRequest extracts object lock headers from PUT requests
// and applies bucket default retention if no explicit retention is provided
func (s3a *S3ApiServer) extractObjectLockMetadataFromRequest(r *http.Request, entry *filer_pb.Entry) error {
if entry.Extended == nil {
entry.Extended = make(map[string][]byte)
}
// Extract explicit object lock mode (GOVERNANCE or COMPLIANCE)
explicitMode := r.Header.Get(s3_constants.AmzObjectLockMode)
if explicitMode != "" {
entry.Extended[s3_constants.ExtObjectLockModeKey] = []byte(explicitMode)
glog.V(2).Infof("extractObjectLockMetadataFromRequest: storing explicit object lock mode: %s", explicitMode)
}
// Extract explicit retention until date
explicitRetainUntilDate := r.Header.Get(s3_constants.AmzObjectLockRetainUntilDate)
if explicitRetainUntilDate != "" {
// Parse the ISO8601 date and convert to Unix timestamp for storage
parsedTime, err := time.Parse(time.RFC3339, explicitRetainUntilDate)
if err != nil {
glog.Errorf("extractObjectLockMetadataFromRequest: failed to parse retention until date, expected format: %s, error: %v", time.RFC3339, err)
return ErrInvalidRetentionDateFormat
}
entry.Extended[s3_constants.ExtRetentionUntilDateKey] = []byte(strconv.FormatInt(parsedTime.Unix(), 10))
glog.V(2).Infof("extractObjectLockMetadataFromRequest: storing explicit retention until date (timestamp: %d)", parsedTime.Unix())
}
// Extract legal hold status
if legalHold := r.Header.Get(s3_constants.AmzObjectLockLegalHold); legalHold != "" {
// Store S3 standard "ON"/"OFF" values directly
if legalHold == s3_constants.LegalHoldOn || legalHold == s3_constants.LegalHoldOff {
entry.Extended[s3_constants.ExtLegalHoldKey] = []byte(legalHold)
glog.V(2).Infof("extractObjectLockMetadataFromRequest: storing legal hold: %s", legalHold)
} else {
glog.Errorf("extractObjectLockMetadataFromRequest: unexpected legal hold value provided, expected 'ON' or 'OFF'")
return ErrInvalidLegalHoldStatus
}
}
// Apply bucket default retention if no explicit retention was provided
// This implements AWS S3 behavior where bucket default retention automatically applies to new objects
if explicitMode == "" && explicitRetainUntilDate == "" {
bucket, _ := s3_constants.GetBucketAndObject(r)
if err := s3a.applyBucketDefaultRetention(bucket, entry); err != nil {
glog.V(2).Infof("extractObjectLockMetadataFromRequest: skipping bucket default retention for %s: %v", bucket, err)
// Don't fail the upload if default retention can't be applied - this matches AWS behavior
}
}
return nil
}
// applyBucketDefaultEncryption applies bucket default encryption settings to a new object
// This implements AWS S3 behavior where bucket default encryption automatically applies to new objects
// when no explicit encryption headers are provided in the upload request.
// Returns the modified dataReader and encryption keys instead of using pointer parameters for better code clarity.
func (s3a *S3ApiServer) applyBucketDefaultEncryption(bucket string, r *http.Request, dataReader io.Reader) (*BucketDefaultEncryptionResult, error) {
// Check if bucket has default encryption configured
encryptionConfig, err := s3a.GetBucketEncryptionConfig(bucket)
if err != nil {
// Check if this is just "no encryption configured" vs a real error
if errors.Is(err, ErrNoEncryptionConfig) {
// No default encryption configured, return original reader
return &BucketDefaultEncryptionResult{DataReader: dataReader}, nil
}
// Real error - propagate to prevent silent encryption bypass
return nil, fmt.Errorf("failed to read bucket encryption config: %v", err)
}
if encryptionConfig == nil {
// No default encryption configured, return original reader
return &BucketDefaultEncryptionResult{DataReader: dataReader}, nil
}
if encryptionConfig.SseAlgorithm == "" {
// No encryption algorithm specified
return &BucketDefaultEncryptionResult{DataReader: dataReader}, nil
}
glog.V(3).Infof("applyBucketDefaultEncryption: applying default encryption %s for bucket %s", encryptionConfig.SseAlgorithm, bucket)
switch encryptionConfig.SseAlgorithm {
case EncryptionTypeAES256:
// Apply SSE-S3 (AES256) encryption
return s3a.applySSES3DefaultEncryption(dataReader)
case EncryptionTypeKMS:
// Apply SSE-KMS encryption
return s3a.applySSEKMSDefaultEncryption(bucket, r, dataReader, encryptionConfig)
default:
return nil, fmt.Errorf("unsupported default encryption algorithm: %s", encryptionConfig.SseAlgorithm)
}
}
// applySSES3DefaultEncryption applies SSE-S3 encryption as bucket default
func (s3a *S3ApiServer) applySSES3DefaultEncryption(dataReader io.Reader) (*BucketDefaultEncryptionResult, error) {
// Generate SSE-S3 key
keyManager := GetSSES3KeyManager()
key, err := keyManager.GetOrCreateKey("")
if err != nil {
return nil, fmt.Errorf("failed to generate SSE-S3 key for default encryption: %v", err)
}
// Create encrypted reader
encryptedReader, iv, encErr := CreateSSES3EncryptedReader(dataReader, key)
if encErr != nil {
return nil, fmt.Errorf("failed to create SSE-S3 encrypted reader for default encryption: %v", encErr)
}
// Store IV on the key object for later decryption
key.IV = iv
// Store key in manager for later retrieval
keyManager.StoreKey(key)
glog.V(3).Infof("applySSES3DefaultEncryption: applied SSE-S3 default encryption with key ID: %s", key.KeyID)
return &BucketDefaultEncryptionResult{
DataReader: encryptedReader,
SSES3Key: key,
}, nil
}
// applySSEKMSDefaultEncryption applies SSE-KMS encryption as bucket default
func (s3a *S3ApiServer) applySSEKMSDefaultEncryption(bucket string, r *http.Request, dataReader io.Reader, encryptionConfig *s3_pb.EncryptionConfiguration) (*BucketDefaultEncryptionResult, error) {
// Use the KMS key ID from bucket configuration, or default if not specified
keyID := encryptionConfig.KmsKeyId
if keyID == "" {
keyID = "alias/aws/s3" // AWS default KMS key for S3
}
// Check if bucket key is enabled in configuration
bucketKeyEnabled := encryptionConfig.BucketKeyEnabled
// Build encryption context for KMS
// Use bucket parameter passed to function (not from request parsing)
_, object := s3_constants.GetBucketAndObject(r)
encryptionContext := BuildEncryptionContext(bucket, object, bucketKeyEnabled)
// Create SSE-KMS encrypted reader
encryptedReader, sseKey, encErr := CreateSSEKMSEncryptedReaderWithBucketKey(dataReader, keyID, encryptionContext, bucketKeyEnabled)
if encErr != nil {
return nil, fmt.Errorf("failed to create SSE-KMS encrypted reader for default encryption: %v", encErr)
}
glog.V(3).Infof("applySSEKMSDefaultEncryption: applied SSE-KMS default encryption with key ID: %s", keyID)
return &BucketDefaultEncryptionResult{
DataReader: encryptedReader,
SSEKMSKey: sseKey,
}, nil
}
// applyBucketDefaultRetention applies bucket default retention settings to a new object
// This implements AWS S3 behavior where bucket default retention automatically applies to new objects
// when no explicit retention headers are provided in the upload request
func (s3a *S3ApiServer) applyBucketDefaultRetention(bucket string, entry *filer_pb.Entry) error {
// Safety check - if bucket config cache is not available, skip default retention
if s3a.bucketConfigCache == nil {
return nil
}
// Get bucket configuration (getBucketConfig handles caching internally)
bucketConfig, errCode := s3a.getBucketConfig(bucket)
if errCode != s3err.ErrNone {
return fmt.Errorf("failed to get bucket config: %v", errCode)
}
// Check if bucket has cached Object Lock configuration
if bucketConfig.ObjectLockConfig == nil {
return nil // No Object Lock configuration
}
objectLockConfig := bucketConfig.ObjectLockConfig
// Check if there's a default retention rule
if objectLockConfig.Rule == nil || objectLockConfig.Rule.DefaultRetention == nil {
return nil // No default retention configured
}
defaultRetention := objectLockConfig.Rule.DefaultRetention
// Validate default retention has required fields
if defaultRetention.Mode == "" {
return fmt.Errorf("default retention missing mode")
}
if !defaultRetention.DaysSet && !defaultRetention.YearsSet {
return fmt.Errorf("default retention missing period")
}
// Calculate retention until date based on default retention period
var retainUntilDate time.Time
now := time.Now()
if defaultRetention.DaysSet && defaultRetention.Days > 0 {
retainUntilDate = now.AddDate(0, 0, defaultRetention.Days)
} else if defaultRetention.YearsSet && defaultRetention.Years > 0 {
retainUntilDate = now.AddDate(defaultRetention.Years, 0, 0)
}
// Apply default retention to the object
if entry.Extended == nil {
entry.Extended = make(map[string][]byte)
}
entry.Extended[s3_constants.ExtObjectLockModeKey] = []byte(defaultRetention.Mode)
entry.Extended[s3_constants.ExtRetentionUntilDateKey] = []byte(strconv.FormatInt(retainUntilDate.Unix(), 10))
glog.V(2).Infof("applyBucketDefaultRetention: applied default retention %s until %s for bucket %s",
defaultRetention.Mode, retainUntilDate.Format(time.RFC3339), bucket)
return nil
}
// validateObjectLockHeaders validates object lock headers in PUT requests
func (s3a *S3ApiServer) validateObjectLockHeaders(r *http.Request, versioningEnabled bool) error {
// Extract object lock headers from request
mode := r.Header.Get(s3_constants.AmzObjectLockMode)
retainUntilDateStr := r.Header.Get(s3_constants.AmzObjectLockRetainUntilDate)
legalHold := r.Header.Get(s3_constants.AmzObjectLockLegalHold)
// Check if any object lock headers are present
hasObjectLockHeaders := mode != "" || retainUntilDateStr != "" || legalHold != ""
// Object lock headers can only be used on versioned buckets
if hasObjectLockHeaders && !versioningEnabled {
return ErrObjectLockVersioningRequired
}
// Validate object lock mode if present
if mode != "" {
if mode != s3_constants.RetentionModeGovernance && mode != s3_constants.RetentionModeCompliance {
return ErrInvalidObjectLockMode
}
}
// Validate retention date if present
if retainUntilDateStr != "" {
retainUntilDate, err := time.Parse(time.RFC3339, retainUntilDateStr)
if err != nil {
return ErrInvalidRetentionDateFormat
}
// Retention date must be in the future
if retainUntilDate.Before(time.Now()) {
return ErrRetentionDateMustBeFuture
}
}
// If mode is specified, retention date must also be specified
if mode != "" && retainUntilDateStr == "" {
return ErrObjectLockModeRequiresDate
}
// If retention date is specified, mode must also be specified
if retainUntilDateStr != "" && mode == "" {
return ErrRetentionDateRequiresMode
}
// Validate legal hold if present
if legalHold != "" {
if legalHold != s3_constants.LegalHoldOn && legalHold != s3_constants.LegalHoldOff {
return ErrInvalidLegalHoldStatus
}
}
// Check for governance bypass header - only valid for versioned buckets
bypassGovernance := r.Header.Get("x-amz-bypass-governance-retention") == "true"
// Governance bypass headers are only valid for versioned buckets (like object lock headers)
if bypassGovernance && !versioningEnabled {
return ErrGovernanceBypassVersioningRequired
}
return nil
}
// mapValidationErrorToS3Error maps object lock validation errors to appropriate S3 error codes
func mapValidationErrorToS3Error(err error) s3err.ErrorCode {
// Check for sentinel errors first
switch {
case errors.Is(err, ErrObjectLockVersioningRequired):
// For object lock operations on non-versioned buckets, return InvalidRequest
// This matches the test expectations
return s3err.ErrInvalidRequest
case errors.Is(err, ErrInvalidObjectLockMode):
// For invalid object lock mode, return InvalidRequest
// This matches the test expectations
return s3err.ErrInvalidRequest
case errors.Is(err, ErrInvalidLegalHoldStatus):
// For invalid legal hold status in XML body, return MalformedXML
// AWS S3 treats invalid status values in XML as malformed content
return s3err.ErrMalformedXML
case errors.Is(err, ErrInvalidRetentionDateFormat):
// For malformed retention date format, return MalformedDate
// This matches the test expectations
return s3err.ErrMalformedDate
case errors.Is(err, ErrRetentionDateMustBeFuture):
// For retention dates in the past, return InvalidRequest
// This matches the test expectations
return s3err.ErrInvalidRequest
case errors.Is(err, ErrObjectLockModeRequiresDate):
// For mode without retention date, return InvalidRequest
// This matches the test expectations
return s3err.ErrInvalidRequest
case errors.Is(err, ErrRetentionDateRequiresMode):
// For retention date without mode, return InvalidRequest
// This matches the test expectations
return s3err.ErrInvalidRequest
case errors.Is(err, ErrGovernanceBypassVersioningRequired):
// For governance bypass on non-versioned bucket, return InvalidRequest
// This matches the test expectations
return s3err.ErrInvalidRequest
case errors.Is(err, ErrMalformedXML):
// For malformed XML in request body, return MalformedXML
// This matches the test expectations for invalid retention mode and legal hold status
return s3err.ErrMalformedXML
case errors.Is(err, ErrInvalidRetentionPeriod):
// For invalid retention period (e.g., Days <= 0), return InvalidRetentionPeriod
// This matches the test expectations
return s3err.ErrInvalidRetentionPeriod
case errors.Is(err, ErrComplianceModeActive):
// For compliance mode retention violations, return AccessDenied
// This matches the test expectations
return s3err.ErrAccessDenied
case errors.Is(err, ErrGovernanceModeActive):
// For governance mode retention violations, return AccessDenied
// This matches the test expectations
return s3err.ErrAccessDenied
case errors.Is(err, ErrObjectUnderLegalHold):
// For legal hold violations, return AccessDenied
// This matches the test expectations
return s3err.ErrAccessDenied
case errors.Is(err, ErrGovernanceBypassNotPermitted):
// For governance bypass permission violations, return AccessDenied
// This matches the test expectations
return s3err.ErrAccessDenied
// Validation error constants
case errors.Is(err, ErrObjectLockConfigurationMissingEnabled):
return s3err.ErrMalformedXML
case errors.Is(err, ErrInvalidObjectLockEnabledValue):
return s3err.ErrMalformedXML
case errors.Is(err, ErrRuleMissingDefaultRetention):
return s3err.ErrMalformedXML
case errors.Is(err, ErrDefaultRetentionMissingMode):
return s3err.ErrMalformedXML
case errors.Is(err, ErrInvalidDefaultRetentionMode):
return s3err.ErrMalformedXML
case errors.Is(err, ErrDefaultRetentionMissingPeriod):
return s3err.ErrMalformedXML
case errors.Is(err, ErrDefaultRetentionBothDaysAndYears):
return s3err.ErrMalformedXML
case errors.Is(err, ErrDefaultRetentionDaysOutOfRange):
return s3err.ErrInvalidRetentionPeriod
case errors.Is(err, ErrDefaultRetentionYearsOutOfRange):
return s3err.ErrInvalidRetentionPeriod
}
// Check for error constants from the updated validation functions
switch {
case errors.Is(err, ErrRetentionMissingMode):
return s3err.ErrInvalidRequest
case errors.Is(err, ErrRetentionMissingRetainUntilDate):
return s3err.ErrInvalidRequest
case errors.Is(err, ErrInvalidRetentionModeValue):
return s3err.ErrMalformedXML
}
return s3err.ErrInvalidRequest
}
// EntryGetter interface for dependency injection in tests
// Simplified to only mock the data access dependency
type EntryGetter interface {
getEntry(parentDirectoryPath, entryName string) (*filer_pb.Entry, error)
}
// conditionalHeaders holds parsed conditional header values
type conditionalHeaders struct {
ifMatch string
ifNoneMatch string
ifModifiedSince time.Time
ifUnmodifiedSince time.Time
isSet bool // true if any conditional headers are present
}
// parseConditionalHeaders extracts and validates conditional headers from the request
func parseConditionalHeaders(r *http.Request) (conditionalHeaders, s3err.ErrorCode) {
headers := conditionalHeaders{
ifMatch: r.Header.Get(s3_constants.IfMatch),
ifNoneMatch: r.Header.Get(s3_constants.IfNoneMatch),
}
ifModifiedSinceStr := r.Header.Get(s3_constants.IfModifiedSince)
ifUnmodifiedSinceStr := r.Header.Get(s3_constants.IfUnmodifiedSince)
// Check if any conditional headers are present
headers.isSet = headers.ifMatch != "" || headers.ifNoneMatch != "" ||
ifModifiedSinceStr != "" || ifUnmodifiedSinceStr != ""
if !headers.isSet {
return headers, s3err.ErrNone
}
// Parse date headers with validation
var err error
if ifModifiedSinceStr != "" {
headers.ifModifiedSince, err = time.Parse(time.RFC1123, ifModifiedSinceStr)
if err != nil {
glog.V(3).Infof("parseConditionalHeaders: Invalid If-Modified-Since format: %v", err)
return headers, s3err.ErrInvalidRequest
}
}
if ifUnmodifiedSinceStr != "" {
headers.ifUnmodifiedSince, err = time.Parse(time.RFC1123, ifUnmodifiedSinceStr)
if err != nil {
glog.V(3).Infof("parseConditionalHeaders: Invalid If-Unmodified-Since format: %v", err)
return headers, s3err.ErrInvalidRequest
}
}
return headers, s3err.ErrNone
}
// S3ApiServer implements EntryGetter interface
func (s3a *S3ApiServer) getObjectETag(entry *filer_pb.Entry) string {
// Try to get ETag from Extended attributes first
if etagBytes, hasETag := entry.Extended[s3_constants.ExtETagKey]; hasETag {
return string(etagBytes)
}
// Check for Md5 in Attributes (matches filer.ETag behavior)
// Note: len(nil slice) == 0 in Go, so no need for explicit nil check
if entry.Attributes != nil && len(entry.Attributes.Md5) > 0 {
return fmt.Sprintf("\"%x\"", entry.Attributes.Md5)
}
// Fallback: calculate ETag from chunks
return s3a.calculateETagFromChunks(entry.Chunks)
}
func (s3a *S3ApiServer) etagMatches(headerValue, objectETag string) bool {
// Clean the object ETag
objectETag = strings.Trim(objectETag, `"`)
// Split header value by commas to handle multiple ETags
etags := strings.Split(headerValue, ",")
for _, etag := range etags {
etag = strings.TrimSpace(etag)
etag = strings.Trim(etag, `"`)
if etag == objectETag {
return true
}
}
return false
}
// checkConditionalHeadersWithGetter is a testable method that accepts a simple EntryGetter
// Uses the production getObjectETag and etagMatches methods to ensure testing of real logic
func (s3a *S3ApiServer) checkConditionalHeadersWithGetter(getter EntryGetter, r *http.Request, bucket, object string) s3err.ErrorCode {
headers, errCode := parseConditionalHeaders(r)
if errCode != s3err.ErrNone {
glog.V(3).Infof("checkConditionalHeaders: Invalid date format")
return errCode
}
if !headers.isSet {
return s3err.ErrNone
}
// Get object entry for conditional checks.
bucketDir := "/buckets/" + bucket
entry, entryErr := getter.getEntry(bucketDir, object)
objectExists := entryErr == nil
// For PUT requests, all specified conditions must be met.
// The evaluation order follows AWS S3 behavior for consistency.
// 1. Check If-Match
if headers.ifMatch != "" {
if !objectExists {
glog.V(3).Infof("checkConditionalHeaders: If-Match failed - object %s/%s does not exist", bucket, object)
return s3err.ErrPreconditionFailed
}
// If `ifMatch` is "*", the condition is met if the object exists.
// Otherwise, we need to check the ETag.
if headers.ifMatch != "*" {
// Use production getObjectETag method
objectETag := s3a.getObjectETag(entry)
// Use production etagMatches method
if !s3a.etagMatches(headers.ifMatch, objectETag) {
glog.V(3).Infof("checkConditionalHeaders: If-Match failed for object %s/%s - expected ETag %s, got %s", bucket, object, headers.ifMatch, objectETag)
return s3err.ErrPreconditionFailed
}
}
glog.V(3).Infof("checkConditionalHeaders: If-Match passed for object %s/%s", bucket, object)
}
// 2. Check If-Unmodified-Since
if !headers.ifUnmodifiedSince.IsZero() {
if objectExists {
objectModTime := time.Unix(entry.Attributes.Mtime, 0)
if objectModTime.After(headers.ifUnmodifiedSince) {
glog.V(3).Infof("checkConditionalHeaders: If-Unmodified-Since failed - object modified after %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
return s3err.ErrPreconditionFailed
}
glog.V(3).Infof("checkConditionalHeaders: If-Unmodified-Since passed - object not modified since %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
}
}
// 3. Check If-None-Match
if headers.ifNoneMatch != "" {
if objectExists {
if headers.ifNoneMatch == "*" {
glog.V(3).Infof("checkConditionalHeaders: If-None-Match=* failed - object %s/%s exists", bucket, object)
return s3err.ErrPreconditionFailed
}
// Use production getObjectETag method
objectETag := s3a.getObjectETag(entry)
// Use production etagMatches method
if s3a.etagMatches(headers.ifNoneMatch, objectETag) {
glog.V(3).Infof("checkConditionalHeaders: If-None-Match failed - ETag matches %s", objectETag)
return s3err.ErrPreconditionFailed
}
glog.V(3).Infof("checkConditionalHeaders: If-None-Match passed - ETag %s doesn't match %s", objectETag, headers.ifNoneMatch)
} else {
glog.V(3).Infof("checkConditionalHeaders: If-None-Match passed - object %s/%s does not exist", bucket, object)
}
}
// 4. Check If-Modified-Since
if !headers.ifModifiedSince.IsZero() {
if objectExists {
objectModTime := time.Unix(entry.Attributes.Mtime, 0)
if !objectModTime.After(headers.ifModifiedSince) {
glog.V(3).Infof("checkConditionalHeaders: If-Modified-Since failed - object not modified since %s", r.Header.Get(s3_constants.IfModifiedSince))
return s3err.ErrPreconditionFailed
}
glog.V(3).Infof("checkConditionalHeaders: If-Modified-Since passed - object modified after %s", r.Header.Get(s3_constants.IfModifiedSince))
}
}
return s3err.ErrNone
}
// checkConditionalHeaders is the production method that uses the S3ApiServer as EntryGetter
func (s3a *S3ApiServer) checkConditionalHeaders(r *http.Request, bucket, object string) s3err.ErrorCode {
return s3a.checkConditionalHeadersWithGetter(s3a, r, bucket, object)
}
// checkConditionalHeadersForReadsWithGetter is a testable method for read operations
// Uses the production getObjectETag and etagMatches methods to ensure testing of real logic
func (s3a *S3ApiServer) checkConditionalHeadersForReadsWithGetter(getter EntryGetter, r *http.Request, bucket, object string) ConditionalHeaderResult {
headers, errCode := parseConditionalHeaders(r)
if errCode != s3err.ErrNone {
glog.V(3).Infof("checkConditionalHeadersForReads: Invalid date format")
return ConditionalHeaderResult{ErrorCode: errCode}
}
if !headers.isSet {
return ConditionalHeaderResult{ErrorCode: s3err.ErrNone}
}
// Get object entry for conditional checks.
bucketDir := "/buckets/" + bucket
entry, entryErr := getter.getEntry(bucketDir, object)
objectExists := entryErr == nil
// If object doesn't exist, fail for If-Match and If-Unmodified-Since
if !objectExists {
if headers.ifMatch != "" {
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match failed - object %s/%s does not exist", bucket, object)
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: nil}
}
if !headers.ifUnmodifiedSince.IsZero() {
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since failed - object %s/%s does not exist", bucket, object)
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: nil}
}
// If-None-Match and If-Modified-Since succeed when object doesn't exist
// No entry to return since object doesn't exist
return ConditionalHeaderResult{ErrorCode: s3err.ErrNone, Entry: nil}
}
// Object exists - check all conditions
// The evaluation order follows AWS S3 behavior for consistency.
// 1. Check If-Match (412 Precondition Failed if fails)
if headers.ifMatch != "" {
// If `ifMatch` is "*", the condition is met if the object exists.
// Otherwise, we need to check the ETag.
if headers.ifMatch != "*" {
// Use production getObjectETag method
objectETag := s3a.getObjectETag(entry)
// Use production etagMatches method
if !s3a.etagMatches(headers.ifMatch, objectETag) {
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match failed for object %s/%s - expected ETag %s, got %s", bucket, object, headers.ifMatch, objectETag)
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: entry}
}
}
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match passed for object %s/%s", bucket, object)
}
// 2. Check If-Unmodified-Since (412 Precondition Failed if fails)
if !headers.ifUnmodifiedSince.IsZero() {
objectModTime := time.Unix(entry.Attributes.Mtime, 0)
if objectModTime.After(headers.ifUnmodifiedSince) {
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since failed - object modified after %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: entry}
}
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since passed - object not modified since %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
}
// 3. Check If-None-Match (304 Not Modified if fails)
if headers.ifNoneMatch != "" {
// Use production getObjectETag method
objectETag := s3a.getObjectETag(entry)
if headers.ifNoneMatch == "*" {
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match=* failed - object %s/%s exists", bucket, object)
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag, Entry: entry}
}
// Use production etagMatches method
if s3a.etagMatches(headers.ifNoneMatch, objectETag) {
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match failed - ETag matches %s", objectETag)
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag, Entry: entry}
}
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match passed - ETag %s doesn't match %s", objectETag, headers.ifNoneMatch)
}
// 4. Check If-Modified-Since (304 Not Modified if fails)
if !headers.ifModifiedSince.IsZero() {
objectModTime := time.Unix(entry.Attributes.Mtime, 0)
if !objectModTime.After(headers.ifModifiedSince) {
// Use production getObjectETag method
objectETag := s3a.getObjectETag(entry)
glog.V(3).Infof("checkConditionalHeadersForReads: If-Modified-Since failed - object not modified since %s", r.Header.Get(s3_constants.IfModifiedSince))
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag, Entry: entry}
}
glog.V(3).Infof("checkConditionalHeadersForReads: If-Modified-Since passed - object modified after %s", r.Header.Get(s3_constants.IfModifiedSince))
}
// Return success with the fetched entry for reuse
return ConditionalHeaderResult{ErrorCode: s3err.ErrNone, Entry: entry}
}
// checkConditionalHeadersForReads is the production method that uses the S3ApiServer as EntryGetter
func (s3a *S3ApiServer) checkConditionalHeadersForReads(r *http.Request, bucket, object string) ConditionalHeaderResult {
return s3a.checkConditionalHeadersForReadsWithGetter(s3a, r, bucket, object)
}
// deleteOrphanedChunks attempts to delete chunks that were uploaded but whose entry creation failed
// This prevents resource leaks and wasted storage. Errors are logged but don't prevent cleanup attempts.
func (s3a *S3ApiServer) deleteOrphanedChunks(chunks []*filer_pb.FileChunk) {
if len(chunks) == 0 {
return
}
// Extract file IDs from chunks
var fileIds []string
for _, chunk := range chunks {
if chunk.GetFileIdString() != "" {
fileIds = append(fileIds, chunk.GetFileIdString())
}
}
if len(fileIds) == 0 {
glog.Warningf("deleteOrphanedChunks: no valid file IDs found in %d chunks", len(chunks))
return
}
glog.V(3).Infof("deleteOrphanedChunks: attempting to delete %d file IDs: %v", len(fileIds), fileIds)
// Create a lookup function that queries the filer for volume locations
// This is similar to createLookupFileIdFunction but returns the format needed by DeleteFileIdsWithLookupVolumeId
lookupFunc := func(vids []string) (map[string]*operation.LookupResult, error) {
results := make(map[string]*operation.LookupResult)
err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
// Query filer for all volume IDs at once
resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
VolumeIds: vids,
})
if err != nil {
return err
}
// Convert filer response to operation.LookupResult format
for vid, locs := range resp.LocationsMap {
result := &operation.LookupResult{
VolumeOrFileId: vid,
}
for _, loc := range locs.Locations {
result.Locations = append(result.Locations, operation.Location{
Url: loc.Url,
PublicUrl: loc.PublicUrl,
DataCenter: loc.DataCenter,
GrpcPort: int(loc.GrpcPort),
})
}
results[vid] = result
}
return nil
})
return results, err
}
// Attempt deletion using the operation package's batch delete with custom lookup
deleteResults := operation.DeleteFileIdsWithLookupVolumeId(s3a.option.GrpcDialOption, fileIds, lookupFunc)
// Log results - track successes and failures
successCount := 0
failureCount := 0
for _, result := range deleteResults {
if result.Error != "" {
glog.Warningf("deleteOrphanedChunks: failed to delete chunk %s: %s (status: %d)",
result.FileId, result.Error, result.Status)
failureCount++
} else {
glog.V(4).Infof("deleteOrphanedChunks: successfully deleted chunk %s (size: %d bytes)",
result.FileId, result.Size)
successCount++
}
}
if failureCount > 0 {
glog.Warningf("deleteOrphanedChunks: cleanup completed with %d successes and %d failures out of %d chunks",
successCount, failureCount, len(fileIds))
} else {
glog.V(3).Infof("deleteOrphanedChunks: successfully deleted all %d orphaned chunks", successCount)
}
}