storage: Add storage interface.

master
Harshavardhana 9 years ago
parent b182e94acc
commit bad2f2afbb
  1. 17
      fs-object.go
  2. 35
      storage-api-interface.go
  3. 35
      storage-common.go
  4. 37
      storage-datatypes.go
  5. 291
      storage-local.go

@ -33,23 +33,6 @@ import (
"github.com/minio/minio/pkg/safe"
)
// isDirEmpty - returns whether given directory is empty or not.
func isDirEmpty(dirname string) (status bool, err error) {
f, err := os.Open(dirname)
if err == nil {
defer f.Close()
if _, err = f.Readdirnames(1); err == io.EOF {
status = true
err = nil
}
}
return
}
/// Object Operations
// GetObject - GET object

@ -0,0 +1,35 @@
/*
* Minio Cloud Storage, (C) 2016 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package main
import "io"
// StorageAPI interface.
type StorageAPI interface {
// Volume operations.
MakeVol(volume string) (err error)
ListVols() (vols []VolInfo, err error)
StatVol(volume string) (vol VolInfo, err error)
DeleteVol(volume string) (err error)
// File operations.
ListFiles(volume, prefix, marker string, recursive bool, count int) (files []FileInfo, isEOF bool, err error)
ReadFile(volume string, path string, offset int64) (readCloser io.ReadCloser, err error)
CreateFile(volume string, path string) (writeCloser io.WriteCloser, err error)
StatFile(volume string, path string) (file FileInfo, err error)
DeleteFile(volume string, path string) (err error)
}

@ -0,0 +1,35 @@
/*
* Minio Cloud Storage, (C) 2016 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package main
import (
"io"
"os"
)
// isDirEmpty - returns whether given directory is empty or not.
func isDirEmpty(dirname string) (status bool, err error) {
f, err := os.Open(dirname)
if err == nil {
defer f.Close()
if _, err = f.Readdirnames(1); err == io.EOF {
status = true
err = nil
}
}
return status, err
}

@ -0,0 +1,37 @@
/*
* Minio Cloud Storage, (C) 2016 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package main
import (
"os"
"time"
)
// FileInfo - file stat information.
type FileInfo struct {
Volume string
Name string
ModTime time.Time
Size int64
Type os.FileMode
}
// VolInfo - volume info
type VolInfo struct {
Name string
Created time.Time
}

@ -0,0 +1,291 @@
/*
* Minio Cloud Storage, (C) 2016 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package main
import (
"errors"
"io"
"io/ioutil"
"os"
"path/filepath"
"strings"
"syscall"
"github.com/minio/minio/pkg/disk"
"github.com/minio/minio/pkg/safe"
)
// ErrDiskPathFull - cannot create volume or files when disk is full.
var ErrDiskPathFull = errors.New("Disk path full.")
// ErrVolumeExists - cannot create same volume again.
var ErrVolumeExists = errors.New("Volume already exists.")
// ErrIsNotRegular - is not a regular file type.
var ErrIsNotRegular = errors.New("Not a regular file type.")
// localStorage implements StorageAPI on top of provided diskPath.
type localStorage struct {
diskPath string
fsInfo disk.Info
minFreeDisk int64
}
// Initialize a new local storage.
func newLocalStorage(diskPath string) (StorageAPI, error) {
if diskPath == "" {
return nil, errInvalidArgument
}
st, e := os.Stat(diskPath)
if e != nil {
return nil, e
}
if !st.IsDir() {
return nil, syscall.ENOTDIR
}
info, e := disk.GetInfo(diskPath)
if e != nil {
return nil, e
}
disk := localStorage{
diskPath: diskPath,
fsInfo: info,
minFreeDisk: 5, // Minimum 5% disk should be free.
}
return disk, nil
}
// checkDiskFree verifies if disk path has sufficient minium free disk
// space.
func checkDiskFree(diskPath string, minFreeDisk int64) error {
di, e := disk.GetInfo(diskPath)
if e != nil {
return e
}
// Remove 5% from total space for cumulative disk space used for journalling, inodes etc.
availableDiskSpace := (float64(di.Free) / (float64(di.Total) - (0.05 * float64(di.Total)))) * 100
if int64(availableDiskSpace) <= minFreeDisk {
return ErrDiskPathFull
}
// Success.
return nil
}
// Make a volume entry.
func (s localStorage) MakeVol(volume string) error {
if e := checkDiskFree(s.diskPath, s.minFreeDisk); e != nil {
return e
}
volumeDir := getVolumeDir(s.diskPath, volume)
if _, e := os.Stat(volumeDir); e == nil {
return ErrVolumeExists
}
// Make a volume entry.
if e := os.Mkdir(volumeDir, 0700); e != nil {
return e
}
return nil
}
// removeDuplicateVols - remove duplicate volumes.
func removeDuplicateVols(vols []VolInfo) []VolInfo {
length := len(vols) - 1
for i := 0; i < length; i++ {
for j := i + 1; j <= length; j++ {
if vols[i].Name == vols[j].Name {
// Pick the latest volume from a duplicate entry.
if vols[i].Created.Sub(vols[j].Created) > 0 {
vols[i] = vols[length]
} else {
vols[j] = vols[length]
}
vols = vols[0:length]
length--
j--
}
}
}
return vols
}
// ListVols - list volumes.
func (s localStorage) ListVols() ([]VolInfo, error) {
files, e := ioutil.ReadDir(s.diskPath)
if e != nil {
return nil, e
}
var volsInfo []VolInfo
for _, file := range files {
if !file.IsDir() {
// If not directory, ignore all file types.
continue
}
volInfo := VolInfo{
Name: file.Name(),
Created: file.ModTime(),
}
volsInfo = append(volsInfo, volInfo)
}
// Remove duplicated volume entries.
volsInfo = removeDuplicateVols(volsInfo)
return volsInfo, nil
}
// getVolumeDir - will convert incoming volume names to
// corresponding valid volume names on the backend in a platform
// compatible way for all operating systems.
func getVolumeDir(diskPath, volume string) string {
volumes, e := ioutil.ReadDir(diskPath)
if e != nil {
return volume
}
for _, vol := range volumes {
// Verify if lowercase version of the volume
// is equal to the incoming volume, then use the proper name.
if strings.ToLower(vol.Name()) == volume {
return filepath.Join(diskPath, vol.Name())
}
}
return filepath.Join(diskPath, volume)
}
// StatVol - get volume info.
func (s localStorage) StatVol(volume string) (VolInfo, error) {
volumeDir := getVolumeDir(s.diskPath, volume)
// Stat a volume entry.
st, e := os.Stat(volumeDir)
if e != nil {
return VolInfo{}, e
}
volInfo := VolInfo{}
volInfo.Name = st.Name()
volInfo.Created = st.ModTime()
return volInfo, nil
}
// DeleteVol - delete a volume.
func (s localStorage) DeleteVol(volume string) error {
return os.Remove(getVolumeDir(s.diskPath, volume))
}
/// File operations.
// ListFiles - list files are prefix and marker.
func (s localStorage) ListFiles(volume, prefix, marker string, recursive bool, count int) (files []FileInfo, isEOF bool, err error) {
// TODO
return files, true, nil
}
// ReadFile - read a file at a given offset.
func (s localStorage) ReadFile(volume string, path string, offset int64) (io.ReadCloser, error) {
filePath := filepath.Join(getVolumeDir(s.diskPath, volume), path)
file, e := os.Open(filePath)
if e != nil {
return nil, e
}
st, e := file.Stat()
if e != nil {
return nil, e
}
// Verify if its not a regular file, since subsequent Seek is undefined.
if !st.Mode().IsRegular() {
return nil, ErrIsNotRegular
}
_, e = file.Seek(offset, os.SEEK_SET)
if e != nil {
return nil, e
}
return file, nil
}
// CreateFile - create a file at path.
func (s localStorage) CreateFile(volume, path string) (writeCloser io.WriteCloser, err error) {
if e := checkDiskFree(s.diskPath, s.minFreeDisk); e != nil {
return nil, e
}
filePath := filepath.Join(getVolumeDir(s.diskPath, volume), path)
// Creates a safe file.
return safe.CreateFileWithPrefix(filePath, "$tmpfile")
}
// StatFile - get file info.
func (s localStorage) StatFile(volume, path string) (file FileInfo, err error) {
filePath := filepath.Join(getVolumeDir(s.diskPath, volume), path)
st, e := os.Stat(filePath)
if e != nil {
return FileInfo{}, e
}
file = FileInfo{
Volume: volume,
Name: st.Name(),
ModTime: st.ModTime(),
Size: st.Size(),
Type: st.Mode(),
}
return file, nil
}
// deleteFile - delete file path if its empty.
func deleteFile(basePath, deletePath, volume, path string) error {
if basePath == deletePath {
return nil
}
// Verify if the path exists.
pathSt, e := os.Stat(deletePath)
if e != nil {
return e
}
if pathSt.IsDir() {
// Verify if directory is empty.
empty, e := isDirEmpty(deletePath)
if e != nil {
return e
}
if !empty {
return nil
}
}
// Attempt to remove path.
if e := os.Remove(deletePath); e != nil {
return e
}
// Recursively go down the next path and delete again.
if e := deleteFile(basePath, filepath.Dir(deletePath), volume, path); e != nil {
return e
}
return nil
}
// DeleteFile - delete a file at path.
func (s localStorage) DeleteFile(volume, path string) error {
volumeDir := getVolumeDir(s.diskPath, volume)
// Following code is needed so that we retain "/" suffix if any
// in path argument. Do not use filepath.Join() since it would
// strip off any suffixes.
filePath := s.diskPath + string(os.PathSeparator) + volume + string(os.PathSeparator) + path
// Convert to platform friendly paths.
filePath = filepath.FromSlash(filePath)
// Delete file and delete parent directory as well if its empty.
return deleteFile(volumeDir, filePath, volume, path)
}
Loading…
Cancel
Save