forked from TrueCloudLab/frostfs-node
161 lines
4.3 KiB
Go
161 lines
4.3 KiB
Go
package blobovniczatree
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobovnicza"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
|
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
var errRebuildInProgress = errors.New("rebuild is in progress, the operation cannot be performed")
|
|
|
|
func (b *Blobovniczas) Rebuild(ctx context.Context, prm common.RebuildPrm) (common.RebuildRes, error) {
|
|
if b.readOnly {
|
|
return common.RebuildRes{}, common.ErrReadOnly
|
|
}
|
|
|
|
b.rebuildGuard.Lock()
|
|
defer b.rebuildGuard.Unlock()
|
|
|
|
b.log.Debug(logs.BlobovniczaTreeCollectingDBToRebuild)
|
|
var res common.RebuildRes
|
|
dbsToMigrate, err := b.getDBsToRebuild(ctx)
|
|
if err != nil {
|
|
b.log.Warn(logs.BlobovniczaTreeCollectingDBToRebuildFailed, zap.Error(err))
|
|
return res, err
|
|
}
|
|
b.log.Info(logs.BlobovniczaTreeCollectingDBToRebuildSuccess, zap.Int("blobovniczas_to_rebuild", len(dbsToMigrate)))
|
|
for _, db := range dbsToMigrate {
|
|
b.log.Debug(logs.BlobovniczaTreeRebuildingBlobovnicza, zap.String("path", db))
|
|
movedObjects, err := b.rebuildDB(ctx, db, prm.MetaStorage)
|
|
res.ObjectsMoved += movedObjects
|
|
if err != nil {
|
|
b.log.Warn(logs.BlobovniczaTreeRebuildingBlobovniczaFailed, zap.String("path", db), zap.Uint64("moved_objects_count", movedObjects), zap.Error(err))
|
|
return res, err
|
|
}
|
|
b.log.Debug(logs.BlobovniczaTreeRebuildingBlobovniczaSuccess, zap.String("path", db), zap.Uint64("moved_objects_count", movedObjects))
|
|
res.FilesRemoved++
|
|
}
|
|
|
|
return res, nil
|
|
}
|
|
|
|
func (b *Blobovniczas) getDBsToRebuild(ctx context.Context) ([]string, error) {
|
|
dbsToMigrate := make(map[string]struct{})
|
|
if err := b.iterateExistingDBPaths(ctx, func(s string) (bool, error) {
|
|
dbsToMigrate[s] = struct{}{}
|
|
return false, nil
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := b.iterateSortedLeaves(ctx, nil, func(s string) (bool, error) {
|
|
delete(dbsToMigrate, s)
|
|
return false, nil
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
result := make([]string, 0, len(dbsToMigrate))
|
|
for db := range dbsToMigrate {
|
|
result = append(result, db)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (b *Blobovniczas) rebuildDB(ctx context.Context, path string, meta common.MetaStorage) (uint64, error) {
|
|
shDB := b.getBlobovnicza(path)
|
|
blz, err := shDB.Open()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
shDBClosed := false
|
|
defer func() {
|
|
if shDBClosed {
|
|
return
|
|
}
|
|
shDB.Close()
|
|
}()
|
|
|
|
migratedObjects, err := b.moveObjects(ctx, blz, meta)
|
|
if err != nil {
|
|
return migratedObjects, err
|
|
}
|
|
shDBClosed, err = b.dropDB(ctx, path, shDB)
|
|
return migratedObjects, err
|
|
}
|
|
|
|
func (b *Blobovniczas) moveObjects(ctx context.Context, blz *blobovnicza.Blobovnicza, meta common.MetaStorage) (uint64, error) {
|
|
var result uint64
|
|
|
|
var prm blobovnicza.IteratePrm
|
|
prm.DecodeAddresses()
|
|
prm.SetHandler(func(ie blobovnicza.IterationElement) error {
|
|
e := b.moveObject(ctx, ie.Address(), ie.ObjectData(), meta)
|
|
if e == nil {
|
|
result++
|
|
}
|
|
return e
|
|
})
|
|
|
|
_, err := blz.Iterate(ctx, prm)
|
|
return result, err
|
|
}
|
|
|
|
func (b *Blobovniczas) moveObject(ctx context.Context, addr oid.Address, data []byte, metaStore common.MetaStorage) error {
|
|
var pPrm common.PutPrm
|
|
pPrm.Address = addr
|
|
pPrm.RawData = data
|
|
pRes, err := b.Put(ctx, pPrm)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return metaStore.UpdateStorageID(ctx, addr, pRes.StorageID)
|
|
}
|
|
|
|
func (b *Blobovniczas) dropDB(ctx context.Context, path string, shDb *sharedDB) (bool, error) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return false, ctx.Err()
|
|
case <-time.After(b.waitBeforeDropDB): // to complete requests with old storage ID
|
|
}
|
|
|
|
b.dbCache.EvictAndMarkNonCached(path)
|
|
defer b.dbCache.RemoveFromNonCached(path)
|
|
|
|
b.dbFilesGuard.Lock()
|
|
defer b.dbFilesGuard.Unlock()
|
|
|
|
if err := shDb.CloseAndRemoveFile(); err != nil {
|
|
return false, err
|
|
}
|
|
b.commondbManager.CleanResources(path)
|
|
if err := b.dropDirectoryIfEmpty(filepath.Dir(path)); err != nil {
|
|
return true, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func (b *Blobovniczas) dropDirectoryIfEmpty(path string) error {
|
|
if path == "." {
|
|
return nil
|
|
}
|
|
|
|
sysPath := filepath.Join(b.rootPath, path)
|
|
entries, err := os.ReadDir(sysPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(entries) > 0 {
|
|
return nil
|
|
}
|
|
if err := os.Remove(sysPath); err != nil {
|
|
return err
|
|
}
|
|
return b.dropDirectoryIfEmpty(filepath.Dir(path))
|
|
}
|