ilm: ObjectSizeLessThan and ObjectSizeGreaterThan (#18500)

cmd/bucket-lifecycle.go

@@ -872,6 +872,7 @@ func (oi ObjectInfo) ToLifecycleOpts() lifecycle.ObjectOpts {
 		UserTags:         oi.UserTags,
 		VersionID:        oi.VersionID,
 		ModTime:          oi.ModTime,
+		Size:             oi.Size,
 		IsLatest:         oi.IsLatest,
 		NumVersions:      oi.NumVersions,
 		DeleteMarker:     oi.DeleteMarker,

internal/bucket/lifecycle/and.go

@@ -26,25 +26,36 @@ var errDuplicateTagKey = Errorf("Duplicate Tag Keys are not allowed")
 // And - a tag to combine a prefix and multiple tags for lifecycle configuration rule.
 type And struct {
 	XMLName               xml.Name `xml:"And"`
+	ObjectSizeGreaterThan int64    `xml:"ObjectSizeGreaterThan,omitempty"`
+	ObjectSizeLessThan    int64    `xml:"ObjectSizeLessThan,omitempty"`
 	Prefix                Prefix   `xml:"Prefix,omitempty"`
 	Tags                  []Tag    `xml:"Tag,omitempty"`
 }
 
 // isEmpty returns true if Tags field is null
 func (a And) isEmpty() bool {
-	return len(a.Tags) == 0 && !a.Prefix.set
+	return len(a.Tags) == 0 && !a.Prefix.set &&
+		a.ObjectSizeGreaterThan == 0 && a.ObjectSizeLessThan == 0
 }
 
 // Validate - validates the And field
 func (a And) Validate() error {
-	emptyPrefix := !a.Prefix.set
-	emptyTags := len(a.Tags) == 0
-
-	if emptyPrefix && emptyTags {
-		return nil
+	// > This is used in a Lifecycle Rule Filter to apply a logical AND to two or more predicates.
+	// ref: https://docs.aws.amazon.com/AmazonS3/latest/API/API_LifecycleRuleAndOperator.html
+	// i.e, predCount >= 2
+	var predCount int
+	if a.Prefix.set {
+		predCount++
+	}
+	predCount += len(a.Tags)
+	if a.ObjectSizeGreaterThan > 0 {
+		predCount++
+	}
+	if a.ObjectSizeLessThan > 0 {
+		predCount++
 	}
 
-	if emptyPrefix && !emptyTags || !emptyPrefix && emptyTags {
+	if predCount < 2 {
 		return errXMLNotWellFormed
 	}
 
@@ -56,6 +67,10 @@ func (a And) Validate() error {
 			return err
 		}
 	}
+
+	if a.ObjectSizeGreaterThan < 0 || a.ObjectSizeLessThan < 0 {
+		return errXMLNotWellFormed
+	}
 	return nil
 }
 
@@ -72,3 +87,19 @@ func (a And) ContainsDuplicateTag() bool {
 
 	return false
 }
+
+// BySize returns true when sz satisfies a
+// ObjectSizeLessThan/ObjectSizeGreaterthan or a logial AND of these predicates
+// Note: And combines size and other predicates like Tags, Prefix, etc. This
+// method applies exclusively to size predicates only.
+func (a And) BySize(sz int64) bool {
+	if a.ObjectSizeGreaterThan > 0 &&
+		sz <= a.ObjectSizeGreaterThan {
+		return false
+	}
+	if a.ObjectSizeLessThan > 0 &&
+		sz >= a.ObjectSizeLessThan {
+		return false
+	}
+	return true
+}

internal/bucket/lifecycle/filter.go

@@ -33,6 +33,9 @@ type Filter struct {
 
 	Prefix Prefix
 
+	ObjectSizeGreaterThan int64 `xml:"ObjectSizeGreaterThan,omitempty"`
+	ObjectSizeLessThan    int64 `xml:"ObjectSizeLessThan,omitempty"`
+
 	And    And
 	andSet bool
 
@@ -64,6 +67,17 @@ func (f Filter) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
 		if err := e.EncodeElement(f.Prefix, xml.StartElement{Name: xml.Name{Local: "Prefix"}}); err != nil {
 			return err
 		}
+
+		if f.ObjectSizeLessThan > 0 {
+			if err := e.EncodeElement(f.ObjectSizeLessThan, xml.StartElement{Name: xml.Name{Local: "ObjectSizeLessThan"}}); err != nil {
+				return err
+			}
+		}
+		if f.ObjectSizeGreaterThan > 0 {
+			if err := e.EncodeElement(f.ObjectSizeGreaterThan, xml.StartElement{Name: xml.Name{Local: "ObjectSizeGreaterThan"}}); err != nil {
+				return err
+			}
+		}
 	}
 
 	return e.EncodeToken(xml.EndElement{Name: start.Name})
@@ -104,6 +118,18 @@ func (f *Filter) UnmarshalXML(d *xml.Decoder, start xml.StartElement) (err error
 				}
 				f.Tag = tag
 				f.tagSet = true
+			case "ObjectSizeLessThan":
+				var sz int64
+				if err = d.DecodeElement(&sz, &se); err != nil {
+					return err
+				}
+				f.ObjectSizeLessThan = sz
+			case "ObjectSizeGreaterThan":
+				var sz int64
+				if err = d.DecodeElement(&sz, &se); err != nil {
+					return err
+				}
+				f.ObjectSizeGreaterThan = sz
 			default:
 				return errUnknownXMLTag
 			}
@@ -122,32 +148,64 @@ func (f Filter) Validate() error {
 	if f.IsEmpty() {
 		return errXMLNotWellFormed
 	}
-	// A Filter must have exactly one of Prefix, Tag, or And specified.
+	// A Filter must have exactly one of Prefix, Tag,
+	// ObjectSize{LessThan,GreaterThan} or And specified.
+	type predType uint8
+	const (
+		nonePred predType = iota
+		prefixPred
+		andPred
+		tagPred
+		sizeLtPred
+		sizeGtPred
+	)
+	var predCount int
+	var pType predType
 	if !f.And.isEmpty() {
+		pType = andPred
+		predCount++
+	}
 	if f.Prefix.set {
-			return errInvalidFilter
+		pType = prefixPred
+		predCount++
 	}
 	if !f.Tag.IsEmpty() {
+		pType = tagPred
+		predCount++
+	}
+	if f.ObjectSizeGreaterThan != 0 {
+		pType = sizeGtPred
+		predCount++
+	}
+	if f.ObjectSizeLessThan != 0 {
+		pType = sizeLtPred
+		predCount++
+	}
+	// Note: S3 supports empty <Filter></Filter>, so predCount == 0 is
+	// valid.
+	if predCount > 1 {
 		return errInvalidFilter
 	}
-		if err := f.And.Validate(); err != nil {
+
+	var err error
+	switch pType {
+	case nonePred:
+	// S3 supports empty <Filter></Filter>
+	case prefixPred:
+	case andPred:
+		err = f.And.Validate()
+	case tagPred:
+		err = f.Tag.Validate()
+	case sizeLtPred:
+		if f.ObjectSizeLessThan < 0 {
+			err = errXMLNotWellFormed
+		}
+	case sizeGtPred:
+		if f.ObjectSizeGreaterThan < 0 {
+			err = errXMLNotWellFormed
+		}
+	}
 	return err
-		}
-	}
-	if f.Prefix.set {
-		if !f.Tag.IsEmpty() {
-			return errInvalidFilter
-		}
-	}
-	if !f.Tag.IsEmpty() {
-		if f.Prefix.set {
-			return errInvalidFilter
-		}
-		if err := f.Tag.Validate(); err != nil {
-			return err
-		}
-	}
-	return nil
 }
 
 // TestTags tests if the object tags satisfy the Filter tags requirement,
@@ -190,3 +248,20 @@ func (f Filter) TestTags(userTags string) bool {
 	}
 	return false
 }
