[s3] Put bucket lifecycle configuration (#5510)

.github/workflows/s3tests.yml

@@ -194,4 +194,7 @@ jobs:
           s3tests_boto3/functional/test_s3.py::test_ranged_request_skip_leading_bytes_response_code \
           s3tests_boto3/functional/test_s3.py::test_ranged_request_return_trailing_bytes_response_code \
           s3tests_boto3/functional/test_s3.py::test_copy_object_ifmatch_good \
-          s3tests_boto3/functional/test_s3.py::test_copy_object_ifnonematch_failed
+          s3tests_boto3/functional/test_s3.py::test_copy_object_ifnonematch_failed \
+          s3tests_boto3/functional/test_s3.py::test_lifecycle_set \
+          s3tests_boto3/functional/test_s3.py::test_lifecycle_get \
+          s3tests_boto3/functional/test_s3.py::test_lifecycle_set_filter

weed/s3api/s3api_bucket_handlers.go

@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"bytes"
 	"context"
 	"encoding/xml"
 	"errors"
@@ -10,6 +11,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"math"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -325,38 +327,155 @@ func (s3a *S3ApiServer) GetBucketLifecycleConfigurationHandler(w http.ResponseWr
 		s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchLifecycleConfiguration)
 		return
 	}
+
 	response := Lifecycle{}
-	for prefix, internalTtl := range ttls {
+	for locationPrefix, internalTtl := range ttls {
 		ttl, _ := needle.ReadTTL(internalTtl)
 		days := int(ttl.Minutes() / 60 / 24)
 		if days == 0 {
 			continue
 		}
+		prefix, found := strings.CutPrefix(locationPrefix, fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, bucket))
+		if !found {
+			continue
+		}
 		response.Rules = append(response.Rules, Rule{
-			Status: Enabled, Filter: Filter{
+			ID:         prefix,
-				Prefix: Prefix{string: prefix, set: true},
+			Status:     Enabled,
-				set:    true,
+			Prefix:     Prefix{val: prefix, set: true},
-			},
 			Expiration: Expiration{Days: days, set: true},
 		})
 	}
+
 	writeSuccessResponseXML(w, r, response)
 }
 
 // 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) {
+	// collect parameters
+	bucket, _ := s3_constants.GetBucketAndObject(r)
+	glog.V(3).Infof("PutBucketLifecycleConfigurationHandler %s", bucket)
 
-	s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, err)
+		return
+	}
 
+	lifeCycleConfig := Lifecycle{}
+	if err := xmlDecoder(r.Body, &lifeCycleConfig, r.ContentLength); err != nil {
+		glog.Warningf("PutBucketLifecycleConfigurationHandler xml decode: %s", err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
+		return
+	}
+
+	fc, err := filer.ReadFilerConf(s3a.option.Filer, s3a.option.GrpcDialOption, nil)
+	if err != nil {
+		glog.Errorf("PutBucketLifecycleConfigurationHandler read filer config: %s", err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		return
+	}
+	collectionName := s3a.getCollectionName(bucket)
+	collectionTtls := fc.GetCollectionTtls(collectionName)
+	changed := false
+
+	for _, rule := range lifeCycleConfig.Rules {
+		if rule.Status != Enabled {
+			continue
+		}
+		var rulePrefix string
+		switch {
+		case rule.Filter.Prefix.set:
+			rulePrefix = rule.Filter.Prefix.val
+		case rule.Prefix.set:
+			rulePrefix = rule.Prefix.val
+		case !rule.Expiration.Date.IsZero() || rule.Transition.Days > 0 || !rule.Transition.Date.IsZero():
+			s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
+			return
+		}
+
+		if rule.Expiration.Days == 0 {
+			continue
+		}
+
+		locConf := &filer_pb.FilerConf_PathConf{
+			LocationPrefix: fmt.Sprintf("%s/%s/%s", s3a.option.BucketsPath, bucket, rulePrefix),
+			Collection:     collectionName,
+			Ttl:            fmt.Sprintf("%dd", rule.Expiration.Days),
+		}
+		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
+		}
+		changed = true
+	}
+
+	if changed {
+		var buf bytes.Buffer
+		if err := fc.ToText(&buf); err != nil {
+			glog.Errorf("PutBucketLifecycleConfigurationHandler save config to text: %s", err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		}
+		if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
+		}); err != nil {
+			glog.Errorf("PutBucketLifecycleConfigurationHandler save config inside filer: %s", err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			return
+		}
+	}
+
+	writeSuccessResponseEmpty(w, r)
 }
 
 // DeleteBucketLifecycleHandler Delete Bucket Lifecycle
 // https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteBucketLifecycle.html
 func (s3a *S3ApiServer) DeleteBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
+	// collect parameters
+	bucket, _ := s3_constants.GetBucketAndObject(r)
+	glog.V(3).Infof("DeleteBucketLifecycleHandler %s", bucket)
+
+	if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, err)
+		return
+	}
+
+	fc, err := filer.ReadFilerConf(s3a.option.Filer, s3a.option.GrpcDialOption, nil)
+	if err != nil {
+		glog.Errorf("DeleteBucketLifecycleHandler read filer config: %s", err)
+		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		return
+	}
+	collectionTtls := fc.GetCollectionTtls(s3a.getCollectionName(bucket))
+	changed := false
+	for prefix, ttl := range collectionTtls {
+		bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, bucket)
+		if strings.HasPrefix(prefix, bucketPrefix) && strings.HasSuffix(ttl, "d") {
+			fc.DeleteLocationConf(prefix)
+			changed = true
+		}
+	}
+
+	if changed {
+		var buf bytes.Buffer
+		if err := fc.ToText(&buf); err != nil {
+			glog.Errorf("DeleteBucketLifecycleHandler save config to text: %s", err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+		}
+		if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+			return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
+		}); err != nil {
+			glog.Errorf("DeleteBucketLifecycleHandler save config inside filer: %s", err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			return
+		}
+	}
 
 	s3err.WriteEmptyResponse(w, r, http.StatusNoContent)
-
 }
 
 // GetBucketLocationHandler Get bucket location

weed/s3api/s3api_policy.go

@@ -47,8 +47,14 @@ type Filter struct {
 
 // Prefix holds the prefix xml tag in <Rule> and <Filter>
 type Prefix struct {
-	string
+	XMLName xml.Name `xml:"Prefix"`
-	set bool
+	set     bool
+
+	val string
+}
+
+func (p Prefix) String() string {
+	return p.val
 }
 
 // MarshalXML encodes Prefix field into an XML form.
@@ -56,11 +62,21 @@ func (p Prefix) MarshalXML(e *xml.Encoder, startElement xml.StartElement) error
 	if !p.set {
 		return nil
 	}
-	return e.EncodeElement(p.string, startElement)
+	return e.EncodeElement(p.val, startElement)
+}
+
+func (p *Prefix) UnmarshalXML(d *xml.Decoder, startElement xml.StartElement) error {
+	prefix := ""
+	_ = d.DecodeElement(&prefix, &startElement)
+	*p = Prefix{set: true, val: prefix}
+	return nil
 }
 
 // MarshalXML encodes Filter field into an XML form.
 func (f Filter) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
+	if !f.set {
+		return nil
+	}
 	if err := e.EncodeToken(start); err != nil {
 		return err
 	}