Merge pull request #622 from harshavardhana/pr_out_add_filesystem_factory_functions_and_add_related_cli_options

This commit is contained in:
Harshavardhana 2015-05-30 03:26:58 -07:00
commit 437c2b31dd
13 changed files with 843 additions and 11 deletions

View file

@ -29,7 +29,7 @@ lint:
cyclo:
@echo "Running $@:"
@test -z "$$(gocyclo -over 16 . | grep -v Godeps/_workspace/src/ | tee /dev/stderr)"
@test -z "$$(gocyclo -over 19 . | grep -v Godeps/_workspace/src/ | tee /dev/stderr)"
pre-build:
@echo "Running pre-build:"

34
main.go
View file

@ -42,6 +42,7 @@ var commands = []cli.Command{
var modeCommands = []cli.Command{
memoryCmd,
fsCmd,
donutCmd,
}
@ -70,6 +71,23 @@ EXAMPLES:
`,
}
var fsCmd = cli.Command{
Name: "fs",
Description: "Specify a path to instantiate filesystem driver",
Action: runFilesystem,
CustomHelpTemplate: `NAME:
minio mode {{.Name}} - {{.Description}}
USAGE:
minio mode {{.Name}} limit SIZE expire TIME
EXAMPLES:
1. Export an existing filesystem path
$ minio mode {{.Name}} /var/www
`,
}
var donutCmd = cli.Command{
Name: "donut",
Description: "Specify a path to instantiate donut",
@ -204,7 +222,6 @@ func runDonut(c *cli.Context) {
if len(c.Args()) < 1 {
cli.ShowCommandHelpAndExit(c, "donut", 1) // last argument is exit code
}
// supporting multiple paths
var paths []string
if strings.TrimSpace(c.Args().First()) == "" {
@ -226,6 +243,21 @@ func runDonut(c *cli.Context) {
server.StartMinio(servers)
}
func runFilesystem(c *cli.Context) {
if len(c.Args()) != 1 {
cli.ShowCommandHelpAndExit(c, "fs", 1) // last argument is exit code
}
apiServerConfig := getAPIServerConfig(c)
fsDriver := server.FilesystemFactory{
Config: apiServerConfig,
Path: c.Args()[0],
}
apiServer := fsDriver.GetStartServerFunc()
webServer := getWebServerConfigFunc(c)
servers := []server.StartServerFunc{apiServer, webServer}
server.StartMinio(servers)
}
func getAPIServerConfig(c *cli.Context) httpserver.Config {
certFile := c.String("cert")
keyFile := c.String("key")

View file

@ -27,6 +27,7 @@ import (
"github.com/minio/minio/pkg/iodine"
"github.com/minio/minio/pkg/server/httpserver"
"github.com/minio/minio/pkg/storage/drivers/donut"
fs "github.com/minio/minio/pkg/storage/drivers/fs"
"github.com/minio/minio/pkg/storage/drivers/memory"
"github.com/minio/minio/pkg/utils/log"
)
@ -47,6 +48,21 @@ func (f MemoryFactory) GetStartServerFunc() StartServerFunc {
}
}
// FilesystemFactory is used to build filesystem api servers
type FilesystemFactory struct {
httpserver.Config
Path string
}
// GetStartServerFunc builds memory api servers
func (f FilesystemFactory) GetStartServerFunc() StartServerFunc {
return func() (chan<- string, <-chan error) {
_, _, driver := fs.Start(f.Path)
ctrl, status, _ := httpserver.Start(api.HTTPHandler(driver), f.Config)
return ctrl, status
}
}
// WebFactory is used to build web cli servers
type WebFactory struct {
httpserver.Config

View file

@ -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
{

View file

@ -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)

View file

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

View file

@ -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)
}

View file

@ -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
}

View file

@ -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 }

View file

@ -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
}

View file

@ -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)
}

View file

@ -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
}

View file

@ -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)
}
}