objectLayer: Check for format.json in a wrapped disk. (#3311)

This is needed to validate if the `format.json` indeed exists when a fresh node is brought online. This wrapped implementation also connects to the remote node by attempting a re-login. Subsequently after a successful connect `format.json` is validated as well. Fixes #3207
2026-07-23 22:16:15 +03:00 · 2016-11-23 15:48:10 -08:00
parent 7a5bbf7a2e
commit 6efee2072d
26 changed files with 877 additions and 194 deletions
@@ -22,14 +22,16 @@ import (
 	"net/rpc"
 	"net/url"
 	"path"
+	"sync/atomic"

 	"github.com/minio/minio/pkg/disk"
 )

 type networkStorage struct {
-	netAddr   string
-	netPath   string
-	rpcClient *AuthRPCClient
+	networkIOErrCount int32 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
+	netAddr           string
+	netPath           string
+	rpcClient         *AuthRPCClient
 }

 const (
@@ -54,8 +56,6 @@ func toStorageErr(err error) error {
 		return io.EOF
 	case io.ErrUnexpectedEOF.Error():
 		return io.ErrUnexpectedEOF
-	case rpc.ErrShutdown.Error():
-		return errDiskNotFound
 	case errUnexpected.Error():
 		return errUnexpected
 	case errDiskFull.Error():
@@ -132,12 +132,39 @@ func newStorageRPC(ep *url.URL) (StorageAPI, error) {
 }

 // Stringer interface compatible representation of network device.
-func (n networkStorage) String() string {
+func (n *networkStorage) String() string {
 	return n.netAddr + ":" + n.netPath
 }

+// maximum allowed network IOError.
+const maxAllowedNetworkIOError = 1024
+
+// Initializes the remote RPC connection by attempting a login attempt.
+func (n *networkStorage) Init() (err error) {
+	// Attempt a login to reconnect.
+	return n.rpcClient.Login()
+}
+
+// Closes the underlying RPC connection.
+func (n *networkStorage) Close() (err error) {
+	// Close the underlying connection.
+	return n.rpcClient.Close()
+}
+
 // DiskInfo - fetch disk information for a remote disk.
-func (n networkStorage) DiskInfo() (info disk.Info, err error) {
+func (n *networkStorage) DiskInfo() (info disk.Info, err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return disk.Info{}, errFaultyRemoteDisk
+	}
+
 	args := GenericArgs{}
 	if err = n.rpcClient.Call("Storage.DiskInfoHandler", &args, &info); err != nil {
 		return disk.Info{}, toStorageErr(err)
@@ -146,7 +173,19 @@ func (n networkStorage) DiskInfo() (info disk.Info, err error) {
 }

 // MakeVol - create a volume on a remote disk.
-func (n networkStorage) MakeVol(volume string) error {
+func (n *networkStorage) MakeVol(volume string) (err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return errFaultyRemoteDisk
+	}
+
 	reply := GenericReply{}
 	args := GenericVolArgs{Vol: volume}
 	if err := n.rpcClient.Call("Storage.MakeVolHandler", &args, &reply); err != nil {
@@ -156,7 +195,19 @@ func (n networkStorage) MakeVol(volume string) error {
 }

 // ListVols - List all volumes on a remote disk.
-func (n networkStorage) ListVols() (vols []VolInfo, err error) {
+func (n *networkStorage) ListVols() (vols []VolInfo, err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return nil, errFaultyRemoteDisk
+	}
+
 	ListVols := ListVolsReply{}
 	err = n.rpcClient.Call("Storage.ListVolsHandler", &GenericArgs{}, &ListVols)
 	if err != nil {
@@ -165,8 +216,20 @@ func (n networkStorage) ListVols() (vols []VolInfo, err error) {
 	return ListVols.Vols, nil
 }

-// StatVol - get current Stat volume info.
-func (n networkStorage) StatVol(volume string) (volInfo VolInfo, err error) {
+// StatVol - get volume info over the network.
+func (n *networkStorage) StatVol(volume string) (volInfo VolInfo, err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return VolInfo{}, errFaultyRemoteDisk
+	}
+
 	args := GenericVolArgs{Vol: volume}
 	if err = n.rpcClient.Call("Storage.StatVolHandler", &args, &volInfo); err != nil {
 		return VolInfo{}, toStorageErr(err)
@@ -174,8 +237,20 @@ func (n networkStorage) StatVol(volume string) (volInfo VolInfo, err error) {
 	return volInfo, nil
 }

-// DeleteVol - Delete a volume.
-func (n networkStorage) DeleteVol(volume string) error {
+// DeleteVol - Deletes a volume over the network.
+func (n *networkStorage) DeleteVol(volume string) (err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return errFaultyRemoteDisk
+	}
+
 	reply := GenericReply{}
 	args := GenericVolArgs{Vol: volume}
 	if err := n.rpcClient.Call("Storage.DeleteVolHandler", &args, &reply); err != nil {
@@ -186,7 +261,18 @@ func (n networkStorage) DeleteVol(volume string) error {

 // File operations.

-func (n networkStorage) PrepareFile(volume, path string, length int64) (err error) {
+func (n *networkStorage) PrepareFile(volume, path string, length int64) (err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return errFaultyRemoteDisk
+	}
+
 	reply := GenericReply{}
 	if err = n.rpcClient.Call("Storage.PrepareFileHandler", &PrepareFileArgs{
 		Vol:  volume,
@@ -198,8 +284,20 @@ func (n networkStorage) PrepareFile(volume, path string, length int64) (err erro
 	return nil
 }

-// CreateFile - create file.
-func (n networkStorage) AppendFile(volume, path string, buffer []byte) (err error) {
+// AppendFile - append file writes buffer to a remote network path.
+func (n *networkStorage) AppendFile(volume, path string, buffer []byte) (err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return errFaultyRemoteDisk
+	}
+
 	reply := GenericReply{}
 	if err = n.rpcClient.Call("Storage.AppendFileHandler", &AppendFileArgs{
 		Vol:    volume,
@@ -212,7 +310,19 @@ func (n networkStorage) AppendFile(volume, path string, buffer []byte) (err erro
 }

 // StatFile - get latest Stat information for a file at path.
-func (n networkStorage) StatFile(volume, path string) (fileInfo FileInfo, err error) {
+func (n *networkStorage) StatFile(volume, path string) (fileInfo FileInfo, err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return FileInfo{}, errFaultyRemoteDisk
+	}
+
 	if err = n.rpcClient.Call("Storage.StatFileHandler", &StatFileArgs{
 		Vol:  volume,
 		Path: path,
@@ -226,7 +336,19 @@ func (n networkStorage) StatFile(volume, path string) (fileInfo FileInfo, err er
 // contents in a byte slice. Returns buf == nil if err != nil.
 // This API is meant to be used on files which have small memory footprint, do
 // not use this on large files as it would cause server to crash.
-func (n networkStorage) ReadAll(volume, path string) (buf []byte, err error) {
+func (n *networkStorage) ReadAll(volume, path string) (buf []byte, err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return nil, errFaultyRemoteDisk
+	}
+
 	if err = n.rpcClient.Call("Storage.ReadAllHandler", &ReadAllArgs{
 		Vol:  volume,
 		Path: path,
@@ -236,8 +358,20 @@ func (n networkStorage) ReadAll(volume, path string) (buf []byte, err error) {
 	return buf, nil
 }

-// ReadFile - reads a file.
-func (n networkStorage) ReadFile(volume string, path string, offset int64, buffer []byte) (m int64, err error) {
+// ReadFile - reads a file at remote path and fills the buffer.
+func (n *networkStorage) ReadFile(volume string, path string, offset int64, buffer []byte) (m int64, err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return 0, errFaultyRemoteDisk
+	}
+
 	var result []byte
 	err = n.rpcClient.Call("Storage.ReadFileHandler", &ReadFileArgs{
 		Vol:    volume,
@@ -252,7 +386,19 @@ func (n networkStorage) ReadFile(volume string, path string, offset int64, buffe
 }

 // ListDir - list all entries at prefix.
-func (n networkStorage) ListDir(volume, path string) (entries []string, err error) {
+func (n *networkStorage) ListDir(volume, path string) (entries []string, err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return nil, errFaultyRemoteDisk
+	}
+
 	if err = n.rpcClient.Call("Storage.ListDirHandler", &ListDirArgs{
 		Vol:  volume,
 		Path: path,
@@ -264,7 +410,19 @@ func (n networkStorage) ListDir(volume, path string) (entries []string, err erro
 }

 // DeleteFile - Delete a file at path.
-func (n networkStorage) DeleteFile(volume, path string) (err error) {
+func (n *networkStorage) DeleteFile(volume, path string) (err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return errFaultyRemoteDisk
+	}
+
 	reply := GenericReply{}
 	if err = n.rpcClient.Call("Storage.DeleteFileHandler", &DeleteFileArgs{
 		Vol:  volume,
@@ -275,8 +433,20 @@ func (n networkStorage) DeleteFile(volume, path string) (err error) {
 	return nil
 }

-// RenameFile - Rename file.
-func (n networkStorage) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) (err error) {
+// RenameFile - rename a remote file from source to destination.
+func (n *networkStorage) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) (err error) {
+	defer func() {
+		if err == errDiskNotFound || err == rpc.ErrShutdown {
+			atomic.AddInt32(&n.networkIOErrCount, 1)
+		}
+	}()
+
+	// Take remote disk offline if the total network errors.
+	// are more than maximum allowable IO error limit.
+	if n.networkIOErrCount > maxAllowedNetworkIOError {
+		return errFaultyRemoteDisk
+	}
+
 	reply := GenericReply{}
 	if err = n.rpcClient.Call("Storage.RenameFileHandler", &RenameFileArgs{
 		SrcVol:  srcVolume,