oyd
/
minio
6
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

logging: Enable logging across storage fs layer. (#1367)

Browse Source 
Adds log.Debugf at all the layer - fixes #1074
master
Harshavardhana 9 years ago committed by Anand Babu (AB) Periasamy
parent d63d17012d
commit e9fba04b36
8 changed files with 287 additions and 39 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 30
      Logging.md
  2. 203
      fs.go
  3. 15
      logger-file-hook.go
  4. 8
      logger.go
  5. 1
      main.go
  6. 13
      network-fs.go
  7. 21
      storage-errors.go
  8. 35
      storage-rpc-server.go

30
Logging.md
Unescape View File

@ -0,0 +1,30 @@
### Logging.
- `log.Fatalf`
- `log.Errorf`
- `log.Warnf`
- `log.Infof`
- `log.Debugf`
Logging is enabled across the codebase. There are three types of logging supported.
- console
- file
- syslog
```
"console": {
"enable": true,
"level": "debug"
},
"file": {
"enable": false,
"fileName": "",
"level": "error"
},
"syslog": {
"enable": false,
"address": "",
"level": "debug"
}
```

203
fs.go
Unescape View File

@ -24,6 +24,7 @@ import (
"sync"
"syscall"
"github.com/Sirupsen/logrus"
"github.com/minio/minio/pkg/disk"
"github.com/minio/minio/pkg/safe"
)
@ -76,13 +77,20 @@ func isDirExist(dirname string) (bool, error) {
// Initialize a new storage disk.
func newFS(diskPath string) (StorageAPI, error) {
if diskPath == "" {
log.Debug("Disk cannot be empty")
return nil, errInvalidArgument
}
st, err := os.Stat(diskPath)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": diskPath,
}).Debugf("Stat failed, with error %s.", err)
return nil, err
}
if !st.IsDir() {
log.WithFields(logrus.Fields{
"diskPath": diskPath,
}).Debugf("Disk %s.", syscall.ENOTDIR)
return nil, syscall.ENOTDIR
}
fs := fsStorage{
@ -91,13 +99,21 @@ func newFS(diskPath string) (StorageAPI, error) {
listObjectMap: make(map[listParams][]*treeWalker),
listObjectMapMutex: &sync.Mutex{},
}
log.WithFields(logrus.Fields{
"diskPath": diskPath,
"minFreeDisk": 5,
}).Debugf("Successfully configured FS storage API.")
return fs, nil
}
// checkDiskFree verifies if disk path has sufficient minium free disk space.
// checkDiskFree verifies if disk path has sufficient minium free disk
// space.
func checkDiskFree(diskPath string, minFreeDisk int64) (err error) {
di, err := disk.GetInfo(diskPath)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": diskPath,
}).Debugf("Failed to get disk info, %s", err)
return err
}
@ -105,6 +121,10 @@ func checkDiskFree(diskPath string, minFreeDisk int64) (err error) {
// space used for journalling, inodes etc.
availableDiskSpace := (float64(di.Free) / (float64(di.Total) - (0.05 * float64(di.Total)))) * 100
if int64(availableDiskSpace) <= minFreeDisk {
log.WithFields(logrus.Fields{
"availableDiskSpace": int64(availableDiskSpace),
"minFreeDiskSpace": minFreeDisk,
}).Debugf("Disk free space has reached its limit.")
return errDiskFull
}
@ -142,12 +162,19 @@ func getAllUniqueVols(dirPath string) ([]VolInfo, error) {
namesOnly := true // Returned are only names.
dirents, err := scandir(dirPath, volumeFn, namesOnly)
if err != nil {
log.WithFields(logrus.Fields{
"dirPath": dirPath,
"namesOnly": true,
}).Debugf("Scandir failed with error %s", err)
return nil, err
}
var volsInfo []VolInfo
for _, dirent := range dirents {
fi, err := os.Stat(filepath.Join(dirPath, dirent.name))
if err != nil {
log.WithFields(logrus.Fields{
"path": filepath.Join(dirPath, dirent.name),
}).Debugf("Stat failed with error %s", err)
return nil, err
}
volsInfo = append(volsInfo, VolInfo{
@ -181,8 +208,10 @@ func (s fsStorage) getVolumeDir(volume string) (string, error) {
return volumeDir, errVolumeNotFound
}
for _, vol := range volsInfo {
// Verify if lowercase version of the volume
// is equal to the incoming volume, then use the proper name.
// Verify if lowercase version of
// the volume
// is equal to the incoming volume, then use the proper
// name.
if strings.ToLower(vol.Name) == volume {
volumeDir = filepath.Join(s.diskPath, vol.Name)
return volumeDir, nil
@ -190,30 +219,41 @@ func (s fsStorage) getVolumeDir(volume string) (string, error) {
}
return volumeDir, errVolumeNotFound
} else if os.IsPermission(err) {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
}).Debugf("Stat failed with error %s", err)
return volumeDir, errVolumeAccessDenied
}
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
}).Debugf("Stat failed with error %s", err)
return volumeDir, err
}
// Make a volume entry.
func (s fsStorage) MakeVol(volume string) (err error) {
// Validate if disk is free.
if err = checkDiskFree(s.diskPath, s.minFreeDisk); err != nil {
return err
}
volumeDir, err := s.getVolumeDir(volume)
if err == nil {
// Volume already exists, return error.
return errVolumeExists
}
// Validate if disk is free.
if e := checkDiskFree(s.diskPath, s.minFreeDisk); e != nil {
return e
}
// If volume not found create it.
if err == errVolumeNotFound {
// Make a volume entry.
return os.Mkdir(volumeDir, 0700)
}
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("MakeVol failed with %s", err)
// For all other errors return here.
return err
}
@ -224,10 +264,16 @@ func (s fsStorage) ListVols() (volsInfo []VolInfo, err error) {
var diskInfo disk.Info
diskInfo, err = disk.GetInfo(s.diskPath)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
}).Debugf("Failed to get disk info, %s", err)
return nil, err
}
volsInfo, err = getAllUniqueVols(s.diskPath)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
}).Debugf("getAllUniqueVols failed with %s", err)
return nil, err
}
for i, vol := range volsInfo {
@ -252,12 +298,20 @@ func (s fsStorage) StatVol(volume string) (volInfo VolInfo, err error) {
// Verify if volume is valid and it exists.
volumeDir, err := s.getVolumeDir(volume)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("getVolumeDir failed with %s", err)
return VolInfo{}, err
}
// Stat a volume entry.
var st os.FileInfo
st, err = os.Stat(volumeDir)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("Stat on the volume failed with %s", err)
if os.IsNotExist(err) {
return VolInfo{}, errVolumeNotFound
}
@ -267,6 +321,10 @@ func (s fsStorage) StatVol(volume string) (volInfo VolInfo, err error) {
var diskInfo disk.Info
diskInfo, err = disk.GetInfo(s.diskPath)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("Failed to get disk info, %s", err)
return VolInfo{}, err
}
// As os.Stat() doesn't carry other than ModTime(), use ModTime()
@ -286,18 +344,28 @@ func (s fsStorage) DeleteVol(volume string) error {
// Verify if volume is valid and it exists.
volumeDir, err := s.getVolumeDir(volume)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("getVolumeDir failed with %s", err)
return err
}
err = os.Remove(volumeDir)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("Volume remove failed with %s", err)
if os.IsNotExist(err) {
return errVolumeNotFound
} else if strings.Contains(err.Error(), "directory is not empty") {
// On windows the string is slightly different, handle it
// On windows the string is
// slightly different, handle it
// here.
return errVolumeNotEmpty
} else if strings.Contains(err.Error(), "directory not empty") {
// Hopefully for all other operating systems, this is
// Hopefully for all other
// operating systems, this is
// assumed to be consistent.
return errVolumeNotEmpty
}
@ -311,10 +379,18 @@ func (s *fsStorage) saveTreeWalk(params listParams, walker *treeWalker) {
s.listObjectMapMutex.Lock()
defer s.listObjectMapMutex.Unlock()
log.WithFields(logrus.Fields{
"bucket": params.bucket,
"recursive": params.recursive,
"marker": params.marker,
"prefix": params.prefix,
}).Debugf("saveTreeWalk has been invoked.")
walkers, _ := s.listObjectMap[params]
walkers = append(walkers, walker)
s.listObjectMap[params] = walkers
log.Debugf("Successfully saved in listObjectMap.")
}
// Lookup the goroutine reference from map
@ -322,6 +398,12 @@ func (s *fsStorage) lookupTreeWalk(params listParams) *treeWalker {
s.listObjectMapMutex.Lock()
defer s.listObjectMapMutex.Unlock()
log.WithFields(logrus.Fields{
"bucket": params.bucket,
"recursive": params.recursive,
"marker": params.marker,
"prefix": params.prefix,
}).Debugf("lookupTreeWalk has been invoked.")
if walkChs, ok := s.listObjectMap[params]; ok {
for i, walkCh := range walkChs {
if !walkCh.timedOut {
@ -331,6 +413,12 @@ func (s *fsStorage) lookupTreeWalk(params listParams) *treeWalker {
} else {
delete(s.listObjectMap, params)
}
log.WithFields(logrus.Fields{
"bucket": params.bucket,
"recursive": params.recursive,
"marker": params.marker,
"prefix": params.prefix,
}).Debugf("Found the previous saved listsObjects params.")
return walkCh
}
}
@ -345,6 +433,10 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
// Verify if volume is valid and it exists.
volumeDir, err := s.getVolumeDir(volume)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("getVolumeDir failed with %s", err)
return nil, true, err
}
var fileInfos []FileInfo
@ -352,6 +444,11 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
if marker != "" {
// Verify if marker has prefix.
if marker != "" && !strings.HasPrefix(marker, prefix) {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"marker": marker,
"prefix": prefix,
}).Debugf("Marker doesn't have prefix in common.")
return nil, true, errInvalidArgument
}
}
@ -377,6 +474,11 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
// Prefix exists as a file.
return nil, true, nil
}
log.WithFields(logrus.Fields{
"volumeDir": volumeDir,
"prefixRootDir": prefixRootDir,
}).Debugf("isDirExist returned an unhandled error %s", err)
// Rest errors should be treated as failure.
return nil, true, err
}
@ -391,6 +493,7 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
walker = startTreeWalk(s.diskPath, volume, filepath.FromSlash(prefix), filepath.FromSlash(marker), recursive)
}
nextMarker := ""
log.Debugf("Reading from the tree walk channel has begun.")
for i := 0; i < count; {
walkResult, ok := <-walker.ch
if !ok {
@ -399,6 +502,13 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
}
// For any walk error return right away.
if walkResult.err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
"prefix": prefix,
"marker": marker,
"recursive": recursive,
}).Debugf("Walk resulted in an error %s", walkResult.err)
return nil, true, walkResult.err
}
fileInfo := walkResult.fileInfo
@ -411,7 +521,14 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
nextMarker = fileInfo.Name
i++
}
s.saveTreeWalk(listParams{volume, recursive, nextMarker, prefix}, walker)
params := listParams{volume, recursive, nextMarker, prefix}
log.WithFields(logrus.Fields{
"bucket": params.bucket,
"recursive": params.recursive,
"marker": params.marker,
"prefix": params.prefix,
}).Debugf("Save the tree walk into map for subsequent requests.")
s.saveTreeWalk(params, walker)
return fileInfos, false, nil
}
@ -419,6 +536,10 @@ func (s fsStorage) ListFiles(volume, prefix, marker string, recursive bool, coun
func (s fsStorage) ReadFile(volume string, path string, offset int64) (readCloser io.ReadCloser, err error) {
volumeDir, err := s.getVolumeDir(volume)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("getVolumeDir failed with %s", err)
return nil, err
}
@ -430,19 +551,36 @@ func (s fsStorage) ReadFile(volume string, path string, offset int64) (readClose
} else if os.IsPermission(err) {
return nil, errFileAccessDenied
}
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"filePath": filePath,
}).Debugf("Opening a file failed with %s", err)
return nil, err
}
st, err := file.Stat()
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"filePath": filePath,
}).Debugf("Stat failed with %s", err)
return nil, err
}
// Verify if its not a regular file, since subsequent Seek is undefined.
if !st.Mode().IsRegular() {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"filePath": filePath,
}).Debugf("Unexpected type %s", errIsNotRegular)
return nil, errIsNotRegular
}
// Seek to requested offset.
_, err = file.Seek(offset, os.SEEK_SET)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"filePath": filePath,
"offset": offset,
}).Debugf("Seek failed with %s", err)
return nil, err
}
return file, nil
@ -452,6 +590,10 @@ func (s fsStorage) ReadFile(volume string, path string, offset int64) (readClose
func (s fsStorage) CreateFile(volume, path string) (writeCloser io.WriteCloser, err error) {
volumeDir, err := s.getVolumeDir(volume)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("getVolumeDir failed with %s", err)
return nil, err
}
if err := checkDiskFree(s.diskPath, s.minFreeDisk); err != nil {
@ -461,6 +603,10 @@ func (s fsStorage) CreateFile(volume, path string) (writeCloser io.WriteCloser,
// Verify if the file already exists and is not of regular type.
if st, err := os.Stat(filePath); err == nil {
if st.IsDir() {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"filePath": filePath,
}).Debugf("Unexpected type %s", errIsNotRegular)
return nil, errIsNotRegular
}
}
@ -471,35 +617,47 @@ func (s fsStorage) CreateFile(volume, path string) (writeCloser io.WriteCloser,
func (s fsStorage) StatFile(volume, path string) (file FileInfo, err error) {
volumeDir, err := s.getVolumeDir(volume)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("getVolumeDir failed with %s", err)
return FileInfo{}, err
}
filePath := filepath.Join(volumeDir, filepath.FromSlash(path))
st, err := os.Stat(filePath)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"filePath": filePath,
}).Debugf("Stat failed with %s", err)
// File is really not found.
if os.IsNotExist(err) {
return FileInfo{}, errFileNotFound
}
// File path cannot be verified since one of the parents is a file.
if strings.Contains(err.Error(), "not a directory") {
return FileInfo{}, errIsNotRegular
}
// Return all errors here.
return FileInfo{}, err
}
// If its a directory its not a regular file.
if st.Mode().IsDir() {
log.Debugf("File is %s", errIsNotRegular)
return FileInfo{}, errIsNotRegular
}
file = FileInfo{
return FileInfo{
Volume: volume,
Name: path,
ModTime: st.ModTime(),
Size: st.Size(),
Mode: st.Mode(),
}
return file, nil
}, nil
}
// deleteFile - delete file path if its empty.
@ -510,6 +668,9 @@ func deleteFile(basePath, deletePath string) error {
// Verify if the path exists.
pathSt, err := os.Stat(deletePath)
if err != nil {
log.WithFields(logrus.Fields{
"deletePath": deletePath,
}).Debugf("Stat failed with %s", err)
if os.IsNotExist(err) {
return errFileNotFound
} else if os.IsPermission(err) {
@ -521,6 +682,9 @@ func deleteFile(basePath, deletePath string) error {
// Verify if directory is empty.
empty, err := isDirEmpty(deletePath)
if err != nil {
log.WithFields(logrus.Fields{
"deletePath": deletePath,
}).Debugf("isDirEmpty failed with %s", err)
return err
}
if !empty {
@ -529,10 +693,17 @@ func deleteFile(basePath, deletePath string) error {
}
// Attempt to remove path.
if err := os.Remove(deletePath); err != nil {
log.WithFields(logrus.Fields{
"deletePath": deletePath,
}).Debugf("Remove failed with %s", err)
return err
}
// Recursively go down the next path and delete again.
if err := deleteFile(basePath, filepath.Dir(deletePath)); err != nil {
log.WithFields(logrus.Fields{
"basePath": basePath,
"deleteDir": filepath.Dir(deletePath),
}).Debugf("deleteFile failed with %s", err)
return err
}
return nil
@ -542,6 +713,10 @@ func deleteFile(basePath, deletePath string) error {
func (s fsStorage) DeleteFile(volume, path string) error {
volumeDir, err := s.getVolumeDir(volume)
if err != nil {
log.WithFields(logrus.Fields{
"diskPath": s.diskPath,
"volume": volume,
}).Debugf("getVolumeDir failed with %s", err)
return err
}

15
logger-file-hook.go
Unescape View File

@ -34,15 +34,24 @@ type localFile struct {
*os.File
}
func enableFileLogger(filename string) {
file, e := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600)
func enableFileLogger() {
flogger := serverConfig.GetFileLogger()
if !flogger.Enable || flogger.Filename != "" {
return
}
file, e := os.OpenFile(flogger.Filename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600)
fatalIf(probe.NewError(e), "Unable to open log file.", nil)
// Add a local file hook.
log.Hooks.Add(&localFile{file})
lvl, e := logrus.ParseLevel(flogger.Level)
fatalIf(probe.NewError(e), "Unknown log level detected, please fix your console logger configuration.", nil)
// Set default JSON formatter.
log.Formatter = new(logrus.JSONFormatter)
log.Level = logrus.InfoLevel // Minimum log level.
log.Level = lvl // Minimum log level.
}
// Fire fires the file logger hook and logs to the file.

8
logger.go
Unescape View File

@ -43,12 +43,12 @@ type logger struct {
}
// errorIf synonymous with fatalIf but doesn't exit on error != nil
func errorIf(err *probe.Error, msg string, fields map[string]interface{}) {
func errorIf(err *probe.Error, msg string, fields logrus.Fields) {
if err == nil {
return
}
if fields == nil {
fields = make(map[string]interface{})
fields = make(logrus.Fields)
}
fields["Error"] = struct {
Cause string `json:"cause,omitempty"`
@ -65,12 +65,12 @@ func errorIf(err *probe.Error, msg string, fields map[string]interface{}) {
}
// fatalIf wrapper function which takes error and prints jsonic error messages.
func fatalIf(err *probe.Error, msg string, fields map[string]interface{}) {
func fatalIf(err *probe.Error, msg string, fields logrus.Fields) {
if err == nil {
return
}
if fields == nil {
fields = make(map[string]interface{})
fields = make(logrus.Fields)
}
fields["error"] = err.ToGoError()

1
main.go
Unescape View File

@ -85,6 +85,7 @@ func migrate() {
func enableLoggers() {
// Enable all loggers here.
enableConsoleLogger()
enableFileLogger()
// Add your logger here.
}

13
network-fs.go
Unescape View File

@ -38,7 +38,7 @@ type networkFS struct {
}
const (
storageRPCPath = reservedBucket + "/rpc/storage"
storageRPCPath = reservedBucket + "/storage"
)
// splits network path into its components Address and Path.
@ -78,6 +78,7 @@ func toStorageErr(err error) error {
func newNetworkFS(networkPath string) (StorageAPI, error) {
// Input validation.
if networkPath == "" && strings.LastIndex(networkPath, ":") != -1 {
log.Debugf("Network path %s is malformed", networkPath)
return nil, errInvalidArgument
}
@ -87,6 +88,7 @@ func newNetworkFS(networkPath string) (StorageAPI, error) {
// Dial minio rpc storage http path.
rpcClient, err := rpc.DialHTTPPath("tcp", netAddr, storageRPCPath)
if err != nil {
log.Debugf("RPC HTTP dial failed for %s at path %s", netAddr, storageRPCPath)
return nil, err
}
@ -116,6 +118,7 @@ func newNetworkFS(networkPath string) (StorageAPI, error) {
func (n networkFS) MakeVol(volume string) error {
reply := GenericReply{}
if err := n.rpcClient.Call("Storage.MakeVolHandler", volume, &reply); err != nil {
log.Debugf("Storage.MakeVolHandler returned an error %s", err)
return toStorageErr(err)
}
return nil
@ -126,6 +129,7 @@ func (n networkFS) ListVols() (vols []VolInfo, err error) {
ListVols := ListVolsReply{}
err = n.rpcClient.Call("Storage.ListVolsHandler", "", &ListVols)
if err != nil {
log.Debugf("Storage.ListVolsHandler returned an error %s", err)
return nil, err
}
return ListVols.Vols, nil
@ -134,6 +138,7 @@ func (n networkFS) ListVols() (vols []VolInfo, err error) {
// StatVol - get current Stat volume info.
func (n networkFS) StatVol(volume string) (volInfo VolInfo, err error) {
if err = n.rpcClient.Call("Storage.StatVolHandler", volume, &volInfo); err != nil {
log.Debugf("Storage.StatVolHandler returned an error %s", err)
return VolInfo{}, toStorageErr(err)
}
return volInfo, nil
@ -143,6 +148,7 @@ func (n networkFS) StatVol(volume string) (volInfo VolInfo, err error) {
func (n networkFS) DeleteVol(volume string) error {
reply := GenericReply{}
if err := n.rpcClient.Call("Storage.DeleteVolHandler", volume, &reply); err != nil {
log.Debugf("Storage.DeleteVolHandler returned an error %s", err)
return toStorageErr(err)
}
return nil
@ -162,6 +168,7 @@ func (n networkFS) CreateFile(volume, path string) (writeCloser io.WriteCloser,
go func() {
resp, err := n.httpClient.Post(writeURL.String(), contentType, readCloser)
if err != nil {
log.Debugf("CreateFile http POST failed to upload the data with error %s", err)
readCloser.CloseWithError(err)
return
}
@ -187,6 +194,7 @@ func (n networkFS) StatFile(volume, path string) (fileInfo FileInfo, err error)
Vol: volume,
Path: path,
}, &fileInfo); err != nil {
log.Debugf("Storage.StatFileHandler failed with %s", err)
return FileInfo{}, toStorageErr(err)
}
return fileInfo, nil
@ -203,6 +211,7 @@ func (n networkFS) ReadFile(volume string, path string, offset int64) (reader io
readURL.RawQuery = readQuery.Encode()
resp, err := n.httpClient.Get(readURL.String())
if err != nil {
log.Debugf("ReadFile http Get failed with error %s", err)
return nil, err
}
if resp != nil {
@ -226,6 +235,7 @@ func (n networkFS) ListFiles(volume, prefix, marker string, recursive bool, coun
Recursive: recursive,
Count: count,
}, &listFilesReply); err != nil {
log.Debugf("Storage.ListFilesHandlers failed with %s", err)
return nil, true, toStorageErr(err)
}
// Return successfully unmarshalled results.
@ -239,6 +249,7 @@ func (n networkFS) DeleteFile(volume, path string) (err error) {
Vol: volume,
Path: path,
}, &reply); err != nil {
log.Debugf("Storage.DeleteFileHandler failed with %s", err)
return toStorageErr(err)
}
return nil

21
storage-errors.go
Unescape View File

@ -19,27 +19,26 @@ package main
import "errors"
// errDiskFull - cannot create volume or files when disk is full.
var errDiskFull = errors.New("Disk path full.")
var errDiskFull = errors.New("disk path full")
// errFileNotFound - cannot find the file.
var errFileNotFound = errors.New("File not found.")
var errFileNotFound = errors.New("file not found")
// errVolumeExists - cannot create same volume again.
var errVolumeExists = errors.New("Volume already exists.")
var errVolumeExists = errors.New("volume already exists")
// errIsNotRegular - not a regular file type.
var errIsNotRegular = errors.New("Not a regular file type.")
// errIsNotRegular - not of regular file type.
var errIsNotRegular = errors.New("not of regular file type")
// errVolumeNotFound - cannot find the volume.
var errVolumeNotFound = errors.New("Volume not found.")
var errVolumeNotFound = errors.New("volume not found")
// errVolumeNotEmpty - volume not empty.
var errVolumeNotEmpty = errors.New("Volume is not empty.")
var errVolumeNotEmpty = errors.New("volume is not empty")
// errVolumeAccessDenied - cannot access volume, insufficient
// permissions.
var errVolumeAccessDenied = errors.New("Volume access denied.")
var errVolumeAccessDenied = errors.New("volume access denied")
// errVolumeAccessDenied - cannot access file, insufficient
// permissions.
var errFileAccessDenied = errors.New("File access denied.")
// errVolumeAccessDenied - cannot access file, insufficient permissions.
var errFileAccessDenied = errors.New("file access denied")

35
storage-rpc-server.go
Unescape View File

@ -19,13 +19,19 @@ type storageServer struct {
// MakeVolHandler - make vol handler is rpc wrapper for MakeVol operation.
func (s *storageServer) MakeVolHandler(arg *string, reply *GenericReply) error {
return s.storage.MakeVol(*arg)
err := s.storage.MakeVol(*arg)
if err != nil {
log.Debugf("MakeVol failed with error %s", err)
return err
}
return nil
}
// ListVolsHandler - list vols handler is rpc wrapper for ListVols operation.
func (s *storageServer) ListVolsHandler(arg *string, reply *ListVolsReply) error {
vols, err := s.storage.ListVols()
if err != nil {
log.Debugf("Listsvols failed with error %s", err)
return err
}
reply.Vols = vols
@ -36,6 +42,7 @@ func (s *storageServer) ListVolsHandler(arg *string, reply *ListVolsReply) error
func (s *storageServer) StatVolHandler(arg *string, reply *VolInfo) error {
volInfo, err := s.storage.StatVol(*arg)
if err != nil {
log.Debugf("StatVol failed with error %s", err)
return err
}
*reply = volInfo
@ -45,7 +52,12 @@ func (s *storageServer) StatVolHandler(arg *string, reply *VolInfo) error {
// DeleteVolHandler - delete vol handler is a rpc wrapper for
// DeleteVol operation.
func (s *storageServer) DeleteVolHandler(arg *string, reply *GenericReply) error {
return s.storage.DeleteVol(*arg)
err := s.storage.DeleteVol(*arg)
if err != nil {
log.Debugf("DeleteVol failed with error %s", err)
return err
}
return nil
}
/// File operations
@ -54,6 +66,7 @@ func (s *storageServer) DeleteVolHandler(arg *string, reply *GenericReply) error
func (s *storageServer) ListFilesHandler(arg *ListFilesArgs, reply *ListFilesReply) error {
files, eof, err := s.storage.ListFiles(arg.Vol, arg.Prefix, arg.Marker, arg.Recursive, arg.Count)
if err != nil {
log.Debugf("ListFiles failed with error %s", err)
return err
}
@ -69,6 +82,7 @@ func (s *storageServer) ListFilesHandler(arg *ListFilesArgs, reply *ListFilesRep
func (s *storageServer) StatFileHandler(arg *StatFileArgs, reply *FileInfo) error {
fileInfo, err := s.storage.StatFile(arg.Vol, arg.Path)
if err != nil {
log.Debugf("StatFile failed with error %s", err)
return err
}
*reply = fileInfo
@ -77,7 +91,12 @@ func (s *storageServer) StatFileHandler(arg *StatFileArgs, reply *FileInfo) erro
// DeleteFileHandler - delete file handler is rpc wrapper to delete file.
func (s *storageServer) DeleteFileHandler(arg *DeleteFileArgs, reply *GenericReply) error {
return s.storage.DeleteFile(arg.Vol, arg.Path)
err := s.storage.DeleteFile(arg.Vol, arg.Path)
if err != nil {
log.Debugf("DeleteFile failed with error %s", err)
return err
}
return nil
}
// registerStorageRPCRouter - register storage rpc router.
@ -89,14 +108,15 @@ func registerStorageRPCRouter(mux *router.Router, storageAPI StorageAPI) {
storageRPCServer.RegisterName("Storage", stServer)
storageRouter := mux.NewRoute().PathPrefix(reservedBucket).Subrouter()
// Add minio storage routes.
storageRouter.Path("/rpc/storage").Handler(storageRPCServer)
storageRouter.Path("/storage").Handler(storageRPCServer)
// StreamUpload - stream upload handler.
storageRouter.Methods("POST").Path("/rpc/storage/upload/{volume}/{path:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
storageRouter.Methods("POST").Path("/storage/upload/{volume}/{path:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
vars := router.Vars(r)
volume := vars["volume"]
path := vars["path"]
writeCloser, err := stServer.storage.CreateFile(volume, path)
if err != nil {
log.Debugf("CreateFile failed with error %s", err)
httpErr := http.StatusInternalServerError
if err == errVolumeNotFound {
httpErr = http.StatusNotFound
@ -108,6 +128,7 @@ func registerStorageRPCRouter(mux *router.Router, storageAPI StorageAPI) {
}
reader := r.Body
if _, err = io.Copy(writeCloser, reader); err != nil {
log.Debugf("Copying incoming reader to writer failed %s", err)
safeCloseAndRemove(writeCloser)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
@ -116,17 +137,19 @@ func registerStorageRPCRouter(mux *router.Router, storageAPI StorageAPI) {
reader.Close()
})
// StreamDownloadHandler - stream download handler.
storageRouter.Methods("GET").Path("/rpc/storage/download/{volume}/{path:.+}").Queries("offset", "{offset:.*}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
storageRouter.Methods("GET").Path("/storage/download/{volume}/{path:.+}").Queries("offset", "{offset:.*}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
vars := router.Vars(r)
volume := vars["volume"]
path := vars["path"]
offset, err := strconv.ParseInt(r.URL.Query().Get("offset"), 10, 64)
if err != nil {
log.Debugf("Parse offset failure with error %s", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
readCloser, err := stServer.storage.ReadFile(volume, path, offset)
if err != nil {
log.Debugf("ReadFile failed with error %s", err)
httpErr := http.StatusBadRequest
if err == errVolumeNotFound {
httpErr = http.StatusNotFound

Write Preview
Loading…
Cancel
Save