mirror of
https://github.com/minio/minio.git
synced 2024-12-24 06:05:55 -05:00
nas: Clean stale background appended files (#14295)
When more than one gateway reads and writes from the same mount point and there is a load balancer pointing to those gateways. Each gateway will try to create its own temporary append file but fails to clear it later when not needed. This commit creates a routine that checks all upload IDs saved in multipart directory and remove any stale entry with the same upload id in the memory and in the temporary background append folder as well.
This commit is contained in:
parent
5ec57a9533
commit
4afbb89774
@ -24,6 +24,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"os"
|
"os"
|
||||||
|
"path/filepath"
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@ -35,6 +36,11 @@ import (
|
|||||||
"github.com/minio/pkg/trie"
|
"github.com/minio/pkg/trie"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
bgAppendsDirName = "bg-appends"
|
||||||
|
bgAppendsCleanupInterval = 10 * time.Minute
|
||||||
|
)
|
||||||
|
|
||||||
// Returns EXPORT/.minio.sys/multipart/SHA256/UPLOADID
|
// Returns EXPORT/.minio.sys/multipart/SHA256/UPLOADID
|
||||||
func (fs *FSObjects) getUploadIDDir(bucket, object, uploadID string) string {
|
func (fs *FSObjects) getUploadIDDir(bucket, object, uploadID string) string {
|
||||||
return pathJoin(fs.fsPath, minioMetaMultipartBucket, getSHA256Hash([]byte(pathJoin(bucket, object))), uploadID)
|
return pathJoin(fs.fsPath, minioMetaMultipartBucket, getSHA256Hash([]byte(pathJoin(bucket, object))), uploadID)
|
||||||
@ -74,7 +80,7 @@ func (fs *FSObjects) backgroundAppend(ctx context.Context, bucket, object, uploa
|
|||||||
file := fs.appendFileMap[uploadID]
|
file := fs.appendFileMap[uploadID]
|
||||||
if file == nil {
|
if file == nil {
|
||||||
file = &fsAppendFile{
|
file = &fsAppendFile{
|
||||||
filePath: pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, fmt.Sprintf("%s.%s", uploadID, mustGetUUID())),
|
filePath: pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, bgAppendsDirName, fmt.Sprintf("%s.%s", uploadID, mustGetUUID())),
|
||||||
}
|
}
|
||||||
fs.appendFileMap[uploadID] = file
|
fs.appendFileMap[uploadID] = file
|
||||||
}
|
}
|
||||||
@ -643,7 +649,7 @@ func (fs *FSObjects) CompleteMultipartUpload(ctx context.Context, bucket string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
appendFallback := true // In case background-append did not append the required parts.
|
appendFallback := true // In case background-append did not append the required parts.
|
||||||
appendFilePath := pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, fmt.Sprintf("%s.%s", uploadID, mustGetUUID()))
|
appendFilePath := pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, "bg-appends", fmt.Sprintf("%s.%s", uploadID, mustGetUUID()))
|
||||||
|
|
||||||
// Most of the times appendFile would already be fully appended by now. We call fs.backgroundAppend()
|
// Most of the times appendFile would already be fully appended by now. We call fs.backgroundAppend()
|
||||||
// to take care of the following corner case:
|
// to take care of the following corner case:
|
||||||
@ -843,58 +849,99 @@ func (fs *FSObjects) AbortMultipartUpload(ctx context.Context, bucket, object, u
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Return all uploads IDs with full path of each upload-id directory.
|
||||||
|
// Do not return an error as this is a lazy operation
|
||||||
|
func (fs *FSObjects) getAllUploadIDs(ctx context.Context) (result map[string]string) {
|
||||||
|
result = make(map[string]string)
|
||||||
|
|
||||||
|
entries, err := readDir(pathJoin(fs.fsPath, minioMetaMultipartBucket))
|
||||||
|
if err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
for _, entry := range entries {
|
||||||
|
uploadIDs, err := readDir(pathJoin(fs.fsPath, minioMetaMultipartBucket, entry))
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// Remove the trailing slash separator
|
||||||
|
for i := range uploadIDs {
|
||||||
|
uploadID := strings.TrimSuffix(uploadIDs[i], SlashSeparator)
|
||||||
|
result[uploadID] = pathJoin(fs.fsPath, minioMetaMultipartBucket, entry, uploadID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// Removes multipart uploads if any older than `expiry` duration
|
// Removes multipart uploads if any older than `expiry` duration
|
||||||
// on all buckets for every `cleanupInterval`, this function is
|
// on all buckets for every `cleanupInterval`, this function is
|
||||||
// blocking and should be run in a go-routine.
|
// blocking and should be run in a go-routine.
|
||||||
func (fs *FSObjects) cleanupStaleUploads(ctx context.Context) {
|
func (fs *FSObjects) cleanupStaleUploads(ctx context.Context) {
|
||||||
timer := time.NewTimer(globalAPIConfig.getStaleUploadsCleanupInterval())
|
expiryUploadsTimer := time.NewTimer(globalAPIConfig.getStaleUploadsCleanupInterval())
|
||||||
defer timer.Stop()
|
defer expiryUploadsTimer.Stop()
|
||||||
|
|
||||||
|
bgAppendTmpCleaner := time.NewTimer(bgAppendsCleanupInterval)
|
||||||
|
defer bgAppendTmpCleaner.Stop()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
case <-timer.C:
|
case <-bgAppendTmpCleaner.C:
|
||||||
// Reset for the next interval
|
bgAppendTmpCleaner.Reset(bgAppendsCleanupInterval)
|
||||||
timer.Reset(globalAPIConfig.getStaleUploadsCleanupInterval())
|
|
||||||
|
|
||||||
expiry := globalAPIConfig.getStaleUploadsExpiry()
|
foundUploadIDs := fs.getAllUploadIDs(ctx)
|
||||||
|
|
||||||
now := time.Now()
|
// Remove background append map from the memory
|
||||||
entries, err := readDir(pathJoin(fs.fsPath, minioMetaMultipartBucket))
|
fs.appendFileMapMu.Lock()
|
||||||
|
for uploadID := range fs.appendFileMap {
|
||||||
|
_, ok := foundUploadIDs[uploadID]
|
||||||
|
if !ok {
|
||||||
|
delete(fs.appendFileMap, uploadID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
fs.appendFileMapMu.Unlock()
|
||||||
|
|
||||||
|
// Remove background appends file from the disk
|
||||||
|
bgAppendsDir := pathJoin(fs.fsPath, minioMetaTmpBucket, fs.fsUUID, bgAppendsDirName)
|
||||||
|
entries, err := readDir(bgAppendsDir)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
break
|
||||||
}
|
}
|
||||||
for _, entry := range entries {
|
for _, entry := range entries {
|
||||||
uploadIDs, err := readDir(pathJoin(fs.fsPath, minioMetaMultipartBucket, entry))
|
uploadID := strings.Split(entry, ".")[0]
|
||||||
|
_, ok := foundUploadIDs[uploadID]
|
||||||
|
if !ok {
|
||||||
|
fsRemoveFile(ctx, pathJoin(bgAppendsDir, entry))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
case <-expiryUploadsTimer.C:
|
||||||
|
// Reset for the next interval
|
||||||
|
expiryUploadsTimer.Reset(globalAPIConfig.getStaleUploadsCleanupInterval())
|
||||||
|
|
||||||
|
expiry := globalAPIConfig.getStaleUploadsExpiry()
|
||||||
|
now := time.Now()
|
||||||
|
|
||||||
|
uploadIDs := fs.getAllUploadIDs(ctx)
|
||||||
|
|
||||||
|
for uploadID, path := range uploadIDs {
|
||||||
|
fi, err := fsStatDir(ctx, path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
if now.Sub(fi.ModTime()) > expiry {
|
||||||
|
fsRemoveAll(ctx, path)
|
||||||
|
// Remove upload ID parent directory if empty
|
||||||
|
fsRemoveDir(ctx, filepath.Base(path))
|
||||||
|
|
||||||
// Remove the trailing slash separator
|
// Remove uploadID from the append file map and its corresponding temporary file
|
||||||
for i := range uploadIDs {
|
fs.appendFileMapMu.Lock()
|
||||||
uploadIDs[i] = strings.TrimSuffix(uploadIDs[i], SlashSeparator)
|
bgAppend, ok := fs.appendFileMap[uploadID]
|
||||||
}
|
if ok {
|
||||||
|
_ = fsRemoveFile(ctx, bgAppend.filePath)
|
||||||
for _, uploadID := range uploadIDs {
|
delete(fs.appendFileMap, uploadID)
|
||||||
fi, err := fsStatDir(ctx, pathJoin(fs.fsPath, minioMetaMultipartBucket, entry, uploadID))
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if now.Sub(fi.ModTime()) > expiry {
|
|
||||||
fsRemoveAll(ctx, pathJoin(fs.fsPath, minioMetaMultipartBucket, entry, uploadID))
|
|
||||||
// It is safe to ignore any directory not empty error (in case there were multiple uploadIDs on the same object)
|
|
||||||
fsRemoveDir(ctx, pathJoin(fs.fsPath, minioMetaMultipartBucket, entry))
|
|
||||||
|
|
||||||
// Remove uploadID from the append file map and its corresponding temporary file
|
|
||||||
fs.appendFileMapMu.Lock()
|
|
||||||
bgAppend, ok := fs.appendFileMap[uploadID]
|
|
||||||
if ok {
|
|
||||||
_ = fsRemoveFile(ctx, bgAppend.filePath)
|
|
||||||
delete(fs.appendFileMap, uploadID)
|
|
||||||
}
|
|
||||||
fs.appendFileMapMu.Unlock()
|
|
||||||
}
|
}
|
||||||
|
fs.appendFileMapMu.Unlock()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -106,6 +106,10 @@ func initMetaVolumeFS(fsPath, fsUUID string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := os.MkdirAll(pathJoin(metaTmpPath, bgAppendsDirName), 0o777); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
if err := os.MkdirAll(pathJoin(fsPath, dataUsageBucket), 0o777); err != nil {
|
if err := os.MkdirAll(pathJoin(fsPath, dataUsageBucket), 0o777); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user