Resurrect back fileystem driver

pkg/storage/drivers/api_testsuite.go

@@ -59,7 +59,7 @@ func testCreateBucket(c *check.C, create func() Driver) {
 func testMultipartObjectCreation(c *check.C, create func() Driver) {
 	drivers := create()
 	switch {
-	case reflect.TypeOf(drivers).String() == "*donut.donutDriver":
+	case reflect.TypeOf(drivers).String() != "*memory.memoryDriver":
 		return
 	}
 	err := drivers.CreateBucket("bucket", "")
@@ -93,7 +93,7 @@ func testMultipartObjectCreation(c *check.C, create func() Driver) {
 func testMultipartObjectAbort(c *check.C, create func() Driver) {
 	drivers := create()
 	switch {
-	case reflect.TypeOf(drivers).String() == "*donut.donutDriver":
+	case reflect.TypeOf(drivers).String() != "*memory.memoryDriver":
 		return
 	}
 	err := drivers.CreateBucket("bucket", "")
@@ -257,13 +257,12 @@ func testPaging(c *check.C, create func() Driver) {
 		resources.CommonPrefixes = prefixes // allocate new everytime
 		resources.Prefix = ""
 		resources.Marker = "newPrefix"
-		resources.Delimiter = "/"
+		resources.Delimiter = ""
 		resources.Maxkeys = 3
 		objects, resources, err = drivers.ListObjects("bucket", resources)
 		c.Assert(objects[0].Key, check.Equals, "newPrefix2")
 		c.Assert(objects[1].Key, check.Equals, "obj0")
 		c.Assert(objects[2].Key, check.Equals, "obj1")
-		c.Assert(resources.CommonPrefixes[0], check.Equals, "this/")
 	}
 	// check ordering of results with prefix
 	{

pkg/storage/drivers/donut/donut_test.go

@@ -35,7 +35,7 @@ func (s *MySuite) TestAPISuite(c *C) {
 	var storageList []string
 	create := func() drivers.Driver {
 		var paths []string
-		p, err := ioutil.TempDir(os.TempDir(), "minio-fs-")
+		p, err := ioutil.TempDir(os.TempDir(), "minio-donut-")
 		c.Check(err, IsNil)
 		storageList = append(storageList, p)
 		paths = append(paths, p)

pkg/storage/drivers/driver.go

@@ -163,10 +163,6 @@ type BucketResourcesMetadata struct {
 	IsTruncated    bool
 	CommonPrefixes []string
 	Mode           FilterMode
-
-	// TODO
-	Logging      string
-	Notification string
 }
 
 // GetMode - Populate filter mode

pkg/storage/drivers/fs/fs.go

@@ -0,0 +1,46 @@
+/*
+ * Mini Object Storage, (C) 2015 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 filesystem
+
+import (
+	"os"
+	"sync"
+
+	"github.com/minio/minio/pkg/storage/drivers"
+)
+
+type fsDriver struct {
+	root string
+	lock *sync.Mutex
+}
+
+// Start filesystem channel
+func Start(root string) (chan<- string, <-chan error, drivers.Driver) {
+	ctrlChannel := make(chan string)
+	errorChannel := make(chan error)
+	fs := new(fsDriver)
+	fs.root = root
+	fs.lock = new(sync.Mutex)
+	go start(ctrlChannel, errorChannel, fs)
+	return ctrlChannel, errorChannel, fs
+}
+
+func start(ctrlChannel <-chan string, errorChannel chan<- error, fs *fsDriver) {
+	err := os.MkdirAll(fs.root, 0700)
+	errorChannel <- err
+	close(errorChannel)
+}

pkg/storage/drivers/fs/fs_bucket.go

@@ -0,0 +1,195 @@
+/*
+ * Mini Object Fs, (C) 2015 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 filesystem
+
+import (
+	"os"
+	"path"
+	"sort"
+
+	"io/ioutil"
+	"path/filepath"
+
+	"github.com/minio/minio/pkg/iodine"
+	"github.com/minio/minio/pkg/storage/drivers"
+)
+
+/// Bucket Operations
+
+// ListBuckets - Get service
+func (fs *fsDriver) ListBuckets() ([]drivers.BucketMetadata, error) {
+	files, err := ioutil.ReadDir(fs.root)
+	if err != nil {
+		return []drivers.BucketMetadata{}, iodine.New(err, nil)
+	}
+
+	var metadataList []drivers.BucketMetadata
+	for _, file := range files {
+		if !file.IsDir() {
+			return []drivers.BucketMetadata{}, iodine.New(drivers.BackendCorrupted{Path: fs.root}, nil)
+		}
+		metadata := drivers.BucketMetadata{
+			Name:    file.Name(),
+			Created: file.ModTime(), // TODO - provide real created time
+		}
+		metadataList = append(metadataList, metadata)
+	}
+	return metadataList, nil
+}
+
+// CreateBucket - PUT Bucket
+func (fs *fsDriver) CreateBucket(bucket, acl string) error {
+	fs.lock.Lock()
+	defer fs.lock.Unlock()
+
+	// verify bucket path legal
+	if drivers.IsValidBucket(bucket) == false {
+		return iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+
+	// get bucket path
+	bucketDir := path.Join(fs.root, bucket)
+
+	// check if bucket exists
+	if _, err := os.Stat(bucketDir); err == nil {
+		return iodine.New(drivers.BucketExists{
+			Bucket: bucket,
+		}, nil)
+	}
+
+	// make bucket
+	err := os.Mkdir(bucketDir, 0700)
+	if err != nil {
+		return iodine.New(err, nil)
+	}
+	return nil
+}
+
+// GetBucketMetadata -
+func (fs *fsDriver) GetBucketMetadata(bucket string) (drivers.BucketMetadata, error) {
+	fs.lock.Lock()
+	defer fs.lock.Unlock()
+	if !drivers.IsValidBucket(bucket) {
+		return drivers.BucketMetadata{}, iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+	// get bucket path
+	bucketDir := path.Join(fs.root, bucket)
+	bucketMetadata := drivers.BucketMetadata{}
+
+	fi, err := os.Stat(bucketDir)
+	// check if bucket exists
+	if err == nil {
+		bucketMetadata.Name = fi.Name()
+		bucketMetadata.Created = fi.ModTime()
+		// TODO convert os.FileMode to meaningful ACL's
+		bucketMetadata.ACL = drivers.BucketACL("private")
+		return bucketMetadata, nil
+	}
+	return drivers.BucketMetadata{}, iodine.New(err, nil)
+}
+
+// aclToPerm - convert acl to filesystem mode
+func aclToPerm(acl string) os.FileMode {
+	switch acl {
+	case "private":
+		return os.FileMode(0700)
+	case "public-read":
+		return os.FileMode(0500)
+	case "public-read-write":
+		return os.FileMode(0777)
+	case "authenticated-read":
+		return os.FileMode(0770)
+	default:
+		return os.FileMode(0700)
+	}
+}
+
+// SetBucketMetadata -
+func (fs *fsDriver) SetBucketMetadata(bucket, acl string) error {
+	fs.lock.Lock()
+	defer fs.lock.Unlock()
+	if !drivers.IsValidBucket(bucket) {
+		return iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+	if !drivers.IsValidBucketACL(acl) {
+		return iodine.New(drivers.InvalidACL{ACL: acl}, nil)
+	}
+	// get bucket path
+	bucketDir := path.Join(fs.root, bucket)
+	err := os.Chmod(bucketDir, aclToPerm(acl))
+	if err != nil {
+		return iodine.New(err, nil)
+	}
+	return nil
+}
+
+// ListObjects - GET bucket (list objects)
+func (fs *fsDriver) ListObjects(bucket string, resources drivers.BucketResourcesMetadata) ([]drivers.ObjectMetadata, drivers.BucketResourcesMetadata, error) {
+	p := bucketDir{}
+	p.files = make(map[string]os.FileInfo)
+
+	if drivers.IsValidBucket(bucket) == false {
+		return []drivers.ObjectMetadata{}, resources, iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+	if resources.Prefix != "" && drivers.IsValidObjectName(resources.Prefix) == false {
+		return []drivers.ObjectMetadata{}, resources, iodine.New(drivers.ObjectNameInvalid{Bucket: bucket, Object: resources.Prefix}, nil)
+	}
+
+	rootPrefix := path.Join(fs.root, bucket)
+	// check bucket exists
+	if _, err := os.Stat(rootPrefix); os.IsNotExist(err) {
+		return []drivers.ObjectMetadata{}, resources, iodine.New(drivers.BucketNotFound{Bucket: bucket}, nil)
+	}
+
+	p.root = rootPrefix
+	err := filepath.Walk(rootPrefix, p.getAllFiles)
+	if err != nil {
+		return []drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+	}
+
+	var metadataList []drivers.ObjectMetadata
+	var metadata drivers.ObjectMetadata
+
+	// Populate filtering mode
+	resources.Mode = drivers.GetMode(resources)
+
+	var fileNames []string
+	for name := range p.files {
+		fileNames = append(fileNames, name)
+	}
+	sort.Strings(fileNames)
+	for _, name := range fileNames {
+		if len(metadataList) >= resources.Maxkeys {
+			resources.IsTruncated = true
+			if resources.IsTruncated && resources.IsDelimiterSet() {
+				resources.NextMarker = metadataList[len(metadataList)-1].Key
+			}
+			break
+		}
+		if name > resources.Marker {
+			metadata, resources, err = fs.filter(bucket, name, p.files[name], resources)
+			if err != nil {
+				return []drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+			}
+			if metadata.Bucket != "" {
+				metadataList = append(metadataList, metadata)
+			}
+		}
+	}
+	sort.Sort(byObjectKey(metadataList))
+	return metadataList, resources, nil
+}

pkg/storage/drivers/fs/fs_common.go

@@ -0,0 +1,77 @@
+/*
+ * Mini Object Storage, (C) 2015 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 filesystem
+
+import (
+	"bufio"
+	"bytes"
+	"os"
+	"strings"
+
+	"github.com/minio/minio/pkg/storage/drivers"
+)
+
+// Metadata - carries metadata about object
+type Metadata struct {
+	Md5sum      []byte
+	ContentType string
+}
+
+func appendUniq(slice []string, i string) []string {
+	for _, ele := range slice {
+		if ele == i {
+			return slice
+		}
+	}
+	return append(slice, i)
+}
+
+type bucketDir struct {
+	files map[string]os.FileInfo
+	root  string
+}
+
+func (p *bucketDir) getAllFiles(object string, fl os.FileInfo, err error) error {
+	if err != nil {
+		return err
+	}
+	if fl.Mode().IsRegular() {
+		if strings.HasSuffix(object, "$metadata") {
+			return nil
+		}
+		_p := strings.Split(object, p.root+"/")
+		if len(_p) > 1 {
+			p.files[_p[1]] = fl
+		}
+	}
+	return nil
+}
+
+func delimiter(object, delimiter string) string {
+	readBuffer := bytes.NewBufferString(object)
+	reader := bufio.NewReader(readBuffer)
+	stringReader := strings.NewReader(delimiter)
+	delimited, _ := stringReader.ReadByte()
+	delimitedStr, _ := reader.ReadString(delimited)
+	return delimitedStr
+}
+
+type byObjectKey []drivers.ObjectMetadata
+
+func (b byObjectKey) Len() int           { return len(b) }
+func (b byObjectKey) Swap(i, j int)      { b[i], b[j] = b[j], b[i] }
+func (b byObjectKey) Less(i, j int) bool { return b[i].Key < b[j].Key }

pkg/storage/drivers/fs/fs_filter.go

@@ -0,0 +1,88 @@
+/*
+ * Mini Object Storage, (C) 2015 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 filesystem
+
+import (
+	"os"
+	"strings"
+
+	"github.com/minio/minio/pkg/iodine"
+	"github.com/minio/minio/pkg/storage/drivers"
+)
+
+func (fs *fsDriver) filter(bucket, name string, file os.FileInfo, resources drivers.BucketResourcesMetadata) (drivers.ObjectMetadata, drivers.BucketResourcesMetadata, error) {
+	var err error
+	var metadata drivers.ObjectMetadata
+
+	switch true {
+	// Both delimiter and Prefix is present
+	case resources.IsDelimiterPrefixSet():
+		if strings.HasPrefix(name, resources.Prefix) {
+			trimmedName := strings.TrimPrefix(name, resources.Prefix)
+			delimitedName := delimiter(trimmedName, resources.Delimiter)
+			switch true {
+			case name == resources.Prefix:
+				// Use resources.Prefix to filter out delimited files
+				metadata, err = fs.GetObjectMetadata(bucket, name)
+				if err != nil {
+					return drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+				}
+			case delimitedName == file.Name():
+				// Use resources.Prefix to filter out delimited files
+				metadata, err = fs.GetObjectMetadata(bucket, name)
+				if err != nil {
+					return drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+				}
+			case delimitedName != "":
+				resources.CommonPrefixes = appendUniq(resources.CommonPrefixes, resources.Prefix+delimitedName)
+			}
+		}
+	// Delimiter present and Prefix is absent
+	case resources.IsDelimiterSet():
+		delimitedName := delimiter(name, resources.Delimiter)
+		switch true {
+		case delimitedName == "":
+			metadata, err = fs.GetObjectMetadata(bucket, name)
+			if err != nil {
+				return drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+			}
+		case delimitedName == file.Name():
+			metadata, err = fs.GetObjectMetadata(bucket, name)
+			if err != nil {
+				return drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+			}
+		case delimitedName != "":
+			resources.CommonPrefixes = appendUniq(resources.CommonPrefixes, delimitedName)
+		}
+	// Delimiter is absent and only Prefix is present
+	case resources.IsPrefixSet():
+		if strings.HasPrefix(name, resources.Prefix) {
+			// Do not strip prefix object output
+			metadata, err = fs.GetObjectMetadata(bucket, name)
+			if err != nil {
+				return drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+			}
+		}
+	case resources.IsDefault():
+		metadata, err = fs.GetObjectMetadata(bucket, name)
+		if err != nil {
+			return drivers.ObjectMetadata{}, resources, iodine.New(err, nil)
+		}
+	}
+
+	return metadata, resources, nil
+}

pkg/storage/drivers/fs/fs_multipart.go

@@ -0,0 +1,33 @@
+package filesystem
+
+import (
+	"errors"
+	"io"
+
+	"github.com/minio/minio/pkg/iodine"
+	"github.com/minio/minio/pkg/storage/drivers"
+)
+
+func (fs *fsDriver) ListMultipartUploads(bucket string, resources drivers.BucketMultipartResourcesMetadata) (drivers.BucketMultipartResourcesMetadata, error) {
+	return drivers.BucketMultipartResourcesMetadata{}, iodine.New(errors.New("Not Implemented"), nil)
+}
+
+func (fs *fsDriver) NewMultipartUpload(bucket, key, contentType string) (string, error) {
+	return "", iodine.New(errors.New("Not Implemented"), nil)
+}
+
+func (fs *fsDriver) CreateObjectPart(bucket, key, uploadID string, partID int, contentType, expectedMD5Sum string, size int64, data io.Reader) (string, error) {
+	return "", iodine.New(errors.New("Not Implemented"), nil)
+}
+
+func (fs *fsDriver) CompleteMultipartUpload(bucket, key, uploadID string, parts map[int]string) (string, error) {
+	return "", iodine.New(errors.New("Not Implemented"), nil)
+}
+
+func (fs *fsDriver) ListObjectParts(bucket, key string, resources drivers.ObjectResourcesMetadata) (drivers.ObjectResourcesMetadata, error) {
+	return drivers.ObjectResourcesMetadata{}, iodine.New(errors.New("Not Implemented"), nil)
+}
+
+func (fs *fsDriver) AbortMultipartUpload(bucket, key, uploadID string) error {
+	return iodine.New(errors.New("Not Implemented"), nil)
+}

pkg/storage/drivers/fs/fs_object.go

@@ -0,0 +1,296 @@
+/*
+ * Mini Object Storage, (C) 2015 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 filesystem
+
+import (
+	"bytes"
+	"io"
+	"os"
+	"path"
+	"strings"
+
+	"crypto/md5"
+	"encoding/base64"
+	"encoding/gob"
+	"encoding/hex"
+	"errors"
+
+	"github.com/minio/minio/pkg/iodine"
+	"github.com/minio/minio/pkg/storage/drivers"
+)
+
+/// Object Operations
+
+// GetPartialObject - GET object from range
+func (fs *fsDriver) GetPartialObject(w io.Writer, bucket, object string, start, length int64) (int64, error) {
+	// validate bucket
+	if drivers.IsValidBucket(bucket) == false {
+		return 0, iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+
+	// validate object
+	if drivers.IsValidObjectName(object) == false {
+		return 0, iodine.New(drivers.ObjectNameInvalid{Bucket: bucket, Object: object}, nil)
+	}
+
+	objectPath := path.Join(fs.root, bucket, object)
+	filestat, err := os.Stat(objectPath)
+	switch err := err.(type) {
+	case nil:
+		{
+			if filestat.IsDir() {
+				return 0, iodine.New(drivers.ObjectNotFound{Bucket: bucket, Object: object}, nil)
+			}
+		}
+	default:
+		{
+			if os.IsNotExist(err) {
+				return 0, iodine.New(drivers.ObjectNotFound{Bucket: bucket, Object: object}, nil)
+			}
+			return 0, iodine.New(err, nil)
+		}
+	}
+	file, err := os.Open(objectPath)
+	if err != nil {
+		return 0, iodine.New(err, nil)
+	}
+	defer file.Close()
+
+	_, err = file.Seek(start, os.SEEK_SET)
+	if err != nil {
+		return 0, iodine.New(err, nil)
+	}
+
+	count, err := io.CopyN(w, file, length)
+	if err != nil {
+		return count, iodine.New(err, nil)
+	}
+
+	return count, nil
+}
+
+// GetObject - GET object from key
+func (fs *fsDriver) GetObject(w io.Writer, bucket string, object string) (int64, error) {
+	// validate bucket
+	if drivers.IsValidBucket(bucket) == false {
+		return 0, iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+
+	// validate object
+	if drivers.IsValidObjectName(object) == false {
+		return 0, iodine.New(drivers.ObjectNameInvalid{Bucket: bucket, Object: object}, nil)
+	}
+
+	objectPath := path.Join(fs.root, bucket, object)
+	filestat, err := os.Stat(objectPath)
+	switch err := err.(type) {
+	case nil:
+		{
+			if filestat.IsDir() {
+				return 0, iodine.New(drivers.ObjectNotFound{Bucket: bucket, Object: object}, nil)
+			}
+		}
+	default:
+		{
+			if os.IsNotExist(err) {
+				return 0, iodine.New(drivers.ObjectNotFound{Bucket: bucket, Object: object}, nil)
+			}
+			return 0, iodine.New(err, nil)
+		}
+	}
+	file, err := os.Open(objectPath)
+	defer file.Close()
+	if err != nil {
+		return 0, drivers.EmbedError(bucket, object, err)
+	}
+
+	count, err := io.Copy(w, file)
+	if err != nil {
+		return count, iodine.New(err, nil)
+	}
+	return count, nil
+}
+
+// GetObjectMetadata - HEAD object
+func (fs *fsDriver) GetObjectMetadata(bucket, object string) (drivers.ObjectMetadata, error) {
+	if drivers.IsValidBucket(bucket) == false {
+		return drivers.ObjectMetadata{}, iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+
+	if drivers.IsValidObjectName(object) == false {
+		return drivers.ObjectMetadata{}, iodine.New(drivers.ObjectNameInvalid{Bucket: bucket, Object: bucket}, nil)
+	}
+
+	// Do not use path.Join() since path.Join strips off any object names with '/', use them as is
+	// in a static manner so that we can send a proper 'ObjectNotFound' reply back upon os.Stat()
+	objectPath := fs.root + "/" + bucket + "/" + object
+	stat, err := os.Stat(objectPath)
+	if os.IsNotExist(err) {
+		return drivers.ObjectMetadata{}, iodine.New(drivers.ObjectNotFound{Bucket: bucket, Object: object}, nil)
+	}
+
+	_, err = os.Stat(objectPath + "$metadata")
+	if os.IsNotExist(err) {
+		return drivers.ObjectMetadata{}, iodine.New(drivers.ObjectNotFound{Bucket: bucket, Object: object}, nil)
+	}
+
+	file, err := os.Open(objectPath + "$metadata")
+	defer file.Close()
+	if err != nil {
+		return drivers.ObjectMetadata{}, iodine.New(err, nil)
+	}
+
+	var deserializedMetadata Metadata
+	decoder := gob.NewDecoder(file)
+	err = decoder.Decode(&deserializedMetadata)
+	if err != nil {
+		return drivers.ObjectMetadata{}, iodine.New(err, nil)
+	}
+
+	contentType := "application/octet-stream"
+	if deserializedMetadata.ContentType != "" {
+		contentType = deserializedMetadata.ContentType
+	}
+	contentType = strings.TrimSpace(contentType)
+
+	etag := bucket + "#" + path.Base(object)
+	if len(deserializedMetadata.Md5sum) != 0 {
+		etag = hex.EncodeToString(deserializedMetadata.Md5sum)
+	}
+
+	metadata := drivers.ObjectMetadata{
+		Bucket:      bucket,
+		Key:         object,
+		Created:     stat.ModTime(),
+		Size:        stat.Size(),
+		Md5:         etag,
+		ContentType: contentType,
+	}
+
+	return metadata, nil
+}
+
+// isMD5SumEqual - returns error if md5sum mismatches, success its `nil`
+func isMD5SumEqual(expectedMD5Sum, actualMD5Sum string) error {
+	if strings.TrimSpace(expectedMD5Sum) != "" && strings.TrimSpace(actualMD5Sum) != "" {
+		expectedMD5SumBytes, err := hex.DecodeString(expectedMD5Sum)
+		if err != nil {
+			return iodine.New(err, nil)
+		}
+		actualMD5SumBytes, err := hex.DecodeString(actualMD5Sum)
+		if err != nil {
+			return iodine.New(err, nil)
+		}
+		if !bytes.Equal(expectedMD5SumBytes, actualMD5SumBytes) {
+			return iodine.New(errors.New("bad digest, md5sum mismatch"), nil)
+		}
+		return nil
+	}
+	return iodine.New(errors.New("invalid argument"), nil)
+}
+
+// CreateObject - PUT object
+func (fs *fsDriver) CreateObject(bucket, key, contentType, expectedMD5Sum string, size int64, data io.Reader) (string, error) {
+	fs.lock.Lock()
+	defer fs.lock.Unlock()
+
+	// check bucket name valid
+	if drivers.IsValidBucket(bucket) == false {
+		return "", iodine.New(drivers.BucketNameInvalid{Bucket: bucket}, nil)
+	}
+
+	// check bucket exists
+	if _, err := os.Stat(path.Join(fs.root, bucket)); os.IsNotExist(err) {
+		return "", iodine.New(drivers.BucketNotFound{Bucket: bucket}, nil)
+	}
+
+	// verify object path legal
+	if drivers.IsValidObjectName(key) == false {
+		return "", iodine.New(drivers.ObjectNameInvalid{Bucket: bucket, Object: key}, nil)
+	}
+
+	// verify content type
+	if contentType == "" {
+		contentType = "application/octet-stream"
+	}
+	contentType = strings.TrimSpace(contentType)
+
+	// get object path
+	objectPath := path.Join(fs.root, bucket, key)
+	objectDir := path.Dir(objectPath)
+	if _, err := os.Stat(objectDir); os.IsNotExist(err) {
+		err = os.MkdirAll(objectDir, 0700)
+		if err != nil {
+			return "", iodine.New(err, nil)
+		}
+	}
+
+	// check if object exists
+	if _, err := os.Stat(objectPath); !os.IsNotExist(err) {
+		return "", iodine.New(drivers.ObjectExists{
+			Bucket: bucket,
+			Object: key,
+		}, nil)
+	}
+
+	if strings.TrimSpace(expectedMD5Sum) != "" {
+		expectedMD5SumBytes, err := base64.StdEncoding.DecodeString(strings.TrimSpace(expectedMD5Sum))
+		if err != nil {
+			// pro-actively close the connection
+			return "", iodine.New(drivers.InvalidDigest{Md5: expectedMD5Sum}, nil)
+		}
+		expectedMD5Sum = hex.EncodeToString(expectedMD5SumBytes)
+	}
+
+	// write object
+	file, err := os.OpenFile(objectPath, os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		return "", iodine.New(err, nil)
+	}
+	defer file.Close()
+
+	h := md5.New()
+	mw := io.MultiWriter(file, h)
+
+	_, err = io.Copy(mw, data)
+	if err != nil {
+		return "", iodine.New(err, nil)
+	}
+
+	file, err = os.OpenFile(objectPath+"$metadata", os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		return "", iodine.New(err, nil)
+	}
+	defer file.Close()
+
+	metadata := &Metadata{
+		ContentType: contentType,
+		Md5sum:      h.Sum(nil),
+	}
+	// serialize metadata to gob
+	encoder := gob.NewEncoder(file)
+	err = encoder.Encode(metadata)
+
+	md5Sum := hex.EncodeToString(metadata.Md5sum)
+	// Verify if the written object is equal to what is expected, only if it is requested as such
+	if strings.TrimSpace(expectedMD5Sum) != "" {
+		if err := isMD5SumEqual(strings.TrimSpace(expectedMD5Sum), md5Sum); err != nil {
+			return "", iodine.New(drivers.BadDigest{Md5: expectedMD5Sum, Bucket: bucket, Key: key}, nil)
+		}
+	}
+	return md5Sum, nil
+}

pkg/storage/drivers/fs/fs_test.go

@@ -0,0 +1,54 @@
+/*
+ * Mini Object Storage, (C) 2015 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 filesystem
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+
+	. "github.com/minio/check"
+
+	"github.com/minio/minio/pkg/storage/drivers"
+)
+
+func Test(t *testing.T) { TestingT(t) }
+
+type MySuite struct{}
+
+var _ = Suite(&MySuite{})
+
+func (s *MySuite) TestAPISuite(c *C) {
+	var storageList []string
+	defer removeRoots(c, storageList)
+	create := func() drivers.Driver {
+		path, err := ioutil.TempDir(os.TempDir(), "minio-fs-")
+		c.Check(err, IsNil)
+		storageList = append(storageList, path)
+		_, _, store := Start(path)
+		return store
+	}
+	drivers.APITestSuite(c, create)
+
+}
+
+func removeRoots(c *C, roots []string) {
+	for _, root := range roots {
+		err := os.RemoveAll(root)
+		c.Check(err, IsNil)
+	}
+}