xl/fs: Object layer - keep common functions into single place. (#1423)

fs-objects-multipart.go

@@ -302,21 +302,6 @@ func (fs fsObjects) NewMultipartUpload(bucket, object string) (string, error) {
 	}
 }
 
-// isUploadIDExists - verify if a given uploadID exists and is valid.
-func isUploadIDExists(storage StorageAPI, bucket, object, uploadID string) (bool, error) {
-	uploadIDPath := path.Join(bucket, object, uploadID)
-	st, err := storage.StatFile(minioMetaVolume, uploadIDPath)
-	if err != nil {
-		// Upload id does not exist.
-		if err == errFileNotFound {
-			return false, nil
-		}
-		return false, err
-	}
-	// Upload id exists and is a regular file.
-	return st.Mode.IsRegular(), nil
-}
-
 // PutObjectPart - writes the multipart upload chunks.
 func (fs fsObjects) PutObjectPart(bucket, object, uploadID string, partID int, size int64, data io.Reader, md5Hex string) (string, error) {
 	// Verify if bucket is valid.

fs-objects.go

@@ -21,7 +21,6 @@ import (
 	"encoding/hex"
 	"io"
 	"path/filepath"
-	"sort"
 	"strings"
 
 	"github.com/minio/minio/pkg/mimedb"
@@ -52,93 +51,26 @@ func newFSObjects(exportPath string) (ObjectLayer, error) {
 	return fsObjects{storage}, nil
 }
 
-// checks whether bucket exists.
-func isBucketExist(storage StorageAPI, bucketName string) (bool, error) {
-	// Check whether bucket exists.
-	if _, e := storage.StatVol(bucketName); e != nil {
-		if e == errVolumeNotFound {
-			return false, nil
-		}
-		return false, e
-	}
-	return true, nil
-}
-
 /// Bucket operations
 
 // MakeBucket - make a bucket.
 func (fs fsObjects) MakeBucket(bucket string) error {
-	// Verify if bucket is valid.
-	if !IsValidBucketName(bucket) {
-		return BucketNameInvalid{Bucket: bucket}
-	}
-	if err := fs.storage.MakeVol(bucket); err != nil {
-		return toObjectErr(err, bucket)
-	}
-	// This happens for the first time, but keep this here since this
-	// is the only place where it can be made expensive optimizing all
-	// other calls.
-	// Create minio meta volume, if it doesn't exist yet.
-	if err := fs.storage.MakeVol(minioMetaVolume); err != nil {
-		if err != errVolumeExists {
-			return toObjectErr(err, minioMetaVolume)
-		}
-	}
-	return nil
+	return makeBucket(fs.storage, bucket)
 }
 
 // GetBucketInfo - get bucket info.
 func (fs fsObjects) GetBucketInfo(bucket string) (BucketInfo, error) {
-	// Verify if bucket is valid.
-	if !IsValidBucketName(bucket) {
-		return BucketInfo{}, BucketNameInvalid{Bucket: bucket}
-	}
-	vi, err := fs.storage.StatVol(bucket)
-	if err != nil {
-		return BucketInfo{}, toObjectErr(err, bucket)
-	}
-	return BucketInfo{
-		Name:    bucket,
-		Created: vi.Created,
-		Total:   vi.Total,
-		Free:    vi.Free,
-	}, nil
+	return getBucketInfo(fs.storage, bucket)
 }
 
 // ListBuckets - list buckets.
 func (fs fsObjects) ListBuckets() ([]BucketInfo, error) {
-	var bucketInfos []BucketInfo
-	vols, err := fs.storage.ListVols()
-	if err != nil {
-		return nil, toObjectErr(err)
-	}
-	for _, vol := range vols {
-		// StorageAPI can send volume names which are incompatible
-		// with buckets, handle it and skip them.
-		if !IsValidBucketName(vol.Name) {
-			continue
-		}
-		bucketInfos = append(bucketInfos, BucketInfo{
-			Name:    vol.Name,
-			Created: vol.Created,
-			Total:   vol.Total,
-			Free:    vol.Free,
-		})
-	}
-	sort.Sort(byBucketName(bucketInfos))
-	return bucketInfos, nil
+	return listBuckets(fs.storage)
 }
 
 // DeleteBucket - delete a bucket.
 func (fs fsObjects) DeleteBucket(bucket string) error {
-	// Verify if bucket is valid.
-	if !IsValidBucketName(bucket) {
-		return BucketNameInvalid{Bucket: bucket}
-	}
-	if err := fs.storage.DeleteVol(bucket); err != nil {
-		return toObjectErr(err)
-	}
-	return nil
+	return deleteBucket(fs.storage, bucket)
 }
 
 /// Object Operations

object-common.go

@@ -0,0 +1,126 @@
+/*
+ * Minio Cloud Storage, (C) 2016 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 main
+
+import (
+	"path"
+	"sort"
+)
+
+/// Common object layer functions.
+
+// makeBucket - create a bucket, is a common function for both object layers.
+func makeBucket(storage StorageAPI, bucket string) error {
+	// Verify if bucket is valid.
+	if !IsValidBucketName(bucket) {
+		return BucketNameInvalid{Bucket: bucket}
+	}
+	if err := storage.MakeVol(bucket); err != nil {
+		return toObjectErr(err, bucket)
+	}
+	// This happens for the first time, but keep this here since this
+	// is the only place where it can be made expensive optimizing all
+	// other calls.
+	// Create minio meta volume, if it doesn't exist yet.
+	if err := storage.MakeVol(minioMetaVolume); err != nil {
+		if err != errVolumeExists {
+			return toObjectErr(err, minioMetaVolume)
+		}
+	}
+	return nil
+}
+
+// getBucketInfo - fetch bucket info, is a common function for both object layers.
+func getBucketInfo(storage StorageAPI, bucket string) (BucketInfo, error) {
+	// Verify if bucket is valid.
+	if !IsValidBucketName(bucket) {
+		return BucketInfo{}, BucketNameInvalid{Bucket: bucket}
+	}
+	vi, err := storage.StatVol(bucket)
+	if err != nil {
+		return BucketInfo{}, toObjectErr(err, bucket)
+	}
+	return BucketInfo{
+		Name:    bucket,
+		Created: vi.Created,
+		Total:   vi.Total,
+		Free:    vi.Free,
+	}, nil
+}
+
+// listBuckets - list all buckets, is a common function for both object layers.
+func listBuckets(storage StorageAPI) ([]BucketInfo, error) {
+	var bucketInfos []BucketInfo
+	vols, err := storage.ListVols()
+	if err != nil {
+		return nil, toObjectErr(err)
+	}
+	for _, vol := range vols {
+		// StorageAPI can send volume names which are incompatible
+		// with buckets, handle it and skip them.
+		if !IsValidBucketName(vol.Name) {
+			continue
+		}
+		bucketInfos = append(bucketInfos, BucketInfo{
+			Name:    vol.Name,
+			Created: vol.Created,
+			Total:   vol.Total,
+			Free:    vol.Free,
+		})
+	}
+	sort.Sort(byBucketName(bucketInfos))
+	return bucketInfos, nil
+}
+
+// deleteBucket - deletes a bucket, is a common function for both the layers.
+func deleteBucket(storage StorageAPI, bucket string) error {
+	// Verify if bucket is valid.
+	if !IsValidBucketName(bucket) {
+		return BucketNameInvalid{Bucket: bucket}
+	}
+	if err := storage.DeleteVol(bucket); err != nil {
+		return toObjectErr(err)
+	}
+	return nil
+}
+
+// isUploadIDExists - verify if a given uploadID exists and is valid.
+func isUploadIDExists(storage StorageAPI, bucket, object, uploadID string) (bool, error) {
+	uploadIDPath := path.Join(bucket, object, uploadID)
+	st, err := storage.StatFile(minioMetaVolume, uploadIDPath)
+	if err != nil {
+		// Upload id does not exist.
+		if err == errFileNotFound {
+			return false, nil
+		}
+		return false, err
+	}
+	// Upload id exists and is a regular file.
+	return st.Mode.IsRegular(), nil
+}
+
+// checks whether bucket exists.
+func isBucketExist(storage StorageAPI, bucketName string) (bool, error) {
+	// Check whether bucket exists.
+	if _, e := storage.StatVol(bucketName); e != nil {
+		if e == errVolumeNotFound {
+			return false, nil
+		}
+		return false, e
+	}
+	return true, nil
+}

xl-objects-multipart.go

@@ -260,7 +260,7 @@ func (xl xlObjects) NewMultipartUpload(bucket, object string) (string, error) {
 		return "", ObjectNameInvalid{Bucket: bucket, Object: object}
 	}
 	// Verify whether the bucket exists.
-	if isExist, err := xl.isBucketExist(bucket); err != nil {
+	if isExist, err := isBucketExist(xl.storage, bucket); err != nil {
 		return "", err
 	} else if !isExist {
 		return "", BucketNotFound{Bucket: bucket}
@@ -302,21 +302,6 @@ func (xl xlObjects) NewMultipartUpload(bucket, object string) (string, error) {
 	}
 }
 
-// isUploadIDExists - verify if a given uploadID exists and is valid.
-func (xl xlObjects) isUploadIDExists(bucket, object, uploadID string) (bool, error) {
-	uploadIDPath := path.Join(bucket, object, uploadID)
-	st, err := xl.storage.StatFile(minioMetaVolume, uploadIDPath)
-	if err != nil {
-		// Upload id does not exist.
-		if err == errFileNotFound {
-			return false, nil
-		}
-		return false, err
-	}
-	// Upload id exists and is a regular file.
-	return st.Mode.IsRegular(), nil
-}
-
 // PutObjectPart - writes the multipart upload chunks.
 func (xl xlObjects) PutObjectPart(bucket, object, uploadID string, partID int, size int64, data io.Reader, md5Hex string) (string, error) {
 	// Verify if bucket is valid.
@@ -327,13 +312,13 @@ func (xl xlObjects) PutObjectPart(bucket, object, uploadID string, partID int, s
 		return "", ObjectNameInvalid{Bucket: bucket, Object: object}
 	}
 	// Verify whether the bucket exists.
-	if isExist, err := xl.isBucketExist(bucket); err != nil {
+	if isExist, err := isBucketExist(xl.storage, bucket); err != nil {
 		return "", err
 	} else if !isExist {
 		return "", BucketNotFound{Bucket: bucket}
 	}
 
-	if status, err := xl.isUploadIDExists(bucket, object, uploadID); err != nil {
+	if status, err := isUploadIDExists(xl.storage, bucket, object, uploadID); err != nil {
 		return "", err
 	} else if !status {
 		return "", InvalidUploadID{UploadID: uploadID}
@@ -393,7 +378,7 @@ func (xl xlObjects) ListObjectParts(bucket, object, uploadID string, partNumberM
 	if !IsValidObjectName(object) {
 		return ListPartsInfo{}, (ObjectNameInvalid{Bucket: bucket, Object: object})
 	}
-	if status, err := xl.isUploadIDExists(bucket, object, uploadID); err != nil {
+	if status, err := isUploadIDExists(xl.storage, bucket, object, uploadID); err != nil {
 		return ListPartsInfo{}, err
 	} else if !status {
 		return ListPartsInfo{}, (InvalidUploadID{UploadID: uploadID})
@@ -457,33 +442,24 @@ func (xl xlObjects) CompleteMultipartUpload(bucket string, object string, upload
 			Object: object,
 		})
 	}
-	if status, err := xl.isUploadIDExists(bucket, object, uploadID); err != nil {
+	if status, err := isUploadIDExists(xl.storage, bucket, object, uploadID); err != nil {
 		return "", err
 	} else if !status {
 		return "", (InvalidUploadID{UploadID: uploadID})
 	}
 
-	fileWriter, err := xl.storage.CreateFile(bucket, object)
-	if err != nil {
-		return "", toObjectErr(err, bucket, object)
-	}
-
 	var md5Sums []string
 	for _, part := range parts {
 		// Construct part suffix.
 		partSuffix := fmt.Sprintf("%s.%d.%s", uploadID, part.PartNumber, part.ETag)
-		err = xl.storage.RenameFile(minioMetaVolume, path.Join(bucket, object, partSuffix), bucket, path.Join(object, fmt.Sprint(part.PartNumber)))
+		err := xl.storage.RenameFile(minioMetaVolume, path.Join(bucket, object, partSuffix), bucket, path.Join(object, fmt.Sprint(part.PartNumber)))
+		// We need a way to roll back if of the renames failed.
 		if err != nil {
 			return "", err
 		}
 		md5Sums = append(md5Sums, part.ETag)
 	}
 
-	err = fileWriter.Close()
-	if err != nil {
-		return "", err
-	}
-
 	// Save the s3 md5.
 	s3MD5, err := makeS3MD5(md5Sums...)
 	if err != nil {
@@ -503,7 +479,7 @@ func (xl xlObjects) AbortMultipartUpload(bucket, object, uploadID string) error
 	if !IsValidObjectName(object) {
 		return (ObjectNameInvalid{Bucket: bucket, Object: object})
 	}
-	if status, err := xl.isUploadIDExists(bucket, object, uploadID); err != nil {
+	if status, err := isUploadIDExists(xl.storage, bucket, object, uploadID); err != nil {
 		return err
 	} else if !status {
 		return (InvalidUploadID{UploadID: uploadID})

xl-objects.go

@@ -23,7 +23,6 @@ import (
 	"fmt"
 	"io"
 	"path/filepath"
-	"sort"
 	"strings"
 
 	"github.com/minio/minio/pkg/mimedb"
@@ -47,93 +46,26 @@ func newXLObjects(exportPaths ...string) (ObjectLayer, error) {
 	return xlObjects{storage}, nil
 }
 
-// checks whether bucket exists.
-func (xl xlObjects) isBucketExist(bucketName string) (bool, error) {
-	// Check whether bucket exists.
-	if _, e := xl.storage.StatVol(bucketName); e != nil {
-		if e == errVolumeNotFound {
-			return false, nil
-		}
-		return false, e
-	}
-	return true, nil
-}
-
 /// Bucket operations
 
 // MakeBucket - make a bucket.
 func (xl xlObjects) MakeBucket(bucket string) error {
-	// Verify if bucket is valid.
-	if !IsValidBucketName(bucket) {
-		return BucketNameInvalid{Bucket: bucket}
-	}
-	if err := xl.storage.MakeVol(bucket); err != nil {
-		return toObjectErr(err, bucket)
-	}
-	// This happens for the first time, but keep this here since this
-	// is the only place where it can be made expensive optimizing all
-	// other calls.
-	// Create minio meta volume, if it doesn't exist yet.
-	if err := xl.storage.MakeVol(minioMetaVolume); err != nil {
-		if err != errVolumeExists {
-			return toObjectErr(err, minioMetaVolume)
-		}
-	}
-	return nil
+	return makeBucket(xl.storage, bucket)
 }
 
 // GetBucketInfo - get bucket info.
 func (xl xlObjects) GetBucketInfo(bucket string) (BucketInfo, error) {
-	// Verify if bucket is valid.
-	if !IsValidBucketName(bucket) {
-		return BucketInfo{}, BucketNameInvalid{Bucket: bucket}
-	}
-	vi, err := xl.storage.StatVol(bucket)
-	if err != nil {
-		return BucketInfo{}, toObjectErr(err, bucket)
-	}
-	return BucketInfo{
-		Name:    bucket,
-		Created: vi.Created,
-		Total:   vi.Total,
-		Free:    vi.Free,
-	}, nil
+	return getBucketInfo(xl.storage, bucket)
 }
 
 // ListBuckets - list buckets.
 func (xl xlObjects) ListBuckets() ([]BucketInfo, error) {
-	var bucketInfos []BucketInfo
-	vols, err := xl.storage.ListVols()
-	if err != nil {
-		return nil, toObjectErr(err)
-	}
-	for _, vol := range vols {
-		// StorageAPI can send volume names which are incompatible
-		// with buckets, handle it and skip them.
-		if !IsValidBucketName(vol.Name) {
-			continue
-		}
-		bucketInfos = append(bucketInfos, BucketInfo{
-			Name:    vol.Name,
-			Created: vol.Created,
-			Total:   vol.Total,
-			Free:    vol.Free,
-		})
-	}
-	sort.Sort(byBucketName(bucketInfos))
-	return bucketInfos, nil
+	return listBuckets(xl.storage)
 }
 
 // DeleteBucket - delete a bucket.
 func (xl xlObjects) DeleteBucket(bucket string) error {
-	// Verify if bucket is valid.
-	if !IsValidBucketName(bucket) {
-		return BucketNameInvalid{Bucket: bucket}
-	}
-	if err := xl.storage.DeleteVol(bucket); err != nil {
-		return toObjectErr(err)
-	}
-	return nil
+	return deleteBucket(xl.storage, bucket)
 }
 
 /// Object Operations
@@ -264,11 +196,9 @@ func (xl xlObjects) PutObject(bucket string, object string, size int64, data io.
 		})
 	}
 	// Check whether the bucket exists.
-	isExist, err := xl.isBucketExist(bucket)
-	if err != nil {
+	if isExist, err := isBucketExist(xl.storage, bucket); err != nil {
 		return "", err
-	}
-	if !isExist {
+	} else if !isExist {
 		return "", BucketNotFound{Bucket: bucket}
 	}