You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
minio/pkg/storage/fs/fs.go

231 lines
5.8 KiB

10 years ago
package fs
import (
"errors"
"io"
"io/ioutil"
"os"
"path"
"strings"
"sync"
mstorage "github.com/minio-io/minio/pkg/storage"
)
type storage struct {
root string
writeLock sync.Mutex
}
type MkdirFailedError struct{}
func (self MkdirFailedError) Error() string {
return "Mkdir Failed"
}
func Start(root string) (chan<- string, <-chan error, *storage) {
ctrlChannel := make(chan string)
errorChannel := make(chan error)
go start(ctrlChannel, errorChannel)
return ctrlChannel, errorChannel, &storage{root: root}
}
func start(ctrlChannel <-chan string, errorChannel chan<- error) {
close(errorChannel)
}
// Bucket Operations
10 years ago
func (storage *storage) ListBuckets(prefix string) ([]mstorage.BucketMetadata, error) {
if mstorage.IsValidBucket(bucket) == false {
return []mstorage.BucketMetadata{}, mstorage.BucketNameInvalid{Bucket: bucket}
}
files, err := ioutil.ReadDir(storage.root)
if err != nil {
return []mstorage.BucketMetadata{}, mstorage.EmbedError("bucket", "", err)
}
var metadataList []mstorage.BucketMetadata
for _, file := range files {
if !file.IsDir() {
return []mstorage.BucketMetadata{}, mstorage.BackendCorrupted{Path: storage.root}
}
if strings.HasPrefix(file.Name(), prefix) {
metadata := mstorage.BucketMetadata{
Name: file.Name(),
Created: file.ModTime(), // TODO - provide real created time
}
metadataList = append(metadata, metadataList)
}
}
return metadataList, nil
10 years ago
}
func (storage *storage) StoreBucket(bucket string) error {
storage.writeLock.Lock()
defer storage.writeLock.Unlock()
// verify bucket path legal
if mstorage.IsValidBucket(bucket) == false {
return mstorage.BucketNameInvalid{Bucket: bucket}
}
// get bucket path
bucketDir := path.Join(storage.root, bucket)
// check if bucket exists
if _, err := os.Stat(bucketDir); err == nil {
return mstorage.BucketExists{
Bucket: bucket,
}
}
// make bucket
err := os.Mkdir(bucketDir, 0700)
if err != nil {
return mstorage.EmbedError(bucket, "", err)
}
return nil
}
// Object Operations
func (storage *storage) CopyObjectToWriter(w io.Writer, bucket string, object string) (int64, error) {
// validate bucket
if mstorage.IsValidBucket(bucket) == false {
return 0, mstorage.BucketNameInvalid{Bucket: bucket}
}
// validate object
if mstorage.IsValidObject(object) == false {
return 0, mstorage.ObjectNameInvalid{Bucket: bucket, Object: object}
}
objectPath := path.Join(storage.root, bucket, object)
file, err := os.Open(objectPath)
if err != nil {
return 0, mstorage.EmbedError(bucket, object, err)
}
count, err := io.Copy(w, file)
if err != nil {
return count, mstorage.EmbedError(bucket, object, err)
}
return count, nil
}
func (storage *storage) GetObjectMetadata(bucket string, object string) (mstorage.ObjectMetadata, error) {
if mstorage.IsValidBucket(bucket) == false {
return mstorage.ObjectMetadata{}, mstorage.BucketNameInvalid{Bucket: bucket}
}
if mstorage.IsValidObject(bucket) == false {
return mstorage.ObjectMetadata{}, mstorage.ObjectNameInvalid{Bucket: bucket, Object: bucket}
}
objectPath := path.Join(storage.root, bucket, object)
stat, err := os.Stat(objectPath)
if os.IsNotExist(err) {
return mstorage.ObjectMetadata{}, mstorage.ObjectNotFound{Bucket: bucket, Object: object}
}
metadata := mstorage.ObjectMetadata{
Bucket: bucket,
Key: object,
Created: stat.ModTime(),
Size: stat.Size(),
ETag: bucket + "#" + object,
}
return metadata, nil
}
func (storage *storage) ListObjects(bucket, prefix string, count int) ([]mstorage.ObjectMetadata, bool, error) {
if mstorage.IsValidBucket(bucket) == false {
return []mstorage.ObjectMetadata{}, false, mstorage.BucketNameInvalid{Bucket: bucket}
}
if mstorage.IsValidObject(prefix) == false {
return []mstorage.ObjectMetadata{}, false, mstorage.ObjectNameInvalid{Bucket: bucket, Object: prefix}
}
rootPrefix := path.Join(storage.root, bucket)
files, err := ioutil.ReadDir(rootPrefix)
if err != nil {
return []mstorage.ObjectMetadata{}, false, mstorage.EmbedError("bucket", "", err)
}
var metadataList []mstorage.ObjectMetadata
for _, file := range files {
if len(metadataList) >= count {
return metadataList, true, nil
}
if strings.HasPrefix(file.Name(), prefix) {
metadata := mstorage.ObjectMetadata{
Bucket: bucket,
Key: file.Name(),
Created: file.ModTime(),
Size: file.Size(),
ETag: bucket + "#" + file.Name(),
}
metadataList = append(metadataList, metadata)
}
}
return metadataList, false, nil
}
func (storage *storage) StoreObject(bucket string, key string, data io.Reader) error {
// TODO Commits should stage then move instead of writing directly
storage.writeLock.Lock()
defer storage.writeLock.Unlock()
// check bucket name valid
if mstorage.IsValidBucket(bucket) == false {
return mstorage.BucketNameInvalid{Bucket: bucket}
}
// check bucket exists
if _, err := os.Stat(path.Join(storage.root, bucket)); os.IsNotExist(err) {
return mstorage.BucketNotFound{Bucket: bucket}
}
// verify object path legal
if mstorage.IsValidObject(key) == false {
return mstorage.ObjectNameInvalid{Bucket: bucket, Object: key}
}
// get object path
objectPath := path.Join(storage.root, bucket, key)
objectDir := path.Dir(objectPath)
if _, err := os.Stat(objectDir); os.IsNotExist(err) {
err = os.MkdirAll(objectDir, 0700)
if err != nil {
return mstorage.EmbedError(bucket, key, err)
}
}
10 years ago
// check if object exists
if _, err := os.Stat(objectPath); !os.IsNotExist(err) {
return mstorage.ObjectExists{
Bucket: bucket,
Key: key,
}
}
// write object
file, err := os.OpenFile(objectPath, os.O_WRONLY|os.O_CREATE, 0600)
defer file.Close()
if err != nil {
return mstorage.EmbedError(bucket, key, err)
}
_, err = io.Copy(file, data)
if err != nil {
return mstorage.EmbedError(bucket, key, err)
}
return nil
}