storj/storage/filestore/dir.go

821 lines
26 KiB
Go

// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package filestore
import (
"context"
"encoding/base32"
"io"
"io/ioutil"
"math"
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/zeebo/errs"
"go.uber.org/zap"
"storj.io/storj/storage"
)
const (
blobPermission = 0600
dirPermission = 0700
v0PieceFileSuffix = ""
v1PieceFileSuffix = ".sj1"
unknownPieceFileSuffix = "/..error_unknown_format../"
)
var pathEncoding = base32.NewEncoding("abcdefghijklmnopqrstuvwxyz234567").WithPadding(base32.NoPadding)
// Dir represents single folder for storing blobs
type Dir struct {
log *zap.Logger
path string
mu sync.Mutex
deleteQueue []string
trashnow func() time.Time // the function used by trash to determine "now"
}
// NewDir returns folder for storing blobs
func NewDir(log *zap.Logger, path string) (*Dir, error) {
dir := &Dir{
log: log,
path: path,
trashnow: time.Now,
}
return dir, errs.Combine(
os.MkdirAll(dir.blobsdir(), dirPermission),
os.MkdirAll(dir.tempdir(), dirPermission),
os.MkdirAll(dir.garbagedir(), dirPermission),
os.MkdirAll(dir.trashdir(), dirPermission),
)
}
// Path returns the directory path
func (dir *Dir) Path() string { return dir.path }
// blobsdir is the sub-directory containing the blobs
func (dir *Dir) blobsdir() string { return filepath.Join(dir.path, "blobs") }
// tempdir is used for temp files prior to being moved into blobsdir
func (dir *Dir) tempdir() string { return filepath.Join(dir.path, "temp") }
// garbagedir contains files that failed to delete but should be deleted
func (dir *Dir) garbagedir() string { return filepath.Join(dir.path, "garbage") }
// trashdir contains files staged for deletion for a period of time
func (dir *Dir) trashdir() string { return filepath.Join(dir.path, "trash") }
// CreateTemporaryFile creates a preallocated temporary file in the temp directory
// prealloc preallocates file to make writing faster
func (dir *Dir) CreateTemporaryFile(ctx context.Context, prealloc int64) (_ *os.File, err error) {
const preallocLimit = 5 << 20 // 5 MB
if prealloc > preallocLimit {
prealloc = preallocLimit
}
file, err := ioutil.TempFile(dir.tempdir(), "blob-*.partial")
if err != nil {
return nil, err
}
if prealloc >= 0 {
if err := file.Truncate(prealloc); err != nil {
return nil, errs.Combine(err, file.Close())
}
}
return file, nil
}
// DeleteTemporary deletes a temporary file
func (dir *Dir) DeleteTemporary(ctx context.Context, file *os.File) (err error) {
defer mon.Task()(&ctx)(&err)
closeErr := file.Close()
return errs.Combine(closeErr, os.Remove(file.Name()))
}
// blobToBasePath converts a blob reference to a filepath in permanent storage. This may not be the
// entire path; blobPathForFormatVersion() must also be used. This is a separate call because this
// part of the filepath is constant, and blobPathForFormatVersion may need to be called multiple
// times with different storage.FormatVersion values.
func (dir *Dir) blobToBasePath(ref storage.BlobRef) (string, error) {
return dir.refToDirPath(ref, dir.blobsdir())
}
// refToDirPath converts a blob reference to a filepath in the specified sub-directory.
func (dir *Dir) refToDirPath(ref storage.BlobRef, subDir string) (string, error) {
if !ref.IsValid() {
return "", storage.ErrInvalidBlobRef.New("")
}
namespace := pathEncoding.EncodeToString(ref.Namespace)
key := pathEncoding.EncodeToString(ref.Key)
if len(key) < 3 {
// ensure we always have enough characters to split [:2] and [2:]
key = "11" + key
}
return filepath.Join(subDir, namespace, key[:2], key[2:]), nil
}
// fileConfirmedInTrash returns true if it is able to confirm the file is in
// the trash. On errors, or if the file is not in the trash, it returns false.
func (dir *Dir) fileConfirmedInTrash(ctx context.Context, ref storage.BlobRef, formatVer storage.FormatVersion) bool {
trashBasePath, err := dir.refToDirPath(ref, dir.trashdir())
if err != nil {
return false
}
trashVerPath := blobPathForFormatVersion(trashBasePath, formatVer)
_, err = os.Stat(trashVerPath)
return err == nil
}
// blobPathForFormatVersion adjusts a bare blob path (as might have been generated by a call to
// blobToBasePath()) to what it should be for the given storage format version.
func blobPathForFormatVersion(path string, formatVersion storage.FormatVersion) string {
switch formatVersion {
case FormatV0:
return path + v0PieceFileSuffix
case FormatV1:
return path + v1PieceFileSuffix
}
return path + unknownPieceFileSuffix
}
// blobToGarbagePath converts a blob reference to a filepath in transient
// storage. The files in garbage are deleted on an interval (in case the
// initial deletion didn't work for some reason).
func (dir *Dir) blobToGarbagePath(ref storage.BlobRef) string {
var name []byte
name = append(name, ref.Namespace...)
name = append(name, ref.Key...)
return filepath.Join(dir.garbagedir(), pathEncoding.EncodeToString(name))
}
// Commit commits the temporary file to permanent storage.
func (dir *Dir) Commit(ctx context.Context, file *os.File, ref storage.BlobRef, formatVersion storage.FormatVersion) (err error) {
defer mon.Task()(&ctx)(&err)
position, seekErr := file.Seek(0, io.SeekCurrent)
truncErr := file.Truncate(position)
syncErr := file.Sync()
chmodErr := os.Chmod(file.Name(), blobPermission)
closeErr := file.Close()
if seekErr != nil || truncErr != nil || syncErr != nil || chmodErr != nil || closeErr != nil {
removeErr := os.Remove(file.Name())
return errs.Combine(seekErr, truncErr, syncErr, chmodErr, closeErr, removeErr)
}
path, err := dir.blobToBasePath(ref)
if err != nil {
removeErr := os.Remove(file.Name())
return errs.Combine(err, removeErr)
}
path = blobPathForFormatVersion(path, formatVersion)
mkdirErr := os.MkdirAll(filepath.Dir(path), dirPermission)
if os.IsExist(mkdirErr) {
mkdirErr = nil
}
if mkdirErr != nil {
removeErr := os.Remove(file.Name())
return errs.Combine(mkdirErr, removeErr)
}
renameErr := rename(file.Name(), path)
if renameErr != nil {
removeErr := os.Remove(file.Name())
return errs.Combine(renameErr, removeErr)
}
return nil
}
// Open opens the file with the specified ref. It may need to check in more than one location in
// order to find the blob, if it was stored with an older version of the storage node software.
// In cases where the storage format version of a blob is already known, OpenWithStorageFormat()
// will generally be a better choice.
func (dir *Dir) Open(ctx context.Context, ref storage.BlobRef) (_ *os.File, _ storage.FormatVersion, err error) {
defer mon.Task()(&ctx)(&err)
path, err := dir.blobToBasePath(ref)
if err != nil {
return nil, FormatV0, err
}
for formatVer := MaxFormatVersionSupported; formatVer >= MinFormatVersionSupported; formatVer-- {
vPath := blobPathForFormatVersion(path, formatVer)
file, err := openFileReadOnly(vPath, blobPermission)
if err == nil {
return file, formatVer, nil
}
if os.IsNotExist(err) {
// Check and monitor if the file is in the trash
if dir.fileConfirmedInTrash(ctx, ref, formatVer) {
monFileInTrash.Mark(1)
}
} else {
return nil, FormatV0, Error.New("unable to open %q: %v", vPath, err)
}
}
return nil, FormatV0, os.ErrNotExist
}
// OpenWithStorageFormat opens an already-located blob file with a known storage format version,
// which avoids the potential need to search through multiple storage formats to find the blob.
func (dir *Dir) OpenWithStorageFormat(ctx context.Context, ref storage.BlobRef, formatVer storage.FormatVersion) (_ *os.File, err error) {
defer mon.Task()(&ctx)(&err)
path, err := dir.blobToBasePath(ref)
if err != nil {
return nil, err
}
vPath := blobPathForFormatVersion(path, formatVer)
file, err := openFileReadOnly(vPath, blobPermission)
if err == nil {
return file, nil
}
if os.IsNotExist(err) {
// Check and monitor if the file is in the trash
if dir.fileConfirmedInTrash(ctx, ref, formatVer) {
monFileInTrash.Mark(1)
}
return nil, err
}
return nil, Error.New("unable to open %q: %v", vPath, err)
}
// Stat looks up disk metadata on the blob file. It may need to check in more than one location
// in order to find the blob, if it was stored with an older version of the storage node software.
// In cases where the storage format version of a blob is already known, StatWithStorageFormat()
// will generally be a better choice.
func (dir *Dir) Stat(ctx context.Context, ref storage.BlobRef) (_ storage.BlobInfo, err error) {
defer mon.Task()(&ctx)(&err)
path, err := dir.blobToBasePath(ref)
if err != nil {
return nil, err
}
for formatVer := MaxFormatVersionSupported; formatVer >= MinFormatVersionSupported; formatVer-- {
vPath := blobPathForFormatVersion(path, formatVer)
stat, err := os.Stat(vPath)
if err == nil {
return newBlobInfo(ref, vPath, stat, formatVer), nil
}
if !os.IsNotExist(err) {
return nil, Error.New("unable to stat %q: %v", vPath, err)
}
}
return nil, os.ErrNotExist
}
// StatWithStorageFormat looks up disk metadata on the blob file with the given storage format
// version. This avoids the need for checking for the file in multiple different storage format
// types.
func (dir *Dir) StatWithStorageFormat(ctx context.Context, ref storage.BlobRef, formatVer storage.FormatVersion) (_ storage.BlobInfo, err error) {
defer mon.Task()(&ctx)(&err)
path, err := dir.blobToBasePath(ref)
if err != nil {
return nil, err
}
vPath := blobPathForFormatVersion(path, formatVer)
stat, err := os.Stat(vPath)
if err == nil {
return newBlobInfo(ref, vPath, stat, formatVer), nil
}
if os.IsNotExist(err) {
return nil, err
}
return nil, Error.New("unable to stat %q: %v", vPath, err)
}
// Trash moves the piece specified by ref to the trashdir for every format version
func (dir *Dir) Trash(ctx context.Context, ref storage.BlobRef) (err error) {
defer mon.Task()(&ctx)(&err)
return dir.iterateStorageFormatVersions(ctx, ref, dir.TrashWithStorageFormat)
}
// TrashWithStorageFormat moves the piece specified by ref to the trashdir for the specified format version
func (dir *Dir) TrashWithStorageFormat(ctx context.Context, ref storage.BlobRef, formatVer storage.FormatVersion) (err error) {
// Ensure trashdir exists so that we know any os.IsNotExist errors below
// are not from a missing trash dir
_, err = os.Stat(dir.trashdir())
if err != nil {
return err
}
blobsBasePath, err := dir.blobToBasePath(ref)
if err != nil {
return err
}
blobsVerPath := blobPathForFormatVersion(blobsBasePath, formatVer)
trashBasePath, err := dir.refToDirPath(ref, dir.trashdir())
if err != nil {
return err
}
trashVerPath := blobPathForFormatVersion(trashBasePath, formatVer)
// ensure the dirs exist for trash path
err = os.MkdirAll(filepath.Dir(trashVerPath), dirPermission)
if err != nil && !os.IsExist(err) {
return err
}
// Change mtime to now. This allows us to check the mtime to know how long
// the file has been in the trash. If the file is restored this may make it
// take longer to be trashed again, but the simplicity is worth the
// trade-off.
//
// We change the mtime prior to moving the file so that if this call fails
// the file will not be in the trash with an unmodified mtime, which could
// result in its permanent deletion too soon.
now := dir.trashnow()
err = os.Chtimes(blobsVerPath, now, now)
if os.IsNotExist(err) {
return nil
}
if err != nil {
return err
}
// move to trash
err = rename(blobsVerPath, trashVerPath)
if os.IsNotExist(err) {
// no piece at that path; either it has a different storage format
// version or there was a concurrent call. (This function is expected
// by callers to return a nil error in the case of concurrent calls.)
return nil
}
return err
}
// ReplaceTrashnow is a helper for tests to replace the trashnow function used
// when moving files to the trash
func (dir *Dir) ReplaceTrashnow(trashnow func() time.Time) {
dir.trashnow = trashnow
}
// RestoreTrash moves every piece in the trash folder back into blobsdir
func (dir *Dir) RestoreTrash(ctx context.Context, namespace []byte) (keysRestored [][]byte, err error) {
err = dir.walkNamespaceInPath(ctx, namespace, dir.trashdir(), func(info storage.BlobInfo) error {
blobsBasePath, err := dir.blobToBasePath(info.BlobRef())
if err != nil {
return err
}
blobsVerPath := blobPathForFormatVersion(blobsBasePath, info.StorageFormatVersion())
trashBasePath, err := dir.refToDirPath(info.BlobRef(), dir.trashdir())
if err != nil {
return err
}
trashVerPath := blobPathForFormatVersion(trashBasePath, info.StorageFormatVersion())
// ensure the dirs exist for blobs path
err = os.MkdirAll(filepath.Dir(blobsVerPath), dirPermission)
if err != nil && !os.IsExist(err) {
return err
}
// move back to blobsdir
err = rename(trashVerPath, blobsVerPath)
if os.IsNotExist(err) {
// no piece at that path; either it has a different storage format
// version or there was a concurrent call. (This function is expected
// by callers to return a nil error in the case of concurrent calls.)
return nil
}
if err != nil {
return err
}
keysRestored = append(keysRestored, info.BlobRef().Key)
return nil
})
return keysRestored, err
}
// EmptyTrash walks the trash files for the given namespace and deletes any
// file whose mtime is older than trashedBefore. The mtime is modified when
// Trash is called.
func (dir *Dir) EmptyTrash(ctx context.Context, namespace []byte, trashedBefore time.Time) (bytesEmptied int64, deletedKeys [][]byte, err error) {
defer mon.Task()(&ctx)(&err)
err = dir.walkNamespaceInPath(ctx, namespace, dir.trashdir(), func(blobInfo storage.BlobInfo) error {
fileInfo, err := blobInfo.Stat(ctx)
if err != nil {
return err
}
mtime := fileInfo.ModTime()
if mtime.Before(trashedBefore) {
err = dir.deleteWithStorageFormatInPath(ctx, dir.trashdir(), blobInfo.BlobRef(), blobInfo.StorageFormatVersion())
if err != nil {
return err
}
deletedKeys = append(deletedKeys, blobInfo.BlobRef().Key)
bytesEmptied += fileInfo.Size()
}
return nil
})
if err != nil {
return 0, nil, err
}
return bytesEmptied, deletedKeys, nil
}
// iterateStorageFormatVersions executes f for all storage format versions,
// starting with the oldest format version. It is more likely, in the general
// case, that we will find the piece with the newest format version instead,
// but if we iterate backward here then we run the risk of a race condition:
// the piece might have existed with _SomeOldVer before the call, and could
// then have been updated atomically with _MaxVer concurrently while we were
// iterating. If we iterate _forwards_, this race should not occur because it
// is assumed that pieces are never rewritten with an _older_ storage format
// version.
//
// f will be executed for every storage formate version regardless of the
// result, and will aggregate errors into a single returned error
func (dir *Dir) iterateStorageFormatVersions(ctx context.Context, ref storage.BlobRef, f func(ctx context.Context, ref storage.BlobRef, i storage.FormatVersion) error) (err error) {
defer mon.Task()(&ctx)(&err)
var combinedErrors errs.Group
for i := MinFormatVersionSupported; i <= MaxFormatVersionSupported; i++ {
combinedErrors.Add(f(ctx, ref, i))
}
return combinedErrors.Err()
}
// Delete deletes blobs with the specified ref (in all supported storage formats).
//
// It doesn't return an error if the blob is not found for any reason or it
// cannot be deleted at this moment and it's delayed.
func (dir *Dir) Delete(ctx context.Context, ref storage.BlobRef) (err error) {
defer mon.Task()(&ctx)(&err)
return dir.iterateStorageFormatVersions(ctx, ref, dir.DeleteWithStorageFormat)
}
// DeleteWithStorageFormat deletes the blob with the specified ref for one
// specific format version. The method tries the following strategies, in order
// of preference until one succeeds:
//
// * moves the blob to garbage dir.
// * directly deletes the blob.
// * push the blobs to queue for retrying later.
//
// It doesn't return an error if the piece isn't found for any reason.
func (dir *Dir) DeleteWithStorageFormat(ctx context.Context, ref storage.BlobRef, formatVer storage.FormatVersion) (err error) {
defer mon.Task()(&ctx)(&err)
return dir.deleteWithStorageFormatInPath(ctx, dir.blobsdir(), ref, formatVer)
}
// DeleteNamespace deletes blobs folder for a specific namespace.
func (dir *Dir) DeleteNamespace(ctx context.Context, ref []byte) (err error) {
defer mon.Task()(&ctx)(&err)
return dir.deleteNamespace(ctx, dir.blobsdir(), ref)
}
func (dir *Dir) deleteWithStorageFormatInPath(ctx context.Context, path string, ref storage.BlobRef, formatVer storage.FormatVersion) (err error) {
defer mon.Task()(&ctx)(&err)
// Ensure garbage dir exists so that we know any os.IsNotExist errors below
// are not from a missing garbage dir
_, err = os.Stat(dir.garbagedir())
if err != nil {
return err
}
pathBase, err := dir.refToDirPath(ref, path)
if err != nil {
return err
}
garbagePath := dir.blobToGarbagePath(ref)
verPath := blobPathForFormatVersion(pathBase, formatVer)
// move to garbage folder, this is allowed for some OS-es
moveErr := rename(verPath, garbagePath)
if os.IsNotExist(moveErr) {
// no piece at that path; either it has a different storage format
// version or there was a concurrent delete. (this function is expected
// by callers to return a nil error in the case of concurrent deletes.)
return nil
}
if moveErr != nil {
// piece could not be moved into the garbage dir; we'll try removing it
// directly
garbagePath = verPath
}
// try removing the file
err = os.Remove(garbagePath)
// ignore concurrent deletes
if os.IsNotExist(err) {
// something is happening at the same time as this; possibly a
// concurrent delete, or possibly a rewrite of the blob.
return nil
}
// the remove may have failed because of an open file handle. put it in a
// queue to be retried later.
if err != nil {
dir.mu.Lock()
dir.deleteQueue = append(dir.deleteQueue, garbagePath)
dir.mu.Unlock()
mon.Event("delete_deferred_to_queue")
}
// ignore is-busy errors, they are still in the queue but no need to notify
if isBusy(err) {
err = nil
}
return err
}
// deleteNamespace deletes folder with everything inside.
func (dir *Dir) deleteNamespace(ctx context.Context, path string, ref []byte) (err error) {
defer mon.Task()(&ctx)(&err)
namespace := pathEncoding.EncodeToString(ref)
folderPath := filepath.Join(path, namespace)
err = os.RemoveAll(folderPath)
return err
}
// GarbageCollect collects files that are pending deletion.
func (dir *Dir) GarbageCollect(ctx context.Context) (err error) {
defer mon.Task()(&ctx)(&err)
offset := int(math.MaxInt32)
// limited deletion loop to avoid blocking `Delete` for too long
for offset >= 0 {
dir.mu.Lock()
limit := 100
if offset >= len(dir.deleteQueue) {
offset = len(dir.deleteQueue) - 1
}
for offset >= 0 && limit > 0 {
path := dir.deleteQueue[offset]
err := os.Remove(path)
if os.IsNotExist(err) {
err = nil
}
if err == nil {
dir.deleteQueue = append(dir.deleteQueue[:offset], dir.deleteQueue[offset+1:]...)
}
offset--
limit--
}
dir.mu.Unlock()
}
// remove anything left in the garbagedir
_ = removeAllContent(ctx, dir.garbagedir())
return nil
}
const nameBatchSize = 1024
// ListNamespaces finds all known namespace IDs in use in local storage. They are not
// guaranteed to contain any blobs.
func (dir *Dir) ListNamespaces(ctx context.Context) (ids [][]byte, err error) {
defer mon.Task()(&ctx)(&err)
return dir.listNamespacesInPath(ctx, dir.blobsdir())
}
func (dir *Dir) listNamespacesInPath(ctx context.Context, path string) (ids [][]byte, err error) {
defer mon.Task()(&ctx)(&err)
openDir, err := os.Open(path)
if err != nil {
return nil, err
}
defer func() { err = errs.Combine(err, openDir.Close()) }()
for {
dirNames, err := openDir.Readdirnames(nameBatchSize)
if err != nil {
if err == io.EOF || os.IsNotExist(err) {
return ids, nil
}
return ids, err
}
if len(dirNames) == 0 {
return ids, nil
}
for _, name := range dirNames {
namespace, err := pathEncoding.DecodeString(name)
if err != nil {
// just an invalid directory entry, and not a namespace. probably
// don't need to pass on this error
continue
}
ids = append(ids, namespace)
}
}
}
// WalkNamespace executes walkFunc for each locally stored blob, stored with storage format V1 or
// greater, in the given namespace. If walkFunc returns a non-nil error, WalkNamespace will stop
// iterating and return the error immediately. The ctx parameter is intended specifically to allow
// canceling iteration early.
func (dir *Dir) WalkNamespace(ctx context.Context, namespace []byte, walkFunc func(storage.BlobInfo) error) (err error) {
defer mon.Task()(&ctx)(&err)
return dir.walkNamespaceInPath(ctx, namespace, dir.blobsdir(), walkFunc)
}
func (dir *Dir) walkNamespaceInPath(ctx context.Context, namespace []byte, path string, walkFunc func(storage.BlobInfo) error) (err error) {
defer mon.Task()(&ctx)(&err)
namespaceDir := pathEncoding.EncodeToString(namespace)
nsDir := filepath.Join(path, namespaceDir)
openDir, err := os.Open(nsDir)
if err != nil {
if os.IsNotExist(err) {
// job accomplished: there are no blobs in this namespace!
return nil
}
return err
}
defer func() { err = errs.Combine(err, openDir.Close()) }()
for {
// check for context done both before and after our readdir() call
if err := ctx.Err(); err != nil {
return err
}
subdirNames, err := openDir.Readdirnames(nameBatchSize)
if err != nil {
if err == io.EOF || os.IsNotExist(err) {
return nil
}
return err
}
if len(subdirNames) == 0 {
return nil
}
if err := ctx.Err(); err != nil {
return err
}
for _, keyPrefix := range subdirNames {
if len(keyPrefix) != 2 {
// just an invalid subdir; could be garbage of many kinds. probably
// don't need to pass on this error
continue
}
err := walkNamespaceWithPrefix(ctx, dir.log, namespace, nsDir, keyPrefix, walkFunc)
if err != nil {
return err
}
}
}
}
func decodeBlobInfo(namespace []byte, keyPrefix, keyDir string, keyInfo os.FileInfo) (info storage.BlobInfo, ok bool) {
blobFileName := keyInfo.Name()
encodedKey := keyPrefix + blobFileName
formatVer := FormatV0
if strings.HasSuffix(blobFileName, v1PieceFileSuffix) {
formatVer = FormatV1
encodedKey = encodedKey[0 : len(encodedKey)-len(v1PieceFileSuffix)]
}
key, err := pathEncoding.DecodeString(encodedKey)
if err != nil {
return nil, false
}
ref := storage.BlobRef{
Namespace: namespace,
Key: key,
}
return newBlobInfo(ref, filepath.Join(keyDir, blobFileName), keyInfo, formatVer), true
}
func walkNamespaceWithPrefix(ctx context.Context, log *zap.Logger, namespace []byte, nsDir, keyPrefix string, walkFunc func(storage.BlobInfo) error) (err error) {
keyDir := filepath.Join(nsDir, keyPrefix)
openDir, err := os.Open(keyDir)
if err != nil {
return err
}
defer func() { err = errs.Combine(err, openDir.Close()) }()
for {
// check for context done both before and after our readdir() call
if err := ctx.Err(); err != nil {
return err
}
names, err := openDir.Readdirnames(nameBatchSize)
if err != nil && err != io.EOF {
return err
}
if os.IsNotExist(err) || len(names) == 0 {
return nil
}
if err := ctx.Err(); err != nil {
return err
}
for _, name := range names {
info, err := os.Lstat(keyDir + "/" + name)
if err != nil {
if os.IsNotExist(err) {
continue
}
// convert to lowercase the perr.Op because Go reports inconsistently
// "lstat" in Linux and "Lstat" in Windows
if perr, ok := err.(*os.PathError); ok && strings.ToLower(perr.Op) == "lstat" {
log.Error("Unable to read the disk, please verify the disk is not corrupt")
}
return errs.Wrap(err)
}
if info.Mode().IsDir() {
continue
}
blobInfo, ok := decodeBlobInfo(namespace, keyPrefix, keyDir, info)
if !ok {
continue
}
err = walkFunc(blobInfo)
if err != nil {
return err
}
// also check for context done between every walkFunc callback.
if err := ctx.Err(); err != nil {
return err
}
}
}
}
// removeAllContent deletes everything in the folder
func removeAllContent(ctx context.Context, path string) (err error) {
defer mon.Task()(&ctx)(&err)
dir, err := os.Open(path)
if err != nil {
return err
}
for {
files, err := dir.Readdirnames(100)
for _, file := range files {
// the file might be still in use, so ignore the error
_ = os.RemoveAll(filepath.Join(path, file))
}
if err == io.EOF || len(files) == 0 {
return dir.Close()
}
if err != nil {
return err
}
}
}
// DiskInfo contains statistics about this dir
type DiskInfo struct {
ID string
AvailableSpace int64
}
// Info returns information about the current state of the dir
func (dir *Dir) Info() (DiskInfo, error) {
path, err := filepath.Abs(dir.path)
if err != nil {
return DiskInfo{}, err
}
return diskInfoFromPath(path)
}
type blobInfo struct {
ref storage.BlobRef
path string
fileInfo os.FileInfo
formatVersion storage.FormatVersion
}
func newBlobInfo(ref storage.BlobRef, path string, fileInfo os.FileInfo, formatVer storage.FormatVersion) storage.BlobInfo {
return &blobInfo{
ref: ref,
path: path,
fileInfo: fileInfo,
formatVersion: formatVer,
}
}
func (info *blobInfo) BlobRef() storage.BlobRef {
return info.ref
}
func (info *blobInfo) StorageFormatVersion() storage.FormatVersion {
return info.formatVersion
}
func (info *blobInfo) Stat(ctx context.Context) (os.FileInfo, error) {
return info.fileInfo, nil
}
func (info *blobInfo) FullPath(ctx context.Context) (string, error) {
return info.path, nil
}