mirror of
https://gitee.com/milvus-io/milvus.git
synced 2025-12-08 01:58:34 +08:00
fix garbage collector err handling (#18277)
Signed-off-by: xiaofan-luan <xiaofan.luan@zilliz.com>
This commit is contained in:
parent
eb2de5aa59
commit
6d82ef8c20
@ -25,6 +25,7 @@ import (
|
|||||||
"github.com/milvus-io/milvus/internal/log"
|
"github.com/milvus-io/milvus/internal/log"
|
||||||
"github.com/milvus-io/milvus/internal/proto/commonpb"
|
"github.com/milvus-io/milvus/internal/proto/commonpb"
|
||||||
"github.com/milvus-io/milvus/internal/proto/datapb"
|
"github.com/milvus-io/milvus/internal/proto/datapb"
|
||||||
|
"github.com/milvus-io/milvus/internal/storage"
|
||||||
"github.com/minio/minio-go/v7"
|
"github.com/minio/minio-go/v7"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
@ -112,11 +113,11 @@ func (gc *garbageCollector) close() {
|
|||||||
// scan load meta file info and compares OSS keys
|
// scan load meta file info and compares OSS keys
|
||||||
// if missing found, performs gc cleanup
|
// if missing found, performs gc cleanup
|
||||||
func (gc *garbageCollector) scan() {
|
func (gc *garbageCollector) scan() {
|
||||||
var v, m int
|
var total, valid, missing int
|
||||||
valid := gc.meta.ListSegmentFiles()
|
segmentFiles := gc.meta.ListSegmentFiles()
|
||||||
vm := make(map[string]struct{})
|
filesMap := make(map[string]struct{})
|
||||||
for _, k := range valid {
|
for _, k := range segmentFiles {
|
||||||
vm[k.GetLogPath()] = struct{}{}
|
filesMap[k.GetLogPath()] = struct{}{}
|
||||||
}
|
}
|
||||||
|
|
||||||
// walk only data cluster related prefixes
|
// walk only data cluster related prefixes
|
||||||
@ -131,30 +132,36 @@ func (gc *garbageCollector) scan() {
|
|||||||
Prefix: prefix,
|
Prefix: prefix,
|
||||||
Recursive: true,
|
Recursive: true,
|
||||||
}) {
|
}) {
|
||||||
_, has := vm[info.Key]
|
total++
|
||||||
|
_, has := filesMap[info.Key]
|
||||||
if has {
|
if has {
|
||||||
v++
|
valid++
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
// binlog path should consist of "/files/insertLog/collID/partID/segID/fieldID/fileName"
|
segmentID, err := storage.ParseSegmentIDByBinlog(info.Key)
|
||||||
segmentID, err := parseSegmentIDByBinlog(info.Key)
|
if err != nil {
|
||||||
if err == nil {
|
log.Error("parse segment id error", zap.String("infoKey", info.Key), zap.Error(err))
|
||||||
if gc.segRefer.HasSegmentLock(segmentID) {
|
continue
|
||||||
v++
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
m++
|
if gc.segRefer.HasSegmentLock(segmentID) {
|
||||||
|
valid++
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
missing++
|
||||||
// not found in meta, check last modified time exceeds tolerance duration
|
// not found in meta, check last modified time exceeds tolerance duration
|
||||||
if time.Since(info.LastModified) > gc.option.missingTolerance {
|
if time.Since(info.LastModified) > gc.option.missingTolerance {
|
||||||
// ignore error since it could be cleaned up next time
|
// ignore error since it could be cleaned up next time
|
||||||
removedKeys = append(removedKeys, info.Key)
|
removedKeys = append(removedKeys, info.Key)
|
||||||
_ = gc.option.cli.RemoveObject(context.TODO(), gc.option.bucketName, info.Key, minio.RemoveObjectOptions{})
|
err = gc.option.cli.RemoveObject(context.TODO(), gc.option.bucketName, info.Key, minio.RemoveObjectOptions{})
|
||||||
|
if err != nil {
|
||||||
|
log.Error("failed to remove object", zap.String("infoKey", info.Key), zap.Error(err))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
log.Info("scan result", zap.Int("valid", v), zap.Int("missing", m), zap.Strings("removed keys", removedKeys))
|
log.Info("scan file to do garbage collection", zap.Int("total", total),
|
||||||
|
zap.Int("valid", valid), zap.Int("missing", missing), zap.Strings("removed keys", removedKeys))
|
||||||
}
|
}
|
||||||
|
|
||||||
func (gc *garbageCollector) clearEtcd() {
|
func (gc *garbageCollector) clearEtcd() {
|
||||||
|
|||||||
@ -284,9 +284,9 @@ func initUtOSSEnv(bucket, root string, n int) (cli *minio.Client, inserts []stri
|
|||||||
content := []byte("test")
|
content := []byte("test")
|
||||||
for i := 0; i < n; i++ {
|
for i := 0; i < n; i++ {
|
||||||
reader := bytes.NewReader(content)
|
reader := bytes.NewReader(content)
|
||||||
token := path.Join(funcutil.RandomString(8), funcutil.RandomString(8), strconv.Itoa(i), funcutil.RandomString(8), funcutil.RandomString(8))
|
token := path.Join(funcutil.RandomString(8), strconv.Itoa(i), strconv.Itoa(i), funcutil.RandomString(8), funcutil.RandomString(8))
|
||||||
if i == 1 {
|
if i == 1 {
|
||||||
token = path.Join(funcutil.RandomString(8), funcutil.RandomString(8), strconv.Itoa(i), funcutil.RandomString(8))
|
token = path.Join(funcutil.RandomString(8), strconv.Itoa(i), strconv.Itoa(i), funcutil.RandomString(8))
|
||||||
}
|
}
|
||||||
// insert
|
// insert
|
||||||
filePath := path.Join(root, insertLogPrefix, token)
|
filePath := path.Join(root, insertLogPrefix, token)
|
||||||
|
|||||||
@ -19,8 +19,6 @@ package datacoord
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"strconv"
|
|
||||||
"strings"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/milvus-io/milvus/internal/proto/commonpb"
|
"github.com/milvus-io/milvus/internal/proto/commonpb"
|
||||||
@ -87,9 +85,3 @@ func getCompactTime(ctx context.Context, allocator allocator) (*compactTime, err
|
|||||||
// no expiration time
|
// no expiration time
|
||||||
return &compactTime{ttRetentionLogic, 0}, nil
|
return &compactTime{ttRetentionLogic, 0}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func parseSegmentIDByBinlog(path string) (UniqueID, error) {
|
|
||||||
// binlog path should consist of "files/insertLog/collID/partID/segID/fieldID/fileName"
|
|
||||||
keyStr := strings.Split(path, "/")
|
|
||||||
return strconv.ParseInt(keyStr[len(keyStr)-3], 10, 64)
|
|
||||||
}
|
|
||||||
|
|||||||
12
internal/storage/binlog_util.go
Normal file
12
internal/storage/binlog_util.go
Normal file
@ -0,0 +1,12 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
func ParseSegmentIDByBinlog(path string) (UniqueID, error) {
|
||||||
|
// binlog path should consist of "files/insertLog/collID/partID/segID/fieldID/fileName"
|
||||||
|
keyStr := strings.Split(path, "/")
|
||||||
|
return strconv.ParseInt(keyStr[len(keyStr)-3], 10, 64)
|
||||||
|
}
|
||||||
Loading…
x
Reference in New Issue
Block a user