+
+// BySize returns true if sz satisifies one of ObjectSizeGreaterThan,
+// ObjectSizeLessThan predicates or a combination of them via And.
+func (f Filter) BySize(sz int64) bool {
+	if f.ObjectSizeGreaterThan > 0 &&
+		sz <= f.ObjectSizeGreaterThan {
+		return false
+	}
+	if f.ObjectSizeLessThan > 0 &&
+		sz >= f.ObjectSizeLessThan {
+		return false
+	}
+	if !f.And.isEmpty() {
+		return f.And.BySize(sz)
+	}
+	return true
+}

internal/bucket/lifecycle/filter_test.go

@@ -21,6 +21,8 @@ import (
 	"encoding/xml"
 	"fmt"
 	"testing"
+
+	"github.com/dustin/go-humanize"
 )
 
 // TestUnsupportedFilters checks if parsing Filter xml with
@@ -124,3 +126,118 @@ func TestUnsupportedFilters(t *testing.T) {
 		})
 	}
 }
+
+func TestObjectSizeFilters(t *testing.T) {
+	f1 := Filter{
+		set: true,
+		Prefix: Prefix{
+			string: "doc/",
+			set:    true,
+			Unused: struct{}{},
+		},
+		ObjectSizeGreaterThan: 100 * humanize.MiByte,
+		ObjectSizeLessThan:    100 * humanize.GiByte,
+	}
+	b, err := xml.Marshal(f1)
+	if err != nil {
+		t.Fatalf("Failed to marshal %v", f1)
+	}
+	var f2 Filter
+	err = xml.Unmarshal(b, &f2)
+	if err != nil {
+		t.Fatalf("Failed to unmarshal %s", string(b))
+	}
+	if f1.ObjectSizeLessThan != f2.ObjectSizeLessThan {
+		t.Fatalf("Expected %v but got %v", f1.ObjectSizeLessThan, f2.And.ObjectSizeLessThan)
+	}
+	if f1.ObjectSizeGreaterThan != f2.ObjectSizeGreaterThan {
+		t.Fatalf("Expected %v but got %v", f1.ObjectSizeGreaterThan, f2.And.ObjectSizeGreaterThan)
+	}
+
+	f1 = Filter{
+		set: true,
+		And: And{
+			ObjectSizeGreaterThan: 100 * humanize.MiByte,
+			ObjectSizeLessThan:    1 * humanize.GiByte,
+			Prefix:                Prefix{},
+		},
+		andSet: true,
+	}
+	b, err = xml.Marshal(f1)
+	if err != nil {
+		t.Fatalf("Failed to marshal %v", f1)
+	}
+	f2 = Filter{}
+	err = xml.Unmarshal(b, &f2)
+	if err != nil {
+		t.Fatalf("Failed to unmarshal %s", string(b))
+	}
+	if f1.And.ObjectSizeLessThan != f2.And.ObjectSizeLessThan {
+		t.Fatalf("Expected %v but got %v", f1.And.ObjectSizeLessThan, f2.And.ObjectSizeLessThan)
+	}
+	if f1.And.ObjectSizeGreaterThan != f2.And.ObjectSizeGreaterThan {
+		t.Fatalf("Expected %v but got %v", f1.And.ObjectSizeGreaterThan, f2.And.ObjectSizeGreaterThan)
+	}
+
+	fiGt := Filter{
+		ObjectSizeGreaterThan: 1 * humanize.MiByte,
+	}
+	fiLt := Filter{
+		ObjectSizeLessThan: 100 * humanize.MiByte,
+	}
+	fiLtAndGt := Filter{
+		And: And{
+			ObjectSizeGreaterThan: 1 * humanize.MiByte,
+			ObjectSizeLessThan:    100 * humanize.MiByte,
+		},
+	}
+
+	tests := []struct {
+		filter  Filter
+		objSize int64
+		want    bool
+	}{
+		{
+			filter:  fiLt,
+			objSize: 101 * humanize.MiByte,
+			want:    false,
+		},
+		{
+			filter:  fiLt,
+			objSize: 99 * humanize.MiByte,
+			want:    true,
+		},
+		{
+			filter:  fiGt,
+			objSize: 1*humanize.MiByte - 1,
+			want:    false,
+		},
+		{
+			filter:  fiGt,
+			objSize: 1*humanize.MiByte + 1,
+			want:    true,
+		},
+		{
+			filter:  fiLtAndGt,
+			objSize: 1*humanize.MiByte - 1,
+			want:    false,
+		},
+		{
+			filter:  fiLtAndGt,
+			objSize: 2 * humanize.MiByte,
+			want:    true,
+		},
+		{
+			filter:  fiLtAndGt,
+			objSize: 100*humanize.MiByte + 1,
+			want:    false,
+		},
+	}
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("Test %d", i+1), func(t *testing.T) {
+			if got := test.filter.BySize(test.objSize); got != test.want {
+				t.Fatalf("Expected %v but got %v", test.want, got)
+			}
+		})
+	}
+}

