cleanup zero size objects; handle associated conflicts (ref)

* refactoring

Signed-off-by: Alex Aizman <[email protected]>

ais/target.go

@@ -1309,7 +1309,7 @@ func (t *target) delobj(lom *core.LOM, evict bool) (int, error, bool) {
 			if cmn.IsErrObjNought(err) {
 				// cleanup in place
 				if errNested := lom.RemoveMain(); errNested != nil {
-					nlog.Errorln(t.String(), "failed to cleanup in place", errNested)
+					nlog.Errorln(t.String(), "failed to cleanup in place: nested err [", err, errNested, "]")
 				}
 				return http.StatusNotFound, nil, false
 			}

cmd/cli/cli/xact.go

@@ -365,6 +365,7 @@ func xstart(c *cli.Context, xargs *xact.ArgsMsg, extra string) (xid string, err
 	if !strings.Contains(err.Error(), "marshal") {
 		return "", V(err)
 	}
+	debug.Assert(xargs.Flags != 0) // ditto
 	if smap, e1 := getClusterMap(c); e1 == nil {
 		if ds, e2 := api.GetStatsAndStatus(apiBP, smap.Primary); e2 == nil {
 			err = fmt.Errorf("CLI version %s is not compatible with (an older) AIS v%s", c.App.Version, ds.Version)

cmn/cos/err.go

@@ -232,10 +232,8 @@ func CheckMvToVirtDir(err error, dst string) error {
 	if IsErrMvToVirtDir(err) {
 		return err
 	}
-	if os.IsExist(err) {
-		if finfo, errN := os.Stat(dst); errN == nil && finfo.IsDir() {
-			return &ErrMvToVirtDir{dst}
-		}
+	if finfo, errN := os.Stat(dst); errN == nil && finfo.IsDir() {
+		return &ErrMvToVirtDir{dst}
 	}
 	return err
 }

cmn/cos/ioutils.go

@@ -79,8 +79,9 @@ func CreateDir(dir string) error {
 }
 
 // CreateFile creates a new write-only (O_WRONLY) file with default cos.PermRWR permissions.
-// NOTE: if the file pathname doesn't exist it'll be created.
-// NOTE: if the file already exists it'll be also silently truncated.
+// [NOTE]
+// - if the file pathname doesn't exist it'll be created
+// - but if it does it'll be also silently truncated
 func CreateFile(fqn string) (*os.File, error) {
 	if err := CreateDir(filepath.Dir(fqn)); err != nil {
 		return nil, err
@@ -89,20 +90,24 @@ func CreateFile(fqn string) (*os.File, error) {
 }
 
 // (creates destination directory if doesn't exist)
-func Rename(src, dst string) (err error) {
-	err = os.Rename(src, dst)
+func Rename(src, dst string) error {
+	err := os.Rename(src, dst)
 	if err == nil {
 		return nil
 	}
 	if !os.IsNotExist(err) {
 		return CheckMvToVirtDir(err, dst)
 	}
-	// create and retry (slow path)
+
+	// slow path
 	err = CreateDir(filepath.Dir(dst))
 	if err == nil || os.IsExist(err) /*race*/ {
 		err = os.Rename(src, dst)
 	}
-	return err
+	if err == nil {
+		return nil
+	}
+	return CheckMvToVirtDir(err, dst)
 }
 
 // RemoveFile removes path; returns nil upon success or if the path does not exist.
@@ -114,36 +119,36 @@ func RemoveFile(path string) (err error) {
 	return
 }
 
-// and computes checksum if requested
+// and computes checksum, if requested
 func CopyFile(src, dst string, buf []byte, cksumType string) (written int64, cksum *CksumHash, err error) {
-	var srcFile, dstFile *os.File
-	if srcFile, err = os.Open(src); err != nil {
-		return
+	const tag = "copy-file:"
+	var srcfh, dstfh *os.File
+	if srcfh, err = os.Open(src); err != nil {
+		return 0, nil, err
 	}
-	if dstFile, err = CreateFile(dst); err != nil {
-		nlog.Errorln("Failed to create", dst+":", err)
-		Close(srcFile)
-		return
+	if dstfh, err = CreateFile(dst); err != nil {
+		nlog.Errorln(tag, err)
+		Close(srcfh)
+		return 0, nil, err
 	}
-	written, cksum, err = CopyAndChecksum(dstFile, srcFile, buf, cksumType)
-	Close(srcFile)
-	defer func() {
-		if err == nil {
-			return
-		}
-		if nestedErr := RemoveFile(dst); nestedErr != nil {
-			nlog.Errorf("Nested (%v): failed to remove %s, err: %v", err, dst, nestedErr)
-		}
-	}()
+	written, cksum, err = CopyAndChecksum(dstfh, srcfh, buf, cksumType)
+	Close(srcfh)
 	if err != nil {
-		nlog.Errorln("Failed to copy", src, "=>", dst+":", err)
-		Close(dstFile)
-		return
+		Close(dstfh)
+		goto eret
 	}
-	if err = FlushClose(dstFile); err != nil {
-		nlog.Errorln("Failed to flush and close", dst+":", err)
+	if err = FlushClose(dstfh); err != nil {
+		dstfh.Close()
+		goto eret
 	}
-	return
+	return written, cksum, err
+
+eret:
+	nlog.Errorln(tag, err)
+	if nerr := RemoveFile(dst); nerr != nil {
+		nlog.Errorln(tag, "nested err: [", nerr, "]")
+	}
+	return 0, nil, err
 }
 
 // Saves the `reader` directly to `fqn`, checksums if requested

core/lfile.go

@@ -145,11 +145,13 @@ func (lom *LOM) RenameFinalize(wfqn string) error {
 	if err := cos.Stat(bdir); err != nil {
 		return &errBdir{cname: lom.Cname(), err: err}
 	}
-	if err := lom.RenameToMain(wfqn); err != nil {
-		if !cos.IsErrMvToVirtDir(err) {
-			T.FSHC(err, lom.Mountpath(), wfqn)
-		}
-		return cmn.NewErrFailedTo(T, "finalize", lom.Cname(), err)
+	err := lom.RenameToMain(wfqn)
+	if err == nil {
+		return nil
+	}
+	if cos.IsErrMvToVirtDir(err) {
+		return err
 	}
-	return nil
+	T.FSHC(err, lom.Mountpath(), wfqn)
+	return cmn.NewErrFailedTo(T, "finalize", lom.Cname(), err)
 }

core/lom.go

@@ -560,7 +560,6 @@ func (lom *LOM) FromFS() error {
 			if cos.IsPathErr(err) && strings.Contains(err.Error(), "not a directory") {
 				// e.g. err "stat .../aaa/111: not a directory" when there's existing ".../aaa" object
 				err := fmt.Errorf("%w (path error)", err)
-				nlog.Errorln(err)
 				return err
 			}
 			err = os.NewSyscallError("stat", err)

space/cleanup.go

@@ -30,6 +30,8 @@ import (
 	"github.com/NVIDIA/aistore/xact/xreg"
 )
 
+// stats counters "cleanup.store.n" & "cleanup.store.size" (not to confuse with generic ""loc-objs", "in-objs", etc.)
+
 type (
 	IniCln struct {
 		StatsT  stats.Tracker
@@ -490,10 +492,16 @@ func (j *clnJ) visitObj(fqn string, lom *core.LOM) {
 		}
 		if lom.Lsize() == 0 {
 			if j.ini.Args.Flags&xact.XrmZeroSize == xact.XrmZeroSize {
+				// remove in place
 				if ecode, err := core.T.DeleteObject(lom, false /*evict*/); err != nil {
-					nlog.Errorln("failed to remove zero-size", lom.Cname(), "err:", err, "code:", ecode)
+					nlog.Errorln("failed to remove zero size", lom.Cname(), "err: [", err, ecode, "]")
 				} else {
-					nlog.Warningln("removed zero-size", lom.Cname())
+					if lom.Bck().IsRemote() {
+						nlog.Warningln("removed zero size", lom.Cname(), "(both cluster and remote)")
+					} else {
+						nlog.Warningln("removed zero size", lom.Cname())
+					}
+					j.ini.StatsT.Inc(stats.CleanupStoreCount)
 				}
 			}
 		}