feat(lc): support config file size for lifecycle rule. #1000135253

Signed-off-by: Victor1319 <zengxuewei@oppo.com>

lcnode/lc_scanner.go

@@ -399,6 +399,12 @@ func (s *LcScanner) handleFile(dentry *proto.ScanDentry) {
 		log.LogWarnf("handleFile InodeGet_ll err: %v, dentry: %+v", err, dentry)
 		return
 	}
+
+	if info != nil && info.Size < s.rule.MinSize() {
+		log.LogInfof("handleFile: %+v, minSize(%d) size(%v) no need to process", dentry, s.rule.MinSize(), info.Size)
+		return
+	}
+
 	op := s.inodeExpired(info, s.rule.Expiration, s.rule.Transitions)
 	dentry.Op = op
 	dentry.Size = info.Size
@@ -565,10 +571,12 @@ func (s *LcScanner) inodeExpired(inode *proto.InodeInfo, condE *proto.Expiration
 
 func expired(inode *proto.InodeInfo, now int64, days *int, date *time.Time) bool {
 	if days != nil && *days > 0 {
-		if inode.AccessTime.Before(inode.CreateTime) {
+		// Avoid the impact of time jitter between nodes
+		if inode.AccessTime.Add(time.Second * 10).Before(inode.CreateTime) {
 			log.LogWarnf("AccessTime before CreateTime, skip, inode: %+v, LeaseExpireTime(%v), AccessTime(%v), CreateTime(%v)", inode, inode.LeaseExpireTime, inode.AccessTime, inode.CreateTime)
 			return false
 		}
+
 		inodeTime := inode.AccessTime.Unix()
 		if useCreateTime {
 			inodeTime = inode.CreateTime.Unix()

lcnode/lc_scanner_test.go

@@ -86,4 +86,28 @@ func TestLcScanner(t *testing.T) {
 	require.Equal(t, int64(0), scanner.currentStat.ErrorMToHddNum)
 	require.Equal(t, int64(0), scanner.currentStat.ErrorMToBlobstoreNum)
 	require.Equal(t, int64(0), scanner.currentStat.ErrorReadDirNum)
+
+	dentry := &proto.ScanDentry{
+		Inode: 1,
+	}
+	scanner.rule = &proto.Rule{
+		Filter: &proto.Filter{
+			MinSize: 1024,
+		},
+	}
+	scanner.handleFile(dentry)
+
+	// expired(inode *proto.InodeInfo, now int64, days *int, date *time.Time)
+	inode := &proto.InodeInfo{
+		AccessTime: time.Now().Add(time.Second * -1),
+		CreateTime: time.Now(),
+	}
+	days := 1
+	now := time.Now().Unix() + 24*60*60 + 1
+	res := expired(inode, now, &days, nil)
+	require.True(t, res)
+
+	inode.AccessTime = time.Now().Add(time.Second * -12)
+	res = expired(inode, now, &days, nil)
+	require.False(t, res)
 }

master/api_service_test.go

@@ -53,6 +53,9 @@ const (
 	mds5Addr          = "127.0.0.1:9105"
 	mds6Addr          = "127.0.0.1:9106"
 	mds7Addr          = "127.0.0.1:9107"
+	mds1HddAddr       = "127.0.0.1:9111"
+	mds2HddAddr       = "127.0.0.1:9112"
+	mds3HddAddr       = "127.0.0.1:9113"
 
 	mms1Addr      = "127.0.0.1:8101"
 	mms2Addr      = "127.0.0.1:8102"
@@ -66,6 +69,7 @@ const (
 	testZone1     = "zone1"
 	testZone2     = "zone2"
 	testZone3     = "zone3"
+	testHddZone1  = "hdd_zone1"
 
 	mfs1Addr = "127.0.0.1:10501"
 	mfs2Addr = "127.0.0.1:10502"
@@ -164,6 +168,11 @@ func createDefaultMasterServerForTest() *Server {
 	mockDataServers = append(mockDataServers, addDataServer(mds4Addr, testZone2, defaultMediaType))
 	mockDataServers = append(mockDataServers, addDataServer(mds5Addr, testZone2, defaultMediaType))
 	mockDataServers = append(mockDataServers, addDataServer(mds6Addr, testZone2, defaultMediaType))
+	mockDataServers = append(mockDataServers,
+		addDataServer(mds1HddAddr, testHddZone1, proto.MediaType_HDD),
+		addDataServer(mds2HddAddr, testHddZone1, proto.MediaType_HDD),
+		addDataServer(mds3HddAddr, testHddZone1, proto.MediaType_HDD),
+	)
 
 	// add meta node
 	mockMetaServers = make([]*mocktest.MockMetaServer, 0)
@@ -583,6 +592,74 @@ func TestMarkDeleteVol(t *testing.T) {
 	}
 }
 
+func TestVolSetBucketLifecycle(t *testing.T) {
+	req := &createVolReq{
+		name:                "bktLifecycle",
+		owner:               "cfs",
+		dpSize:              11,
+		mpCount:             3,
+		dpReplicaNum:        3,
+		capacity:            300,
+		followerRead:        false,
+		authenticate:        false,
+		crossZone:           true,
+		zoneName:            "",
+		description:         "",
+		qosLimitArgs:        &qosArgs{},
+		volStorageClass:     defaultVolStorageClass,
+		allowedStorageClass: []uint32{defaultVolStorageClass, proto.StorageClass_Replica_HDD},
+	}
+
+	_, err := server.cluster.createVol(req)
+	if err != nil {
+		log.LogFlush()
+		t.FailNow()
+	}
+
+	days := 10
+
+	rule := proto.Rule{
+		ID:     "r1",
+		Status: proto.RuleEnabled,
+		Filter: &proto.Filter{
+			Prefix:  "test/",
+			MinSize: 1024,
+		},
+		Transitions: []*proto.Transition{
+			{
+				Days:         &days,
+				StorageClass: "HDD",
+			},
+		},
+	}
+
+	lc := &proto.LcConfiguration{
+		VolName: req.name,
+		Rules: []*proto.Rule{
+			&rule,
+		},
+	}
+
+	data, err := json.Marshal(lc)
+	require.NoError(t, err)
+
+	url := fmt.Sprintf("%s%s", hostAddr, proto.SetBucketLifecycle)
+	resp, err := http.Post(url, "application/json", bytes.NewBuffer(data))
+	require.NoError(t, err)
+
+	defer resp.Body.Close()
+
+	data1, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	t.Logf("data: %s, url %s", string(data1), url)
+
+	require.True(t, resp.StatusCode == http.StatusOK)
+
+	lc2 := server.cluster.GetBucketLifecycle(lc.VolName)
+	require.True(t, len(lc2.Rules) == 1)
+	require.True(t, lc2.Rules[0].MinSize() == lc.Rules[0].MinSize())
+}
+
 func TestSetVolCapacity(t *testing.T) {
 	setVolCapacity(600, proto.AdminVolExpand, t)
 	setVolCapacity(300, proto.AdminVolShrink, t)

proto/lifecycle.go

@@ -65,7 +65,8 @@ type Expiration struct {
 }
 
 type Filter struct {
-	Prefix string `json:"Prefix,omitempty" xml:"Prefix,omitempty" bson:"Prefix,omitempty"`
+	Prefix  string `json:"Prefix,omitempty" xml:"Prefix,omitempty" bson:"Prefix,omitempty"`
+	MinSize uint64 `json:"MinSize,omitempty" xml:"MinSize,omitempty" bson:"MinSize,omitempty"`
 }
 
 type Transition struct {
@@ -167,6 +168,13 @@ func (r *Rule) GetPrefix() string {
 	return prefix
 }
 
+func (r *Rule) MinSize() uint64 {
+	if r.Filter != nil {
+		return r.Filter.MinSize
+	}
+	return 0
+}
+
 var regexRuleId = regexp.MustCompile(`^[A-Za-z0-9.-]+$`)
 
 var ExpirationEnabled bool