/* * 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 fs import ( "bufio" "bytes" "crypto/md5" "encoding/gob" "encoding/hex" "encoding/json" "io" "io/ioutil" "os" "path" "path/filepath" "sort" "strings" "sync" mstorage "github.com/minio-io/minio/pkg/storage" "github.com/minio-io/minio/pkg/utils/policy" ) // Storage - fs local variables type Storage struct { root string lock *sync.Mutex } // Metadata - carries metadata about object type Metadata struct { Md5sum []byte ContentType string } // Start filesystem channel func Start(root string) (chan<- string, <-chan error, *Storage) { ctrlChannel := make(chan string) errorChannel := make(chan error) s := Storage{} s.root = root s.lock = new(sync.Mutex) go start(ctrlChannel, errorChannel, &s) return ctrlChannel, errorChannel, &s } func start(ctrlChannel <-chan string, errorChannel chan<- error, s *Storage) { err := os.MkdirAll(s.root, 0700) errorChannel <- err close(errorChannel) } func appendUniq(slice []string, i string) []string { for _, ele := range slice { if ele == i { return slice } } return append(slice, i) } /// Bucket Operations // ListBuckets - Get service func (storage *Storage) ListBuckets() ([]mstorage.BucketMetadata, error) { files, err := ioutil.ReadDir(storage.root) if err != nil { return []mstorage.BucketMetadata{}, mstorage.EmbedError("bucket", "", err) } var metadataList []mstorage.BucketMetadata for _, file := range files { // Skip policy files if strings.HasSuffix(file.Name(), "_policy.json") { continue } if !file.IsDir() { return []mstorage.BucketMetadata{}, mstorage.BackendCorrupted{Path: storage.root} } metadata := mstorage.BucketMetadata{ Name: file.Name(), Created: file.ModTime(), // TODO - provide real created time } metadataList = append(metadataList, metadata) } return metadataList, nil } // StoreBucket - PUT Bucket func (storage *Storage) StoreBucket(bucket string) error { storage.lock.Lock() defer storage.lock.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 } // GetBucketPolicy - GET bucket policy func (storage *Storage) GetBucketPolicy(bucket string) (interface{}, error) { storage.lock.Lock() defer storage.lock.Unlock() var p policy.BucketPolicy // verify bucket path legal if mstorage.IsValidBucket(bucket) == false { return policy.BucketPolicy{}, 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 policy.BucketPolicy{}, mstorage.BucketNotFound{Bucket: bucket} } // get policy path bucketPolicy := path.Join(storage.root, bucket+"_policy.json") filestat, err := os.Stat(bucketPolicy) if os.IsNotExist(err) { return policy.BucketPolicy{}, mstorage.BucketPolicyNotFound{Bucket: bucket} } if filestat.IsDir() { return policy.BucketPolicy{}, mstorage.BackendCorrupted{Path: bucketPolicy} } file, err := os.OpenFile(bucketPolicy, os.O_RDONLY, 0666) defer file.Close() if err != nil { return policy.BucketPolicy{}, mstorage.EmbedError(bucket, "", err) } encoder := json.NewDecoder(file) err = encoder.Decode(&p) if err != nil { return policy.BucketPolicy{}, mstorage.EmbedError(bucket, "", err) } return p, nil } // StoreBucketPolicy - PUT bucket policy func (storage *Storage) StoreBucketPolicy(bucket string, policy interface{}) error { storage.lock.Lock() defer storage.lock.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.BucketNotFound{ Bucket: bucket, } } // get policy path bucketPolicy := path.Join(storage.root, bucket+"_policy.json") filestat, ret := os.Stat(bucketPolicy) if !os.IsNotExist(ret) { if filestat.IsDir() { return mstorage.BackendCorrupted{Path: bucketPolicy} } } file, err := os.OpenFile(bucketPolicy, os.O_WRONLY|os.O_CREATE, 0600) defer file.Close() if err != nil { return mstorage.EmbedError(bucket, "", err) } encoder := json.NewEncoder(file) err = encoder.Encode(policy) if err != nil { return mstorage.EmbedError(bucket, "", err) } return nil } /// Object Operations // CopyObjectToWriter - GET object 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) filestat, err := os.Stat(objectPath) switch err := err.(type) { case nil: { if filestat.IsDir() { return 0, mstorage.ObjectNotFound{Bucket: bucket, Object: object} } } default: { if os.IsNotExist(err) { return 0, mstorage.ObjectNotFound{Bucket: bucket, Object: object} } return 0, mstorage.EmbedError(bucket, object, err) } } file, err := os.Open(objectPath) defer file.Close() 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 } // GetObjectMetadata - HEAD object func (storage *Storage) GetObjectMetadata(bucket, object, prefix string) (mstorage.ObjectMetadata, error) { if mstorage.IsValidBucket(bucket) == false { return mstorage.ObjectMetadata{}, mstorage.BucketNameInvalid{Bucket: bucket} } if mstorage.IsValidObject(object) == false { return mstorage.ObjectMetadata{}, mstorage.ObjectNameInvalid{Bucket: bucket, Object: bucket} } // 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 := storage.root + "/" + bucket + "/" + object stat, err := os.Stat(objectPath) if os.IsNotExist(err) { return mstorage.ObjectMetadata{}, mstorage.ObjectNotFound{Bucket: bucket, Object: object} } _, err = os.Stat(objectPath + "$metadata") if os.IsNotExist(err) { return mstorage.ObjectMetadata{}, mstorage.ObjectNotFound{Bucket: bucket, Object: object} } file, err := os.Open(objectPath + "$metadata") defer file.Close() if err != nil { return mstorage.ObjectMetadata{}, mstorage.EmbedError(bucket, object, err) } var deserializedMetadata Metadata decoder := gob.NewDecoder(file) err = decoder.Decode(&deserializedMetadata) if err != nil { return mstorage.ObjectMetadata{}, mstorage.EmbedError(bucket, object, err) } 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) } trimmedObject := strings.TrimPrefix(object, prefix) metadata := mstorage.ObjectMetadata{ Bucket: bucket, Key: trimmedObject, Created: stat.ModTime(), Size: stat.Size(), ETag: etag, ContentType: contentType, } return metadata, nil } 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 []mstorage.ObjectMetadata // Len func (b byObjectKey) Len() int { return len(b) } // Swap func (b byObjectKey) Swap(i, j int) { b[i], b[j] = b[j], b[i] } // Less func (b byObjectKey) Less(i, j int) bool { return b[i].Key < b[j].Key } // ListObjects - GET bucket (list objects) func (storage *Storage) ListObjects(bucket string, resources mstorage.BucketResourcesMetadata) ([]mstorage.ObjectMetadata, mstorage.BucketResourcesMetadata, error) { p := bucketDir{} p.files = make(map[string]os.FileInfo) if mstorage.IsValidBucket(bucket) == false { return []mstorage.ObjectMetadata{}, resources, mstorage.BucketNameInvalid{Bucket: bucket} } if resources.Prefix != "" && mstorage.IsValidObject(resources.Prefix) == false { return []mstorage.ObjectMetadata{}, resources, mstorage.ObjectNameInvalid{Bucket: bucket, Object: resources.Prefix} } rootPrefix := path.Join(storage.root, bucket) // check bucket exists if _, err := os.Stat(rootPrefix); os.IsNotExist(err) { return []mstorage.ObjectMetadata{}, resources, mstorage.BucketNotFound{Bucket: bucket} } p.root = rootPrefix err := filepath.Walk(rootPrefix, p.getAllFiles) if err != nil { return []mstorage.ObjectMetadata{}, resources, mstorage.EmbedError(bucket, "", err) } var metadataList []mstorage.ObjectMetadata for name, file := range p.files { if len(metadataList) >= resources.Maxkeys { resources.IsTruncated = true goto ret } // TODO handle resources.Marker switch true { // Delimiter present and Prefix is absent case resources.Delimiter != "" && resources.Prefix == "": delimitedName := delimiter(name, resources.Delimiter) switch true { case delimitedName == "": // Do not strip prefix object output metadata, err := storage.GetObjectMetadata(bucket, name, "") if err != nil { return []mstorage.ObjectMetadata{}, resources, mstorage.EmbedError(bucket, "", err) } metadataList = append(metadataList, metadata) case delimitedName == file.Name(): // Do not strip prefix object output metadata, err := storage.GetObjectMetadata(bucket, name, "") if err != nil { return []mstorage.ObjectMetadata{}, resources, mstorage.EmbedError(bucket, "", err) } metadataList = append(metadataList, metadata) case delimitedName != "": resources.CommonPrefixes = appendUniq(resources.CommonPrefixes, delimitedName) } // Both delimiter and Prefix is present case resources.Delimiter != "" && resources.Prefix != "" && 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 := storage.GetObjectMetadata(bucket, name, resources.Prefix) if err != nil { return []mstorage.ObjectMetadata{}, resources, mstorage.EmbedError(bucket, "", err) } metadataList = append(metadataList, metadata) case delimitedName == file.Name(): // Use resources.Prefix to filter out delimited files metadata, err := storage.GetObjectMetadata(bucket, name, resources.Prefix) if err != nil { return []mstorage.ObjectMetadata{}, resources, mstorage.EmbedError(bucket, "", err) } metadataList = append(metadataList, metadata) case delimitedName != "": if delimitedName == resources.Delimiter { resources.CommonPrefixes = appendUniq(resources.CommonPrefixes, resources.Prefix+delimitedName) } else { resources.CommonPrefixes = appendUniq(resources.CommonPrefixes, delimitedName) } } // Delimiter is absent and only Prefix is present case resources.Delimiter == "" && resources.Prefix != "" && strings.HasPrefix(name, resources.Prefix): // Do not strip prefix object output metadata, err := storage.GetObjectMetadata(bucket, name, "") if err != nil { return []mstorage.ObjectMetadata{}, resources, mstorage.EmbedError(bucket, "", err) } metadataList = append(metadataList, metadata) case resources.Prefix == "" && resources.Delimiter == "": metadata, err := storage.GetObjectMetadata(bucket, name, "") if err != nil { return []mstorage.ObjectMetadata{}, resources, mstorage.EmbedError(bucket, "", err) } metadataList = append(metadataList, metadata) } } ret: sort.Sort(byObjectKey(metadataList)) return metadataList, resources, nil } // StoreObject - PUT object func (storage *Storage) StoreObject(bucket, key, contentType string, data io.Reader) error { // TODO Commits should stage then move instead of writing directly storage.lock.Lock() defer storage.lock.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} } // verify content type if contentType == "" { contentType = "application/octet-stream" } contentType = strings.TrimSpace(contentType) // 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) } } // check if object exists if _, err := os.Stat(objectPath); !os.IsNotExist(err) { return mstorage.ObjectExists{ Bucket: bucket, Object: 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) } h := md5.New() mw := io.MultiWriter(file, h) _, err = io.Copy(mw, data) if err != nil { return mstorage.EmbedError(bucket, key, err) } // file, err = os.OpenFile(objectPath+"$metadata", os.O_WRONLY|os.O_CREATE, 0600) defer file.Close() if err != nil { return mstorage.EmbedError(bucket, key, err) } // serialize metadata to gob encoder := gob.NewEncoder(file) err = encoder.Encode(&Metadata{ ContentType: contentType, Md5sum: h.Sum(nil), }) if err != nil { return mstorage.EmbedError(bucket, key, err) } return nil }