internal/bucket/lifecycle/lifecycle.go

@@ -282,6 +282,9 @@ func (lc Lifecycle) FilterRules(obj ObjectOpts) []Rule {
 		if !obj.DeleteMarker && !rule.Filter.TestTags(obj.UserTags) {
 			continue
 		}
+		if !obj.DeleteMarker && !rule.Filter.BySize(obj.Size) {
+			continue
+		}
 		rules = append(rules, rule)
 	}
 	return rules
@@ -293,6 +296,7 @@ type ObjectOpts struct {
 	Name             string
 	UserTags         string
 	ModTime          time.Time
+	Size             int64
 	VersionID        string
 	IsLatest         bool
 	DeleteMarker     bool

internal/bucket/lifecycle/lifecycle_test.go

@@ -28,6 +28,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/dustin/go-humanize"
 	"github.com/google/uuid"
 	xhttp "github.com/minio/minio/internal/http"
 )
@@ -1018,59 +1019,211 @@ func TestFilterAndSetPredictionHeaders(t *testing.T) {
 }
 
 func TestFilterRules(t *testing.T) {
-	lc := Lifecycle{
-		Rules: []Rule{
+	rules := []Rule{
 		{
 			ID:     "rule-1",
 			Status: "Enabled",
 			Filter: Filter{
+				set: true,
 				Tag: Tag{
 					Key:   "key1",
 					Value: "val1",
 				},
 			},
 			Expiration: Expiration{
+				set:  true,
 				Days: 1,
 			},
 		},
+		{
+			ID:     "rule-with-sz-lt",
+			Status: "Enabled",
+			Filter: Filter{
+				set:                true,
+				ObjectSizeLessThan: 100 * humanize.MiByte,
+			},
+			Expiration: Expiration{
+				set:  true,
+				Days: 1,
+			},
+		},
+		{
+			ID:     "rule-with-sz-gt",
+			Status: "Enabled",
+			Filter: Filter{
+				set:                   true,
+				ObjectSizeGreaterThan: 1 * humanize.MiByte,
+			},
+			Expiration: Expiration{
+				set:  true,
+				Days: 1,
+			},
+		},
+		{
+			ID:     "rule-with-sz-lt-and-tag",
+			Status: "Enabled",
+			Filter: Filter{
+				set: true,
+				And: And{
+					ObjectSizeLessThan: 100 * humanize.MiByte,
+					Tags: []Tag{
+						{
+							Key:   "key1",
+							Value: "val1",
+						},
+					},
+				},
+			},
+			Expiration: Expiration{
+				set:  true,
+				Days: 1,
+			},
+		},
+		{
+			ID:     "rule-with-sz-gt-and-tag",
+			Status: "Enabled",
+			Filter: Filter{
+				set: true,
+				And: And{
+					ObjectSizeGreaterThan: 1 * humanize.MiByte,
+					Tags: []Tag{
+						{
+							Key:   "key1",
+							Value: "val1",
+						},
+					},
+				},
+			},
+			Expiration: Expiration{
+				set:  true,
+				Days: 1,
+			},
+		},
+		{
+			ID:     "rule-with-sz-lt-and-gt",
+			Status: "Enabled",
+			Filter: Filter{
+				set: true,
+				And: And{
+					ObjectSizeGreaterThan: 101 * humanize.MiByte,
+					ObjectSizeLessThan:    200 * humanize.MiByte,
+				},
+			},
+			Expiration: Expiration{
+				set:  true,
+				Days: 1,
+			},
 		},
 	}
 	tests := []struct {
+		lc       Lifecycle
 		opts     ObjectOpts
-		wantRule string
+		hasRules bool
 	}{
 		{ // Delete marker should match filter without tags
+			lc: Lifecycle{
+				Rules: []Rule{
+					rules[0],
+				},
+			},
 			opts: ObjectOpts{
 				DeleteMarker: true,
 				IsLatest:     true,
 				Name:         "obj-1",
 			},
-			wantRule: "rule-1",
+			hasRules: true,
 		},
 		{ // PUT version with no matching tags
+			lc: Lifecycle{
+				Rules: []Rule{
+					rules[0],
+				},
+			},
 			opts: ObjectOpts{
 				IsLatest: true,
 				Name:     "obj-1",
+				Size:     1 * humanize.MiByte,
 			},
-			wantRule: "",
+			hasRules: false,
 		},
 		{ // PUT version with matching tags
+			lc: Lifecycle{
+				Rules: []Rule{
+					rules[0],
+				},
+			},
 			opts: ObjectOpts{
 				IsLatest: true,
 				UserTags: "key1=val1",
 				Name:     "obj-1",
+				Size:     2 * humanize.MiByte,
 			},
-			wantRule: "rule-1",
+			hasRules: true,
+		},
+		{ // PUT version with size based filters
+			lc: Lifecycle{
+				Rules: []Rule{
+					rules[1],
+					rules[2],
+					rules[3],
+					rules[4],
+					rules[5],
+				},
+			},
+			opts: ObjectOpts{
+				IsLatest: true,
+				UserTags: "key1=val1",
+				Name:     "obj-1",
+				Size:     1*humanize.MiByte - 1,
+			},
+			hasRules: true,
+		},
+		{ // PUT version with size based filters
+			lc: Lifecycle{
+				Rules: []Rule{
+					rules[1],
+					rules[2],
+					rules[3],
+					rules[4],
+					rules[5],
+				},
+			},
+			opts: ObjectOpts{
+				IsLatest: true,
+				Name:     "obj-1",
+				Size:     1*humanize.MiByte + 1,
+			},
+			hasRules: true,
+		},
+		{ // DEL version with size based filters
+			lc: Lifecycle{
+				Rules: []Rule{
+					rules[1],
+					rules[2],
+					rules[3],
+					rules[4],
+					rules[5],
+				},
+			},
+			opts: ObjectOpts{
+				DeleteMarker: true,
+				IsLatest:     true,
+				Name:         "obj-1",
+			},
+			hasRules: true,
 		},
 	}
 
 	for i, tc := range tests {
 		t.Run(fmt.Sprintf("test-%d", i+1), func(t *testing.T) {
-			rules := lc.FilterRules(tc.opts)
-			if tc.wantRule != "" && len(rules) == 0 {
-				t.Fatalf("%d: Expected rule match %s but none matched", i+1, tc.wantRule)
+			if err := tc.lc.Validate(); err != nil {
+				t.Fatalf("Lifecycle validation failed - %v", err)
 			}
-			if tc.wantRule == "" && len(rules) > 0 {
+			rules := tc.lc.FilterRules(tc.opts)
+			if tc.hasRules && len(rules) == 0 {
+				t.Fatalf("%d: Expected at least one rule to match but none matched", i+1)
+			}
+			if !tc.hasRules && len(rules) > 0 {
 				t.Fatalf("%d: Expected no rules to match but got matches %v", i+1, rules)
 			}
 		})

internal/bucket/lifecycle/rule_test.go

@@ -61,6 +61,50 @@ func TestInvalidRules(t *testing.T) {
 	                    </Rule>`,
 			expectedErr: errInvalidRuleStatus,
 		},
+		{ // Rule with negative values for ObjectSizeLessThan
+			inputXML: `<Rule>
+				<ID>negative-obj-size-less-than</ID>
+				<Filter><ObjectSizeLessThan>-1</ObjectSizeLessThan></Filter>
+				<Expiration>
+					<Days>365</Days>
+				</Expiration>
+                            <Status>Enabled</Status>
+	                    </Rule>`,
+			expectedErr: errXMLNotWellFormed,
+		},
+		{ // Rule with negative values for And>ObjectSizeLessThan
+			inputXML: `<Rule>
+				<ID>negative-and-obj-size-less-than</ID>
+				<Filter><And><ObjectSizeLessThan>-1</ObjectSizeLessThan></And></Filter>
+				<Expiration>
+					<Days>365</Days>
+				</Expiration>
+                            <Status>Enabled</Status>
+	                    </Rule>`,
+			expectedErr: errXMLNotWellFormed,
+		},
+		{ // Rule with negative values for ObjectSizeGreaterThan
+			inputXML: `<Rule>
+				<ID>negative-obj-size-greater-than</ID>
+				<Filter><ObjectSizeGreaterThan>-1</ObjectSizeGreaterThan></Filter>
+				<Expiration>
+					<Days>365</Days>
+				</Expiration>
+                            <Status>Enabled</Status>
+	                    </Rule>`,
+			expectedErr: errXMLNotWellFormed,
+		},
+		{ // Rule with negative values for And>ObjectSizeGreaterThan
+			inputXML: `<Rule>
+				<ID>negative-and-obj-size-greater-than</ID>
+				<Filter><And><ObjectSizeGreaterThan>-1</ObjectSizeGreaterThan></And></Filter>
+				<Expiration>
+					<Days>365</Days>
+				</Expiration>
+                            <Status>Enabled</Status>
+	                    </Rule>`,
+			expectedErr: errXMLNotWellFormed,
+		},
 	}
 
 	for i, tc := range invalidTestCases {