mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 13:30:46 +02:00
refactor(s3api): drop filer.conf TTL routing from PUT lifecycle (#9379)
PutBucketLifecycleConfiguration used to install /buckets/<bucket>/<prefix> day-TTL entries in filer.conf so the volume server's RocksDB compaction filter would expire matching writes. With 9377 the s3api server now stamps volume TTL per-write via LifecycleTTLResolver off the stored XML, which covers the same prefix-only Expiration.Days subset and additionally handles size filters and AWS overlapping-rule precedence. Maintaining both paths means a rule change has to mutate two stores in lockstep, and the filer.conf path can't represent everything the resolver does. Drop the add path. Keep a one-way cleanup loop so an upgrade still wipes day-TTL entries written by older builds — otherwise a stale entry would silently double-stamp writes (volume server expires under the old rule) or contradict the new XML after a rule change. Also removes resolveLifecycleDefaultsFromFilerConf (no longer needed) and the versioning-fast-path guard (the resolver itself returns nil for versioned/object-lock buckets, covered by TestNewLifecycleTTLResolver_NilOnVersionedBucket). Tests covering the deleted helpers are deleted with them; the GET fallback that synthesizes lifecycle rules from existing filer.conf TTLs is unchanged so users who historically configured TTL via filer.conf directly still see a rule.
This commit is contained in:
Chris Lu
@@ -22,7 +22,6 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
|
||||
@@ -872,28 +871,6 @@ func (s3a *S3ApiServer) GetBucketLifecycleConfigurationHandler(w http.ResponseWr
|
||||
writeSuccessResponseXML(w, r, response)
|
||||
}
|
||||
|
||||
// resolveLifecycleDefaultsFromFilerConf returns replication and volumeGrowthCount for use when adding a lifecycle TTL rule.
|
||||
// S3 does not set DataCenter/Rack/DataNode so placement is not pinned to a specific DC/rack.
|
||||
// Precedence: parent path rule first, then filer global. If volumeGrowthCount is 0 but replication is set,
|
||||
// use replication's copy count so the rule is valid (volumeGrowthCount must be divisible by copy count).
|
||||
func resolveLifecycleDefaultsFromFilerConf(fc *filer.FilerConf, filerConfigReplication, bucketsPath, bucket string) (replication string, volumeGrowthCount uint32, err error) {
|
||||
bucketPath := fmt.Sprintf("%s/%s/", bucketsPath, bucket)
|
||||
parentRule := fc.MatchStorageRule(bucketPath)
|
||||
replication = parentRule.Replication
|
||||
if replication == "" {
|
||||
replication = filerConfigReplication
|
||||
}
|
||||
volumeGrowthCount = parentRule.VolumeGrowthCount
|
||||
if volumeGrowthCount == 0 && replication != "" {
|
||||
var rp *super_block.ReplicaPlacement
|
||||
rp, err = super_block.NewReplicaPlacementFromString(replication)
|
||||
if err == nil {
|
||||
volumeGrowthCount = uint32(rp.GetCopyCount())
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// PutBucketLifecycleConfigurationHandler Put Bucket Lifecycle configuration
|
||||
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketLifecycleConfiguration.html
|
||||
func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -926,40 +903,34 @@ func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWr
|
||||
return
|
||||
}
|
||||
|
||||
// Reject Transition rules — they require storage class migration
|
||||
// infrastructure that does not exist yet. Validate before touching
|
||||
// any backing state so a malformed PUT can't half-apply.
|
||||
for _, rule := range lifeCycleConfig.Rules {
|
||||
if rule.Status != lifecycle_xml.Enabled {
|
||||
continue
|
||||
}
|
||||
if rule.Transition.Set() || rule.NoncurrentVersionTransition.Set() {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Migration: clear any day-TTL filer.conf entries this handler
|
||||
// installed in older builds. Per-write TTL is now driven by the
|
||||
// LifecycleTTLResolver constructed off the stored XML, so leaving a
|
||||
// stale day-TTL entry under /buckets/<bucket>/ would double-stamp
|
||||
// (volume server expires under the old rule) or contradict the new
|
||||
// XML after a rule change. The add path is gone — this loop only
|
||||
// shrinks the conf, never grows it.
|
||||
fc, err := filer.ReadFilerConfFromFilers(s3a.option.Filers, s3a.option.GrpcDialOption, nil)
|
||||
if err != nil {
|
||||
glog.Errorf("PutBucketLifecycleConfigurationHandler read filer config: %s", err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
|
||||
// Resolve replication so lifecycle rules do not create filer.conf entries with empty replication.
|
||||
var filerConfigReplication string
|
||||
if filerErr := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
||||
resp, err := client.GetFilerConfiguration(r.Context(), &filer_pb.GetFilerConfigurationRequest{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
filerConfigReplication = resp.GetReplication()
|
||||
return nil
|
||||
}); filerErr != nil {
|
||||
glog.V(2).Infof("PutBucketLifecycleConfigurationHandler: could not get filer config: %v", filerErr)
|
||||
}
|
||||
defaultReplication, defaultVolumeGrowthCount, err := resolveLifecycleDefaultsFromFilerConf(fc, filerConfigReplication, s3a.option.BucketsPath, bucket)
|
||||
if err != nil {
|
||||
glog.Warningf("PutBucketLifecycleConfigurationHandler bucket %s: invalid replication %q: %v", bucket, defaultReplication, err)
|
||||
}
|
||||
|
||||
collectionName := s3a.getCollectionName(bucket)
|
||||
collectionTtls := fc.GetCollectionTtls(collectionName)
|
||||
collectionTtls := fc.GetCollectionTtls(s3a.getCollectionName(bucket))
|
||||
changed := false
|
||||
|
||||
// PUT replaces the entire lifecycle policy, so any day-TTL filer.conf
|
||||
// entry left over from a previous PUT (rule removed, prefix changed,
|
||||
// rule disabled, gained a tag/size filter, bucket switched to
|
||||
// versioned) must be removed first — otherwise a stale entry keeps
|
||||
// routing new writes to the old collection/replication and the volume
|
||||
// server keeps expiring objects under the prior TTL.
|
||||
bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, bucket)
|
||||
for prefix, ttl := range collectionTtls {
|
||||
if !strings.HasPrefix(prefix, bucketPrefix) || !strings.HasSuffix(ttl, "d") {
|
||||
@@ -968,95 +939,6 @@ func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWr
|
||||
fc.DeleteLocationConf(prefix)
|
||||
changed = true
|
||||
}
|
||||
// Re-read after removing so the per-rule dedupe below sees the freshly
|
||||
// cleared state, not the snapshot from before the loop above.
|
||||
collectionTtls = fc.GetCollectionTtls(collectionName)
|
||||
|
||||
// Check whether the bucket has versioning enabled. Versioned buckets must
|
||||
// NOT use the TTL fast-path because:
|
||||
// 1. TTL volumes expire as a unit, destroying all data — including
|
||||
// noncurrent versions that should be preserved.
|
||||
// 2. Filer-backend TTL (RocksDB compaction, Redis expire) removes entries
|
||||
// without triggering chunk deletion, leaving orphaned volume data.
|
||||
// 3. On AWS S3, Expiration.Days on a versioned bucket creates a delete
|
||||
// marker — it does not delete data. TTL has no such nuance.
|
||||
// For versioned buckets the lifecycle worker handles all rule evaluation
|
||||
// at scan time, which correctly operates on individual versions.
|
||||
bucketVersioning, versioningErr := s3a.getBucketVersioningStatus(bucket)
|
||||
if versioningErr != s3err.ErrNone {
|
||||
// Fail closed: if we cannot determine versioning status, treat the
|
||||
// bucket as versioned to avoid creating TTL entries that would
|
||||
// destroy noncurrent versions.
|
||||
glog.V(1).Infof("PutBucketLifecycleConfigurationHandler: could not determine versioning status for %s (err %v), skipping TTL fast-path", bucket, versioningErr)
|
||||
}
|
||||
isVersioned := versioningErr != s3err.ErrNone ||
|
||||
bucketVersioning == s3_constants.VersioningEnabled ||
|
||||
bucketVersioning == s3_constants.VersioningSuspended
|
||||
|
||||
for _, rule := range lifeCycleConfig.Rules {
|
||||
if rule.Status != lifecycle_xml.Enabled {
|
||||
continue
|
||||
}
|
||||
// Reject Transition rules — they require storage class migration
|
||||
// infrastructure that does not exist yet.
|
||||
if rule.Transition.Set() || rule.NoncurrentVersionTransition.Set() {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
|
||||
return
|
||||
}
|
||||
|
||||
if isVersioned {
|
||||
continue // all rules evaluated by lifecycle worker at scan time
|
||||
}
|
||||
|
||||
var rulePrefix string
|
||||
switch {
|
||||
case rule.Filter.AndSet():
|
||||
rulePrefix = rule.Filter.And.Prefix.Val()
|
||||
case rule.Filter.Prefix.Set():
|
||||
rulePrefix = rule.Filter.Prefix.Val()
|
||||
case rule.Prefix.Set():
|
||||
rulePrefix = rule.Prefix.Val()
|
||||
}
|
||||
|
||||
// Only create filer.conf TTL entries for simple Expiration.Days rules
|
||||
// with prefix-only filters (the fast path handled by RocksDB compaction
|
||||
// filter). Rules with tag or size filters must be evaluated at scan time
|
||||
// by the lifecycle worker, because TTL applies to all objects under the
|
||||
// prefix regardless of tags or size.
|
||||
if rule.Expiration.Days == 0 {
|
||||
continue
|
||||
}
|
||||
hasTagOrSizeFilter := rule.Filter.TagSet() ||
|
||||
rule.Filter.ObjectSizeGreaterThan > 0 || rule.Filter.ObjectSizeLessThan > 0 ||
|
||||
(rule.Filter.AndSet() && (len(rule.Filter.And.Tags) > 0 ||
|
||||
rule.Filter.And.ObjectSizeGreaterThan > 0 || rule.Filter.And.ObjectSizeLessThan > 0))
|
||||
if hasTagOrSizeFilter {
|
||||
continue // evaluated by lifecycle worker at scan time
|
||||
}
|
||||
locationPrefix := fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, bucket, rulePrefix)
|
||||
locConf := &filer_pb.FilerConf_PathConf{
|
||||
LocationPrefix: locationPrefix,
|
||||
Collection: collectionName,
|
||||
Ttl: fmt.Sprintf("%dd", rule.Expiration.Days),
|
||||
Replication: defaultReplication,
|
||||
VolumeGrowthCount: defaultVolumeGrowthCount,
|
||||
// DataCenter/Rack/DataNode intentionally not set: S3 is not tied to a specific DC/rack,
|
||||
// requests can hit any filer; setting them would pin placement unnecessarily.
|
||||
}
|
||||
if ttl, ok := collectionTtls[locConf.LocationPrefix]; ok && ttl == locConf.Ttl {
|
||||
continue
|
||||
}
|
||||
if err := fc.AddLocationConf(locConf); err != nil {
|
||||
glog.Errorf("PutBucketLifecycleConfigurationHandler add location config: %s", err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
// Existing entries are not back-stamped here; the lifecycle worker
|
||||
// drives expiration off the meta-log and bootstrap walk so a PUT
|
||||
// stays O(rules) instead of O(objects). New writes inherit TTL from
|
||||
// the filer.conf entry above.
|
||||
changed = true
|
||||
}
|
||||
|
||||
if changed {
|
||||
var buf bytes.Buffer
|
||||
|
||||
@@ -1,80 +0,0 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestResolveLifecycleDefaultsFromFilerConf(t *testing.T) {
|
||||
// Precedence: global (lowest), then path rules top-down (parent overrides global), then query (highest).
|
||||
// So parent path rule has priority over filer global config.
|
||||
|
||||
t.Run("parent_rule_replication_takes_precedence_over_filer_config", func(t *testing.T) {
|
||||
fc := filer.NewFilerConf()
|
||||
fc.SetLocationConf(&filer_pb.FilerConf_PathConf{
|
||||
LocationPrefix: "/buckets/",
|
||||
Replication: "001",
|
||||
})
|
||||
repl, vgc, err := resolveLifecycleDefaultsFromFilerConf(fc, "010", "/buckets", "mybucket")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "001", repl, "parent path rule must override filer global config")
|
||||
assert.Equal(t, uint32(2), vgc, "volumeGrowthCount derived from replication 001 copy count (SameRackCount=1 -> 2 copies)")
|
||||
})
|
||||
|
||||
t.Run("falls_back_to_filer_config_when_parent_rule_replication_empty", func(t *testing.T) {
|
||||
fc := filer.NewFilerConf()
|
||||
fc.SetLocationConf(&filer_pb.FilerConf_PathConf{
|
||||
LocationPrefix: "/buckets/",
|
||||
Replication: "", // no replication on parent
|
||||
})
|
||||
repl, vgc, err := resolveLifecycleDefaultsFromFilerConf(fc, "010", "/buckets", "mybucket")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "010", repl, "replication should come from filer config when parent rule has none")
|
||||
assert.Equal(t, uint32(2), vgc, "volumeGrowthCount derived from replication 010 copy count")
|
||||
})
|
||||
|
||||
t.Run("parent_rule_empty_when_no_matching_prefix_uses_filer_config", func(t *testing.T) {
|
||||
fc := filer.NewFilerConf()
|
||||
// no rules; parent path /buckets/mybucket/ matches nothing
|
||||
repl, vgc, err := resolveLifecycleDefaultsFromFilerConf(fc, "010", "/buckets", "mybucket")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "010", repl, "when no path rule, use filer config replication")
|
||||
assert.Equal(t, uint32(2), vgc, "volumeGrowthCount derived from replication 010")
|
||||
})
|
||||
|
||||
t.Run("all_empty_when_no_parent_rule_and_no_filer_config", func(t *testing.T) {
|
||||
fc := filer.NewFilerConf()
|
||||
repl, vgc, err := resolveLifecycleDefaultsFromFilerConf(fc, "", "/buckets", "mybucket")
|
||||
assert.NoError(t, err)
|
||||
assert.Empty(t, repl)
|
||||
assert.Equal(t, uint32(0), vgc)
|
||||
})
|
||||
|
||||
t.Run("parent_rule_volume_growth_count_used_when_set", func(t *testing.T) {
|
||||
fc := filer.NewFilerConf()
|
||||
fc.SetLocationConf(&filer_pb.FilerConf_PathConf{
|
||||
LocationPrefix: "/buckets/",
|
||||
Replication: "010",
|
||||
VolumeGrowthCount: 4,
|
||||
})
|
||||
repl, vgc, err := resolveLifecycleDefaultsFromFilerConf(fc, "010", "/buckets", "mybucket")
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "010", repl)
|
||||
assert.Equal(t, uint32(4), vgc, "parent VolumeGrowthCount must be used when set")
|
||||
})
|
||||
|
||||
t.Run("invalid_replication_returns_error", func(t *testing.T) {
|
||||
fc := filer.NewFilerConf()
|
||||
fc.SetLocationConf(&filer_pb.FilerConf_PathConf{
|
||||
LocationPrefix: "/buckets/",
|
||||
Replication: "0x1", // invalid: non-digit
|
||||
})
|
||||
repl, vgc, err := resolveLifecycleDefaultsFromFilerConf(fc, "", "/buckets", "mybucket")
|
||||
assert.Error(t, err)
|
||||
assert.Equal(t, "0x1", repl, "replication string is still returned")
|
||||
assert.Equal(t, uint32(0), vgc, "volumeGrowthCount remains 0 when parse fails")
|
||||
})
|
||||
}
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -126,38 +125,3 @@ func (f failingReadCloser) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestShouldSkipTTLFastPathForVersionedBuckets verifies the versioning guard
|
||||
// logic that PutBucketLifecycleConfigurationHandler uses to decide whether
|
||||
// to create filer.conf TTL entries. On AWS S3, Expiration.Days on a versioned
|
||||
// bucket creates a delete marker — it does not delete data. TTL volumes
|
||||
// would destroy all versions indiscriminately, so the lifecycle worker must
|
||||
// handle versioned buckets at scan time instead. (issue #8757)
|
||||
//
|
||||
// Note: an integration test that invokes PutBucketLifecycleConfigurationHandler
|
||||
// directly is not feasible here because the handler requires filer gRPC
|
||||
// connectivity (ReadFilerConfFromFilers, WithFilerClient) before it reaches
|
||||
// the versioning check. The lifecycle worker integration tests in
|
||||
// weed/plugin/worker/lifecycle/ cover the end-to-end versioned-bucket behavior.
|
||||
func TestShouldSkipTTLFastPathForVersionedBuckets(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
versioning string
|
||||
expectSkip bool
|
||||
}{
|
||||
{"versioning_enabled_skips_ttl", s3_constants.VersioningEnabled, true},
|
||||
{"versioning_suspended_skips_ttl", s3_constants.VersioningSuspended, true},
|
||||
{"unversioned_allows_ttl", "", false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// This mirrors the guard in PutBucketLifecycleConfigurationHandler:
|
||||
// isVersioned := versioningErr != s3err.ErrNone ||
|
||||
// bucketVersioning == s3_constants.VersioningEnabled ||
|
||||
// bucketVersioning == s3_constants.VersioningSuspended
|
||||
// When isVersioned is true, the TTL fast-path is skipped entirely.
|
||||
isVersioned := tt.versioning == s3_constants.VersioningEnabled ||
|
||||
tt.versioning == s3_constants.VersioningSuspended
|
||||
assert.Equal(t, tt.expectSkip, isVersioned)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user