astelm/minio
1
0
Fork 0
mirror of https://github.com/pgsty/minio.git synced 2026-08-07 16:21:14 +03:00
Code Issues Packages Projects Releases Wiki Activity

posix: Do not take disk offline on I/O errors (#8836)

Browse Source 
Choosing maxAllowedIOError is arbitrary and
prone to errors, when drives might be perfectly
capable of taking I/O with only few locations
return I/O error. This is a hindrance of sort
where backend filesystems like ZFS can automatically
fix and handle these scenarios.

The added problem with current approach that we
take the drive offline, making it virtually impossible
to bring it online without restart the server which
is not desirable on a busy cluster. Remove this state
such that let the backend return error appropriately
to caller and let the caller decide what to do with
the error.
This commit is contained in:
Harshavardhana
2020-01-17 13:34:43 -08:00
committed by GitHub
parent 005ebbb9b2
commit fc5213258e
2 changed files with 40 additions and 411 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/posix.go
+13 -190
View File
@@ -50,8 +50,7 @@ import (
const (
diskMinFreeSpace = 900 * humanize.MiByte // Min 900MiB free space.
diskMinTotalSpace = diskMinFreeSpace // Min 900MiB total space.
maxAllowedIOError = 5
readBlockSize = 4 * humanize.MiByte // Default read block size 4MiB.
readBlockSize = 4 * humanize.MiByte // Default read block size 4MiB.
// On regular files bigger than this;
readAheadSize = 16 << 20
@@ -83,11 +82,10 @@ func isValidVolname(volname string) bool {
// posix - implements StorageAPI interface.
type posix struct {
// Disk usage metrics
totalUsed uint64 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
ioErrCount int32 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
totalUsed uint64 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG
activeIOCount int32
maxActiveIOCount int32
activeIOCount int32
diskPath string
pool sync.Pool
@@ -219,8 +217,12 @@ func newPosix(path string) (*posix, error) {
return &b
},
},
stopUsageCh: make(chan struct{}),
diskMount: mountinfo.IsLikelyMountPoint(path),
stopUsageCh: make(chan struct{}),
diskMount: mountinfo.IsLikelyMountPoint(path),
// Allow disk usage crawler to run upto 10 concurrent
// I/O ops, if and when activeIOCount reaches this
// value disk usage routine suspends the crawler
// and waits until activeIOCount reaches below this threshold.
maxActiveIOCount: 10,
}
@@ -392,7 +394,7 @@ func (s *posix) CrawlAndGetDataUsage(endCh <-chan struct{}) (DataUsageInfo, erro
return nil
}
if strings.HasSuffix(prefix, "/xl.json") {
if strings.HasSuffix(prefix, SlashSeparator+xlMetaJSONFile) {
xlMetaBuf, err := ioutil.ReadFile(origPath)
if err != nil {
return nil
@@ -434,16 +436,6 @@ type DiskInfo struct {
// DiskInfo provides current information about disk space usage,
// total free inodes and underlying filesystem.
func (s *posix) DiskInfo() (info DiskInfo, err error) {
defer func() {
if s != nil && err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return info, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -557,16 +549,6 @@ func (s *posix) MakeVolBulk(volumes ...string) (err error) {
// Make a volume entry.
func (s *posix) MakeVol(volume string) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
if !isValidVolname(volume) {
return errInvalidArgument
}
@@ -601,16 +583,6 @@ func (s *posix) MakeVol(volume string) (err error) {
// ListVols - list volumes.
func (s *posix) ListVols() (volsInfo []VolInfo, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return nil, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -671,16 +643,6 @@ func listVols(dirPath string) ([]VolInfo, error) {
// StatVol - get volume info.
func (s *posix) StatVol(volume string) (volInfo VolInfo, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return VolInfo{}, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -713,16 +675,6 @@ func (s *posix) StatVol(volume string) (volInfo VolInfo, err error) {
// DeleteVol - delete a volume.
func (s *posix) DeleteVol(volume string) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -756,16 +708,6 @@ func (s *posix) DeleteVol(volume string) (err error) {
func (s *posix) Walk(volume, dirPath, marker string, recursive bool, leafFile string,
readMetadataFn readMetadataFunc, endWalkCh chan struct{}) (ch chan FileInfo, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return nil, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -834,16 +776,6 @@ func (s *posix) Walk(volume, dirPath, marker string, recursive bool, leafFile st
// ListDir - return all the entries at the given directory path.
// If an entry is a directory it will be returned with a trailing SlashSeparator.
func (s *posix) ListDir(volume, dirPath string, count int, leafFile string) (entries []string, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return nil, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -891,16 +823,6 @@ func (s *posix) ListDir(volume, dirPath string, count int, leafFile string) (ent
// 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 (s *posix) ReadAll(volume, path string) (buf []byte, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return nil, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -963,22 +885,10 @@ func (s *posix) ReadAll(volume, path string) (buf []byte, err error) {
// Additionally ReadFile also starts reading from an offset. ReadFile
// semantics are same as io.ReadFull.
func (s *posix) ReadFile(volume, path string, offset int64, buffer []byte, verifier *BitrotVerifier) (int64, error) {
var n int
var err error
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if offset < 0 {
return 0, errInvalidArgument
}
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return 0, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -988,6 +898,9 @@ func (s *posix) ReadFile(volume, path string, offset int64, buffer []byte, verif
if err != nil {
return 0, err
}
var n int
// Stat a volume entry.
_, err = os.Stat((volumeDir))
if err != nil {
@@ -1071,16 +984,6 @@ func (s *posix) ReadFile(volume, path string, offset int64, buffer []byte, verif
}
func (s *posix) openFile(volume, path string, mode int) (f *os.File, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return nil, errFaultyDisk
}
volumeDir, err := s.getVolDir(volume)
if err != nil {
return nil, err
@@ -1137,21 +1040,10 @@ func (s *posix) openFile(volume, path string, mode int) (f *os.File, err error)
// ReadFileStream - Returns the read stream of the file.
func (s *posix) ReadFileStream(volume, path string, offset, length int64) (io.ReadCloser, error) {
var err error
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if offset < 0 {
return nil, errInvalidArgument
}
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return nil, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -1230,15 +1122,6 @@ func (s *posix) CreateFile(volume, path string, fileSize int64, r io.Reader) (er
if fileSize < -1 {
return errInvalidArgument
}
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
@@ -1334,16 +1217,6 @@ func (s *posix) CreateFile(volume, path string, fileSize int64, r io.Reader) (er
}
func (s *posix) WriteAll(volume, path string, reader io.Reader) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -1368,16 +1241,6 @@ func (s *posix) WriteAll(volume, path string, reader io.Reader) (err error) {
// AppendFile - append a byte array at path, if file doesn't exist at
// path this call explicitly creates it.
func (s *posix) AppendFile(volume, path string, buf []byte) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -1400,16 +1263,6 @@ func (s *posix) AppendFile(volume, path string, buf []byte) (err error) {
// StatFile - get file info.
func (s *posix) StatFile(volume, path string) (file FileInfo, err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return FileInfo{}, errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -1501,16 +1354,6 @@ func deleteFile(basePath, deletePath string) error {
// DeleteFile - delete a file at path.
func (s *posix) DeleteFile(volume, path string) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -1555,16 +1398,6 @@ func (s *posix) DeleteFileBulk(volume string, paths []string) (errs []error, err
// RenameFile - rename source path to destination path atomically.
func (s *posix) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)
@@ -1652,16 +1485,6 @@ func (s *posix) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) (err e
}
func (s *posix) VerifyFile(volume, path string, fileSize int64, algo BitrotAlgorithm, sum []byte, shardSize int64) (err error) {
defer func() {
if err == errFaultyDisk {
atomic.AddInt32(&s.ioErrCount, 1)
}
}()
if atomic.LoadInt32(&s.ioErrCount) > maxAllowedIOError {
return errFaultyDisk
}
atomic.AddInt32(&s.activeIOCount, 1)
defer func() {
atomic.AddInt32(&s.activeIOCount, -1)