diff --git a/weed/s3api/s3_constants/extend_key.go b/weed/s3api/s3_constants/extend_key.go index e085e9bd2..e3ad07e5f 100644 --- a/weed/s3api/s3_constants/extend_key.go +++ b/weed/s3api/s3_constants/extend_key.go @@ -26,6 +26,13 @@ const ( // the entry's own mtime so legacy data still expires. ExtNoncurrentSinceNsKey = "Seaweed-X-Amz-Noncurrent-Since-Ns" + // Per-bucket opt-in for the PutObject lifecycle TTL fast path ("true" + // to enable). When on, an Expiration.Days rule is stamped as a volume + // TTL at write time instead of being expired by the worker. Off by + // default: a baked-in TTL can't honor a later policy change (rule + // removed or lengthened) the way worker-driven expiration does. + ExtLifecycleTtlFastPathKey = "Seaweed-X-Amz-Lifecycle-Ttl-Fast-Path" + // S3 checksum storage keys (use x-seaweedfs- prefix to avoid leaking in generic header loop) ExtChecksumAlgorithm = "x-seaweedfs-checksum-algorithm" ExtChecksumValue = "x-seaweedfs-checksum-value" diff --git a/weed/s3api/s3api_bucket_config.go b/weed/s3api/s3api_bucket_config.go index 363903431..c71b47a09 100644 --- a/weed/s3api/s3api_bucket_config.go +++ b/weed/s3api/s3api_bucket_config.go @@ -39,9 +39,9 @@ type BucketConfig struct { BucketPolicy *policy_engine.PolicyDocument // Cached bucket policy for performance // LifecycleTTL answers "what volume TTL should this PutObject get?" // using only fast-path-safe predicates (prefix + size; tags excluded - // because they're mutable post-PUT). nil = no TTL applies (no - // lifecycle config, versioned bucket, or only ineligible rules). - // The full canonical rule set lives inside the resolver; the + // because they're mutable post-PUT). nil = no TTL applies (fast path + // not enabled on the bucket, no lifecycle config, versioned bucket, + // or only ineligible rules). The fast path is opt-in per bucket. The // lifecycle worker reads bucket entries directly off the meta-log // rather than this cache. LifecycleTTL *LifecycleTTLResolver @@ -438,22 +438,26 @@ func (s3a *S3ApiServer) populateBucketConfigDerivedFields(config *BucketConfig) } config.BucketPolicy = loadBucketPolicyFromExtended(entry, bucket) - // Pre-parse lifecycle XML so the per-write TTL resolver doesn't - // pay parsing cost on every PutObject. nil on parse error so - // the PUT path falls through to "no TTL" rather than rejecting - // writes. - if xmlBytes, ok := entry.Extended[bucketLifecycleConfigurationXMLKey]; ok && len(xmlBytes) > 0 { - if rules, err := lifecycle_xml.ParseCanonical(xmlBytes); err == nil { - // Object Lock requires versioning, so an ObjectLockConfig - // implies the bucket is versioned even when the explicit - // Versioning header was never written. BucketIsVersioned - // in this file uses the same OR — keep them aligned. - versioned := config.Versioning == s3_constants.VersioningEnabled || - config.Versioning == s3_constants.VersioningSuspended || - config.ObjectLockConfig != nil - config.LifecycleTTL = NewLifecycleTTLResolver(rules, versioned) - } else { - glog.V(1).Infof("populateBucketConfigDerivedFields: bucket %s lifecycle xml parse: %v", bucket, err) + // The lifecycle TTL fast path is opt-in per bucket: a volume TTL + // stamped at write time can't honor a later policy change (rule + // removed or lengthened) the way worker-driven expiration does, + // so it stays off unless explicitly enabled. Skip the XML parse + // entirely when off. nil on parse error so the PUT path falls + // through to "no TTL" rather than rejecting writes. + if bytes.Equal(entry.Extended[s3_constants.ExtLifecycleTtlFastPathKey], []byte("true")) { + if xmlBytes, ok := entry.Extended[bucketLifecycleConfigurationXMLKey]; ok && len(xmlBytes) > 0 { + if rules, err := lifecycle_xml.ParseCanonical(xmlBytes); err == nil { + // Object Lock requires versioning, so an ObjectLockConfig + // implies the bucket is versioned even when the explicit + // Versioning header was never written. BucketIsVersioned + // in this file uses the same OR — keep them aligned. + versioned := config.Versioning == s3_constants.VersioningEnabled || + config.Versioning == s3_constants.VersioningSuspended || + config.ObjectLockConfig != nil + config.LifecycleTTL = NewLifecycleTTLResolver(rules, versioned) + } else { + glog.V(1).Infof("populateBucketConfigDerivedFields: bucket %s lifecycle xml parse: %v", bucket, err) + } } } } diff --git a/weed/s3api/s3api_object_lifecycle_ttl_test.go b/weed/s3api/s3api_object_lifecycle_ttl_test.go index 302c62aff..43460e30f 100644 --- a/weed/s3api/s3api_object_lifecycle_ttl_test.go +++ b/weed/s3api/s3api_object_lifecycle_ttl_test.go @@ -205,7 +205,9 @@ func TestPopulateBucketConfigDerivedFields_RefreshesLifecycleTTL(t *testing.T) { xmlAdd := []byte(`rEnabledlogs/7`) xmlReplace := []byte(`rEnabledlogs/30`) - cfg := &BucketConfig{Name: "bk", Entry: &filer_pb.Entry{Extended: map[string][]byte{}}} + cfg := &BucketConfig{Name: "bk", Entry: &filer_pb.Entry{Extended: map[string][]byte{ + s3_constants.ExtLifecycleTtlFastPathKey: []byte("true"), + }}} // 1) No XML yet → no resolver. s.populateBucketConfigDerivedFields(cfg) @@ -249,6 +251,7 @@ func TestPopulateBucketConfigDerivedFields_ObjectLockTreatedAsVersioned(t *testi Name: "bk", Entry: &filer_pb.Entry{Extended: map[string][]byte{ s3_constants.ExtObjectLockEnabledKey: []byte(s3_constants.ObjectLockEnabled), + s3_constants.ExtLifecycleTtlFastPathKey: []byte("true"), bucketLifecycleConfigurationXMLKey: xml, }}, } @@ -260,3 +263,27 @@ func TestPopulateBucketConfigDerivedFields_ObjectLockTreatedAsVersioned(t *testi t.Fatalf("ObjectLock buckets must skip the fast-path resolver, got %v", cfg.LifecycleTTL) } } + +func TestPopulateBucketConfigDerivedFields_TtlFastPathOptIn(t *testing.T) { + // The fast path is opt-in per bucket: lifecycle XML alone must not + // stamp volume TTL on writes. Only the explicit flag builds the + // resolver. + s := &S3ApiServer{} + xml := []byte(`rEnabledlogs/7`) + + // XML present but flag off → nil resolver (worker drives expiration). + cfg := &BucketConfig{Name: "bk", Entry: &filer_pb.Entry{Extended: map[string][]byte{ + bucketLifecycleConfigurationXMLKey: xml, + }}} + s.populateBucketConfigDerivedFields(cfg) + if cfg.LifecycleTTL != nil { + t.Fatalf("fast path off must yield nil resolver, got %v", cfg.LifecycleTTL) + } + + // Flag on → resolver applies the rule. + cfg.Entry.Extended[s3_constants.ExtLifecycleTtlFastPathKey] = []byte("true") + s.populateBucketConfigDerivedFields(cfg) + if got := cfg.LifecycleTTL.Resolve("logs/foo", 1); got != 7*86400 { + t.Fatalf("fast path on, want 7d, got %d", got) + } +} diff --git a/weed/shell/command_s3_bucket_lifecycle_fastpath.go b/weed/shell/command_s3_bucket_lifecycle_fastpath.go new file mode 100644 index 000000000..d139f7b0e --- /dev/null +++ b/weed/shell/command_s3_bucket_lifecycle_fastpath.go @@ -0,0 +1,118 @@ +package shell + +import ( + "context" + "flag" + "fmt" + "io" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket" +) + +func init() { + Commands = append(Commands, &commandS3BucketLifecycleFastpath{}) +} + +type commandS3BucketLifecycleFastpath struct { +} + +func (c *commandS3BucketLifecycleFastpath) Name() string { + return "s3.bucket.lifecycle.fastpath" +} + +func (c *commandS3BucketLifecycleFastpath) Help() string { + return `view or toggle the per-bucket lifecycle TTL fast path + + When enabled, an Expiration.Days lifecycle rule is stamped as a volume + TTL at PutObject time, so the volume server reclaims the data on its own + and the lifecycle worker skips per-chunk deletes. Off by default: a + volume TTL is baked into the object at write time and can't honor a later + policy change (rule removed or lengthened), unlike worker-driven + expiration. The fast path never applies to versioned or object-locked + buckets regardless of this flag. + + Example: + # Show the current setting + s3.bucket.lifecycle.fastpath -name + + # Enable + s3.bucket.lifecycle.fastpath -name -enable + + # Disable + s3.bucket.lifecycle.fastpath -name -disable +` +} + +func (c *commandS3BucketLifecycleFastpath) HasTag(CommandTag) bool { + return false +} + +func (c *commandS3BucketLifecycleFastpath) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { + bucketCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + bucketName := bucketCommand.String("name", "", "bucket name") + enable := bucketCommand.Bool("enable", false, "enable the lifecycle TTL fast path") + disable := bucketCommand.Bool("disable", false, "disable the lifecycle TTL fast path") + if err = bucketCommand.Parse(args); err != nil { + return err + } + + if *bucketName == "" { + return fmt.Errorf("empty bucket name") + } + if err := s3bucket.VerifyS3BucketName(*bucketName); err != nil { + return fmt.Errorf("invalid bucket name %q: %w", *bucketName, err) + } + if *enable && *disable { + return fmt.Errorf("only one of -enable or -disable can be set") + } + + return commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) + if err != nil { + return fmt.Errorf("get filer configuration: %w", err) + } + filerBucketsPath := resp.DirBuckets + + lookupResp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{ + Directory: filerBucketsPath, + Name: *bucketName, + }) + if err != nil { + return fmt.Errorf("lookup bucket %s: %w", *bucketName, err) + } + entry := lookupResp.Entry + + if !*enable && !*disable { + state := "disabled" + if string(entry.Extended[s3_constants.ExtLifecycleTtlFastPathKey]) == "true" { + state = "enabled" + } + fmt.Fprintf(writer, "Bucket: %s\n", *bucketName) + fmt.Fprintf(writer, "Lifecycle TTL fast path: %s\n", state) + return nil + } + + if entry.Extended == nil { + entry.Extended = make(map[string][]byte) + } + state := "disabled" + if *enable { + entry.Extended[s3_constants.ExtLifecycleTtlFastPathKey] = []byte("true") + state = "enabled" + } else { + delete(entry.Extended, s3_constants.ExtLifecycleTtlFastPathKey) + } + + if _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ + Directory: filerBucketsPath, + Entry: entry, + }); err != nil { + return fmt.Errorf("failed to update bucket: %w", err) + } + + fmt.Fprintf(writer, "Bucket %s lifecycle TTL fast path %s\n", *bucketName, state) + return nil + }) +}