mirror of
https://github.com/minio/minio.git
synced 2024-12-26 23:25:54 -05:00
583e4ecff6
In the common case, GetObject is called on a bucket that exists and an object that exists and is not a directory. It should be optimized for this case, thus error-related syscalls are pushed back until they are necessary. This should not impact performance negatively in the uncommon case, and instead drops two otherwise unnecessary os.Stat's in the common case. The race conditions around a proper error being returned were present beforehand. It also renames 'err' to 'e'.
400 lines
12 KiB
Go
400 lines
12 KiB
Go
/*
|
|
* Minio Cloud 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 (
|
|
"bytes"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
|
|
"crypto/md5"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"runtime"
|
|
|
|
"github.com/minio/minio/pkg/atomic"
|
|
"github.com/minio/minio/pkg/crypto/sha256"
|
|
"github.com/minio/minio/pkg/disk"
|
|
"github.com/minio/minio/pkg/mimedb"
|
|
"github.com/minio/minio/pkg/probe"
|
|
"github.com/minio/minio/pkg/s3/signature4"
|
|
)
|
|
|
|
/// Object Operations
|
|
|
|
// GetObject - GET object
|
|
func (fs Filesystem) GetObject(w io.Writer, bucket, object string, start, length int64) (int64, *probe.Error) {
|
|
// Critical region requiring read lock.
|
|
fs.rwLock.RLock()
|
|
defer fs.rwLock.RUnlock()
|
|
|
|
// Input validation.
|
|
if !IsValidBucketName(bucket) {
|
|
return 0, probe.NewError(BucketNameInvalid{Bucket: bucket})
|
|
}
|
|
if !IsValidObjectName(object) {
|
|
return 0, probe.NewError(ObjectNameInvalid{Bucket: bucket, Object: object})
|
|
}
|
|
|
|
// normalize buckets.
|
|
bucket = fs.denormalizeBucket(bucket)
|
|
objectPath := filepath.Join(fs.path, bucket, object)
|
|
|
|
file, e := os.Open(objectPath)
|
|
if e != nil {
|
|
// If the object doesn't exist, the bucket might not exist either. Stat for
|
|
// the bucket and give a better error message if that is true.
|
|
if os.IsNotExist(e) {
|
|
_, e = os.Stat(filepath.Join(fs.path, bucket))
|
|
if os.IsNotExist(e) {
|
|
return 0, probe.NewError(BucketNotFound{Bucket: bucket})
|
|
}
|
|
|
|
return 0, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
|
|
}
|
|
return 0, probe.NewError(e)
|
|
}
|
|
defer file.Close()
|
|
|
|
_, e = file.Seek(start, os.SEEK_SET)
|
|
if e != nil {
|
|
// When the "handle is invalid", the file might be a directory on Windows.
|
|
if runtime.GOOS == "windows" && strings.Contains(e.Error(), "handle is invalid") {
|
|
return 0, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
|
|
}
|
|
|
|
return 0, probe.NewError(e)
|
|
}
|
|
|
|
var count int64
|
|
// Copy over the whole file if the length is non-positive.
|
|
if length > 0 {
|
|
count, e = io.CopyN(w, file, length)
|
|
} else {
|
|
count, e = io.Copy(w, file)
|
|
}
|
|
|
|
if e != nil {
|
|
// This call will fail if the object is a directory. Stat the file to see if
|
|
// this is true, if so, return an ObjectNotFound error.
|
|
stat, e := os.Stat(objectPath)
|
|
if e == nil && stat.IsDir() {
|
|
return count, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
|
|
}
|
|
|
|
return count, probe.NewError(e)
|
|
}
|
|
|
|
return count, nil
|
|
}
|
|
|
|
// GetObjectInfo - get object info.
|
|
func (fs Filesystem) GetObjectInfo(bucket, object string) (ObjectInfo, *probe.Error) {
|
|
// Input validation.
|
|
if !IsValidBucketName(bucket) {
|
|
return ObjectInfo{}, probe.NewError(BucketNameInvalid{Bucket: bucket})
|
|
}
|
|
|
|
if !IsValidObjectName(object) {
|
|
return ObjectInfo{}, probe.NewError(ObjectNameInvalid{Bucket: bucket, Object: bucket})
|
|
}
|
|
|
|
// Normalize buckets.
|
|
bucket = fs.denormalizeBucket(bucket)
|
|
bucketPath := filepath.Join(fs.path, bucket)
|
|
if _, e := os.Stat(bucketPath); e != nil {
|
|
if os.IsNotExist(e) {
|
|
return ObjectInfo{}, probe.NewError(BucketNotFound{Bucket: bucket})
|
|
}
|
|
return ObjectInfo{}, probe.NewError(e)
|
|
}
|
|
|
|
info, err := getObjectInfo(fs.path, bucket, object)
|
|
if err != nil {
|
|
return ObjectInfo{}, err.Trace(bucket, object)
|
|
}
|
|
return info, nil
|
|
}
|
|
|
|
// getObjectInfo - get object stat info.
|
|
func getObjectInfo(rootPath, bucket, object string) (ObjectInfo, *probe.Error) {
|
|
// Do not use filepath.Join() since filepath.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().
|
|
var objectPath string
|
|
// For windows use its special os.PathSeparator == "\\"
|
|
if runtime.GOOS == "windows" {
|
|
objectPath = rootPath + string(os.PathSeparator) + bucket + string(os.PathSeparator) + object
|
|
} else {
|
|
objectPath = rootPath + string(os.PathSeparator) + bucket + string(os.PathSeparator) + object
|
|
}
|
|
stat, err := os.Stat(objectPath)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return ObjectInfo{}, probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
|
|
}
|
|
return ObjectInfo{}, probe.NewError(err)
|
|
}
|
|
contentType := "application/octet-stream"
|
|
if runtime.GOOS == "windows" {
|
|
object = filepath.ToSlash(object)
|
|
}
|
|
|
|
if objectExt := filepath.Ext(object); objectExt != "" {
|
|
content, ok := mimedb.DB[strings.ToLower(strings.TrimPrefix(objectExt, "."))]
|
|
if ok {
|
|
contentType = content.ContentType
|
|
}
|
|
}
|
|
metadata := ObjectInfo{
|
|
Bucket: bucket,
|
|
Name: object,
|
|
ModifiedTime: stat.ModTime(),
|
|
Size: stat.Size(),
|
|
ContentType: contentType,
|
|
IsDir: stat.Mode().IsDir(),
|
|
}
|
|
return metadata, nil
|
|
}
|
|
|
|
// isMD5SumEqual - returns error if md5sum mismatches, success its `nil`
|
|
func isMD5SumEqual(expectedMD5Sum, actualMD5Sum string) bool {
|
|
// Verify the md5sum.
|
|
if expectedMD5Sum != "" && actualMD5Sum != "" {
|
|
// Decode md5sum to bytes from their hexadecimal
|
|
// representations.
|
|
expectedMD5SumBytes, err := hex.DecodeString(expectedMD5Sum)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
actualMD5SumBytes, err := hex.DecodeString(actualMD5Sum)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
// Verify md5sum bytes are equal after successful decoding.
|
|
if !bytes.Equal(expectedMD5SumBytes, actualMD5SumBytes) {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// CreateObject - create an object.
|
|
func (fs Filesystem) CreateObject(bucket, object, expectedMD5Sum string, size int64, data io.Reader, sig *signature4.Sign) (ObjectInfo, *probe.Error) {
|
|
di, e := disk.GetInfo(fs.path)
|
|
if e != nil {
|
|
return ObjectInfo{}, probe.NewError(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) <= fs.minFreeDisk {
|
|
return ObjectInfo{}, probe.NewError(RootPathFull{Path: fs.path})
|
|
}
|
|
|
|
// Check bucket name valid.
|
|
if !IsValidBucketName(bucket) {
|
|
return ObjectInfo{}, probe.NewError(BucketNameInvalid{Bucket: bucket})
|
|
}
|
|
|
|
bucket = fs.denormalizeBucket(bucket)
|
|
bucketPath := filepath.Join(fs.path, bucket)
|
|
if _, e = os.Stat(bucketPath); e != nil {
|
|
if os.IsNotExist(e) {
|
|
return ObjectInfo{}, probe.NewError(BucketNotFound{Bucket: bucket})
|
|
}
|
|
return ObjectInfo{}, probe.NewError(e)
|
|
}
|
|
|
|
// Verify object path legal.
|
|
if !IsValidObjectName(object) {
|
|
return ObjectInfo{}, probe.NewError(ObjectNameInvalid{Bucket: bucket, Object: object})
|
|
}
|
|
|
|
// Get object path.
|
|
objectPath := filepath.Join(bucketPath, object)
|
|
if expectedMD5Sum != "" {
|
|
var expectedMD5SumBytes []byte
|
|
expectedMD5SumBytes, e = base64.StdEncoding.DecodeString(expectedMD5Sum)
|
|
if e != nil {
|
|
// Pro-actively close the connection.
|
|
return ObjectInfo{}, probe.NewError(InvalidDigest{MD5: expectedMD5Sum})
|
|
}
|
|
expectedMD5Sum = hex.EncodeToString(expectedMD5SumBytes)
|
|
}
|
|
|
|
// Write object.
|
|
file, e := atomic.FileCreateWithPrefix(objectPath, expectedMD5Sum+"$tmpobject")
|
|
if e != nil {
|
|
switch e := e.(type) {
|
|
case *os.PathError:
|
|
if e.Op == "mkdir" {
|
|
if strings.Contains(e.Error(), "not a directory") {
|
|
return ObjectInfo{}, probe.NewError(ObjectExistsAsPrefix{Bucket: bucket, Prefix: object})
|
|
}
|
|
}
|
|
return ObjectInfo{}, probe.NewError(e)
|
|
default:
|
|
return ObjectInfo{}, probe.NewError(e)
|
|
}
|
|
}
|
|
|
|
// Instantiate checksum hashers and create a multiwriter.
|
|
md5Hasher := md5.New()
|
|
sha256Hasher := sha256.New()
|
|
objectWriter := io.MultiWriter(file, md5Hasher, sha256Hasher)
|
|
|
|
if size > 0 {
|
|
if _, e = io.CopyN(objectWriter, data, size); e != nil {
|
|
file.CloseAndPurge()
|
|
return ObjectInfo{}, probe.NewError(e)
|
|
}
|
|
} else {
|
|
if _, e = io.Copy(objectWriter, data); e != nil {
|
|
file.CloseAndPurge()
|
|
return ObjectInfo{}, probe.NewError(e)
|
|
}
|
|
}
|
|
|
|
md5Sum := hex.EncodeToString(md5Hasher.Sum(nil))
|
|
// Verify if the written object is equal to what is expected, only
|
|
// if it is requested as such.
|
|
if expectedMD5Sum != "" {
|
|
if !isMD5SumEqual(expectedMD5Sum, md5Sum) {
|
|
file.CloseAndPurge()
|
|
return ObjectInfo{}, probe.NewError(BadDigest{MD5: expectedMD5Sum, Bucket: bucket, Object: object})
|
|
}
|
|
}
|
|
sha256Sum := hex.EncodeToString(sha256Hasher.Sum(nil))
|
|
if sig != nil {
|
|
ok, err := sig.DoesSignatureMatch(sha256Sum)
|
|
if err != nil {
|
|
file.CloseAndPurge()
|
|
return ObjectInfo{}, err.Trace()
|
|
}
|
|
if !ok {
|
|
file.CloseAndPurge()
|
|
return ObjectInfo{}, probe.NewError(SignDoesNotMatch{})
|
|
}
|
|
}
|
|
file.Close()
|
|
|
|
// Set stat again to get the latest metadata.
|
|
st, e := os.Stat(objectPath)
|
|
if e != nil {
|
|
return ObjectInfo{}, probe.NewError(e)
|
|
}
|
|
contentType := "application/octet-stream"
|
|
if objectExt := filepath.Ext(objectPath); objectExt != "" {
|
|
content, ok := mimedb.DB[strings.ToLower(strings.TrimPrefix(objectExt, "."))]
|
|
if ok {
|
|
contentType = content.ContentType
|
|
}
|
|
}
|
|
newObject := ObjectInfo{
|
|
Bucket: bucket,
|
|
Name: object,
|
|
ModifiedTime: st.ModTime(),
|
|
Size: st.Size(),
|
|
ContentType: contentType,
|
|
MD5Sum: md5Sum,
|
|
}
|
|
return newObject, nil
|
|
}
|
|
|
|
// deleteObjectPath - delete object path if its empty.
|
|
func deleteObjectPath(basePath, deletePath, bucket, object string) *probe.Error {
|
|
if basePath == deletePath {
|
|
return nil
|
|
}
|
|
// Verify if the path exists.
|
|
pathSt, e := os.Stat(deletePath)
|
|
if e != nil {
|
|
if os.IsNotExist(e) {
|
|
return probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
|
|
}
|
|
return probe.NewError(e)
|
|
}
|
|
if pathSt.IsDir() {
|
|
// Verify if directory is empty.
|
|
empty, e := isDirEmpty(deletePath)
|
|
if e != nil {
|
|
return probe.NewError(e)
|
|
}
|
|
if !empty {
|
|
return nil
|
|
}
|
|
}
|
|
// Attempt to remove path.
|
|
if e := os.Remove(deletePath); e != nil {
|
|
return probe.NewError(e)
|
|
}
|
|
// Recursively go down the next path and delete again.
|
|
if err := deleteObjectPath(basePath, filepath.Dir(deletePath), bucket, object); err != nil {
|
|
return err.Trace(basePath, deletePath, bucket, object)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DeleteObject - delete object.
|
|
func (fs Filesystem) DeleteObject(bucket, object string) *probe.Error {
|
|
// Check bucket name valid
|
|
if !IsValidBucketName(bucket) {
|
|
return probe.NewError(BucketNameInvalid{Bucket: bucket})
|
|
}
|
|
|
|
bucket = fs.denormalizeBucket(bucket)
|
|
bucketPath := filepath.Join(fs.path, bucket)
|
|
// Check bucket exists
|
|
if _, e := os.Stat(bucketPath); e != nil {
|
|
if os.IsNotExist(e) {
|
|
return probe.NewError(BucketNotFound{Bucket: bucket})
|
|
}
|
|
return probe.NewError(e)
|
|
}
|
|
|
|
// Verify object path legal
|
|
if !IsValidObjectName(object) {
|
|
return probe.NewError(ObjectNameInvalid{Bucket: bucket, Object: object})
|
|
}
|
|
|
|
// Do not use filepath.Join() since filepath.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().
|
|
var objectPath string
|
|
if runtime.GOOS == "windows" {
|
|
objectPath = fs.path + string(os.PathSeparator) + bucket + string(os.PathSeparator) + object
|
|
} else {
|
|
objectPath = fs.path + string(os.PathSeparator) + bucket + string(os.PathSeparator) + object
|
|
}
|
|
// Delete object path if its empty.
|
|
err := deleteObjectPath(bucketPath, objectPath, bucket, object)
|
|
if err != nil {
|
|
if os.IsNotExist(err.ToGoError()) {
|
|
return probe.NewError(ObjectNotFound{Bucket: bucket, Object: object})
|
|
}
|
|
return err.Trace(bucketPath, objectPath, bucket, object)
|
|
}
|
|
return nil
|
|
}
|