Merge pull request #690 from harshavardhana/pr_out_put_object_on_successful_write_returns_full_metadata_to_avoid_subsequent_getobjectmetadata_calls_in_driver

pkg/storage/donut/bucket.go

@@ -84,28 +84,6 @@ func newBucket(bucketName, aclType, donutName string, nodes map[string]node) (bu
 func (b bucket) getBucketName() string {
 	return b.name
 }
-
-func (b bucket) GetObjectMetadata(objectName string) (ObjectMetadata, error) {
-	b.lock.RLock()
-	defer b.lock.RUnlock()
-	metadataReaders, err := b.getDiskReaders(normalizeObjectName(objectName), objectMetadataConfig)
-	if err != nil {
-		return ObjectMetadata{}, iodine.New(err, nil)
-	}
-	for _, metadataReader := range metadataReaders {
-		defer metadataReader.Close()
-	}
-	objMetadata := ObjectMetadata{}
-	for _, metadataReader := range metadataReaders {
-		jdec := json.NewDecoder(metadataReader)
-		if err := jdec.Decode(&objMetadata); err != nil {
-			return ObjectMetadata{}, iodine.New(err, nil)
-		}
-		return objMetadata, nil
-	}
-	return ObjectMetadata{}, iodine.New(InvalidArgument{}, nil)
-}
-
 func (b bucket) getBucketMetadataReaders() ([]io.ReadCloser, error) {
 	var readers []io.ReadCloser
 	for _, node := range b.nodes {
@@ -144,6 +122,13 @@ func (b bucket) getBucketMetadata() (*AllBuckets, error) {
 	return nil, iodine.New(InvalidArgument{}, nil)
 }
 
+// GetObjectMetadata - get metadata for an object
+func (b bucket) GetObjectMetadata(objectName string) (ObjectMetadata, error) {
+	b.lock.RLock()
+	defer b.lock.RUnlock()
+	return b.readObjectMetadata(objectName)
+}
+
 // ListObjects - list all objects
 func (b bucket) ListObjects(prefix, marker, delimiter string, maxkeys int) (ListObjects, error) {
 	b.lock.RLock()
@@ -191,13 +176,20 @@ func (b bucket) ListObjects(prefix, marker, delimiter string, maxkeys int) (List
 		}
 		results = AppendU(results, prefix+objectName)
 	}
-	sort.Strings(results)
 	sort.Strings(commonPrefixes)
 
 	listObjects := ListObjects{}
-	listObjects.Objects = results
+	listObjects.Objects = make(map[string]ObjectMetadata)
 	listObjects.CommonPrefixes = commonPrefixes
 	listObjects.IsTruncated = isTruncated
+
+	for _, objectName := range results {
+		objMetadata, err := b.readObjectMetadata(normalizeObjectName(objectName))
+		if err != nil {
+			return ListObjects{}, iodine.New(err, nil)
+		}
+		listObjects.Objects[objectName] = objMetadata
+	}
 	return listObjects, nil
 }
 
@@ -215,40 +207,29 @@ func (b bucket) ReadObject(objectName string) (reader io.ReadCloser, size int64,
 	if _, ok := bucketMetadata.Buckets[b.getBucketName()].BucketObjects[objectName]; !ok {
 		return nil, 0, iodine.New(ObjectNotFound{Object: objectName}, nil)
 	}
-	objMetadata := ObjectMetadata{}
-	metadataReaders, err := b.getDiskReaders(normalizeObjectName(objectName), objectMetadataConfig)
+	objMetadata, err := b.readObjectMetadata(normalizeObjectName(objectName))
 	if err != nil {
 		return nil, 0, iodine.New(err, nil)
 	}
-	for _, metadataReader := range metadataReaders {
-		defer metadataReader.Close()
-	}
-	for _, metadataReader := range metadataReaders {
-		jdec := json.NewDecoder(metadataReader)
-		if err := jdec.Decode(&objMetadata); err != nil {
-			return nil, 0, iodine.New(err, nil)
-		}
-		break
-	}
 	// read and reply back to GetObject() request in a go-routine
 	go b.readEncodedData(normalizeObjectName(objectName), writer, objMetadata)
 	return reader, objMetadata.Size, nil
 }
 
 // WriteObject - write a new object into bucket
-func (b bucket) WriteObject(objectName string, objectData io.Reader, expectedMD5Sum string, metadata map[string]string) (string, error) {
+func (b bucket) WriteObject(objectName string, objectData io.Reader, expectedMD5Sum string, metadata map[string]string) (ObjectMetadata, error) {
 	b.lock.Lock()
 	defer b.lock.Unlock()
 	if objectName == "" || objectData == nil {
-		return "", iodine.New(InvalidArgument{}, nil)
+		return ObjectMetadata{}, iodine.New(InvalidArgument{}, nil)
 	}
 	writers, err := b.getDiskWriters(normalizeObjectName(objectName), "data")
 	if err != nil {
-		return "", iodine.New(err, nil)
+		return ObjectMetadata{}, iodine.New(err, nil)
 	}
 	sumMD5 := md5.New()
 	sum512 := sha512.New()
-	objMetadata := new(ObjectMetadata)
+	objMetadata := ObjectMetadata{}
 	objMetadata.Version = objectMetadataVersion
 	objMetadata.Created = time.Now().UTC()
 	// if total writers are only '1' do not compute erasure
@@ -257,19 +238,19 @@ func (b bucket) WriteObject(objectName string, objectData io.Reader, expectedMD5
 		mw := io.MultiWriter(writers[0], sumMD5, sum512)
 		totalLength, err := io.Copy(mw, objectData)
 		if err != nil {
-			return "", iodine.New(err, nil)
+			return ObjectMetadata{}, iodine.New(err, nil)
 		}
 		objMetadata.Size = totalLength
 	case false:
 		// calculate data and parity dictated by total number of writers
 		k, m, err := b.getDataAndParity(len(writers))
 		if err != nil {
-			return "", iodine.New(err, nil)
+			return ObjectMetadata{}, iodine.New(err, nil)
 		}
 		// encoded data with k, m and write
 		chunkCount, totalLength, err := b.writeEncodedData(k, m, writers, objectData, sumMD5, sum512)
 		if err != nil {
-			return "", iodine.New(err, nil)
+			return ObjectMetadata{}, iodine.New(err, nil)
 		}
 		/// donutMetadata section
 		objMetadata.BlockSize = blockSize
@@ -290,20 +271,19 @@ func (b bucket) WriteObject(objectName string, objectData io.Reader, expectedMD5
 	// Verify if the written object is equal to what is expected, only if it is requested as such
 	if strings.TrimSpace(expectedMD5Sum) != "" {
 		if err := b.isMD5SumEqual(strings.TrimSpace(expectedMD5Sum), objMetadata.MD5Sum); err != nil {
-			return "", iodine.New(err, nil)
+			return ObjectMetadata{}, iodine.New(err, nil)
 		}
 	}
-
 	objMetadata.Metadata = metadata
 	// write object specific metadata
 	if err := b.writeObjectMetadata(normalizeObjectName(objectName), objMetadata); err != nil {
-		return "", iodine.New(err, nil)
+		return ObjectMetadata{}, iodine.New(err, nil)
 	}
 	// close all writers, when control flow reaches here
 	for _, writer := range writers {
 		writer.Close()
 	}
-	return objMetadata.MD5Sum, nil
+	return objMetadata, nil
 }
 
 // isMD5SumEqual - returns error if md5sum mismatches, other its `nil`
@@ -326,8 +306,8 @@ func (b bucket) isMD5SumEqual(expectedMD5Sum, actualMD5Sum string) error {
 }
 
 // writeObjectMetadata - write additional object metadata
-func (b bucket) writeObjectMetadata(objectName string, objMetadata *ObjectMetadata) error {
-	if objMetadata == nil {
+func (b bucket) writeObjectMetadata(objectName string, objMetadata ObjectMetadata) error {
+	if objMetadata.Object == "" {
 		return iodine.New(InvalidArgument{}, nil)
 	}
 	objMetadataWriters, err := b.getDiskWriters(objectName, objectMetadataConfig)
@@ -339,13 +319,36 @@ func (b bucket) writeObjectMetadata(objectName string, objMetadata *ObjectMetada
 	}
 	for _, objMetadataWriter := range objMetadataWriters {
 		jenc := json.NewEncoder(objMetadataWriter)
-		if err := jenc.Encode(objMetadata); err != nil {
+		if err := jenc.Encode(&objMetadata); err != nil {
 			return iodine.New(err, nil)
 		}
 	}
 	return nil
 }
 
+// readObjectMetadata - read object metadata
+func (b bucket) readObjectMetadata(objectName string) (ObjectMetadata, error) {
+	objMetadata := ObjectMetadata{}
+	if objectName == "" {
+		return ObjectMetadata{}, iodine.New(InvalidArgument{}, nil)
+	}
+	objMetadataReaders, err := b.getDiskReaders(objectName, objectMetadataConfig)
+	if err != nil {
+		return ObjectMetadata{}, iodine.New(err, nil)
+	}
+	for _, objMetadataReader := range objMetadataReaders {
+		defer objMetadataReader.Close()
+	}
+	for _, objMetadataReader := range objMetadataReaders {
+		jdec := json.NewDecoder(objMetadataReader)
+		if err := jdec.Decode(&objMetadata); err != nil {
+			return ObjectMetadata{}, iodine.New(err, nil)
+		}
+		break
+	}
+	return objMetadata, nil
+}
+
 // TODO - This a temporary normalization of objectNames, need to find a better way
 //
 // normalizedObjectName - all objectNames with "/" get normalized to a simple objectName

pkg/storage/donut/definitions.go

@@ -67,7 +67,7 @@ type BucketMetadata struct {
 
 // ListObjects container for list objects response
 type ListObjects struct {
-	Objects        []string `json:"objects"`
-	CommonPrefixes []string `json:"commonPrefixes"`
-	IsTruncated    bool     `json:"isTruncated"`
+	Objects        map[string]ObjectMetadata `json:"objects"`
+	CommonPrefixes []string                  `json:"commonPrefixes"`
+	IsTruncated    bool                      `json:"isTruncated"`
 }

pkg/storage/donut/donut.go

@@ -175,7 +175,7 @@ func (dt donut) ListObjects(bucket, prefix, marker, delimiter string, maxkeys in
 }
 
 // PutObject - put object
-func (dt donut) PutObject(bucket, object, expectedMD5Sum string, reader io.Reader, metadata map[string]string) (string, error) {
+func (dt donut) PutObject(bucket, object, expectedMD5Sum string, reader io.Reader, metadata map[string]string) (ObjectMetadata, error) {
 	dt.lock.Lock()
 	defer dt.lock.Unlock()
 	errParams := map[string]string{
@@ -183,33 +183,33 @@ func (dt donut) PutObject(bucket, object, expectedMD5Sum string, reader io.Reade
 		"object": object,
 	}
 	if bucket == "" || strings.TrimSpace(bucket) == "" {
-		return "", iodine.New(InvalidArgument{}, errParams)
+		return ObjectMetadata{}, iodine.New(InvalidArgument{}, errParams)
 	}
 	if object == "" || strings.TrimSpace(object) == "" {
-		return "", iodine.New(InvalidArgument{}, errParams)
+		return ObjectMetadata{}, iodine.New(InvalidArgument{}, errParams)
 	}
 	if err := dt.listDonutBuckets(); err != nil {
-		return "", iodine.New(err, errParams)
+		return ObjectMetadata{}, iodine.New(err, errParams)
 	}
 	if _, ok := dt.buckets[bucket]; !ok {
-		return "", iodine.New(BucketNotFound{Bucket: bucket}, nil)
+		return ObjectMetadata{}, iodine.New(BucketNotFound{Bucket: bucket}, nil)
 	}
 	bucketMeta, err := dt.getDonutBucketMetadata()
 	if err != nil {
-		return "", iodine.New(err, errParams)
+		return ObjectMetadata{}, iodine.New(err, errParams)
 	}
 	if _, ok := bucketMeta.Buckets[bucket].BucketObjects[object]; ok {
-		return "", iodine.New(ObjectExists{Object: object}, errParams)
+		return ObjectMetadata{}, iodine.New(ObjectExists{Object: object}, errParams)
 	}
-	md5sum, err := dt.buckets[bucket].WriteObject(object, reader, expectedMD5Sum, metadata)
+	objMetadata, err := dt.buckets[bucket].WriteObject(object, reader, expectedMD5Sum, metadata)
 	if err != nil {
-		return "", iodine.New(err, errParams)
+		return ObjectMetadata{}, iodine.New(err, errParams)
 	}
 	bucketMeta.Buckets[bucket].BucketObjects[object] = 1
 	if err := dt.setDonutBucketMetadata(bucketMeta); err != nil {
-		return "", iodine.New(err, errParams)
+		return ObjectMetadata{}, iodine.New(err, errParams)
 	}
-	return md5sum, nil
+	return objMetadata, nil
 }
 
 // GetObject - get object

pkg/storage/donut/donut_test.go

@@ -93,7 +93,7 @@ func (s *MySuite) TestEmptyBucket(c *C) {
 	// check if bucket is empty
 	listObjects, err := donut.ListObjects("foo", "", "", "", 1)
 	c.Assert(err, IsNil)
-	c.Assert(listObjects.Objects, IsNil)
+	c.Assert(len(listObjects.Objects), Equals, 0)
 	c.Assert(listObjects.CommonPrefixes, IsNil)
 	c.Assert(listObjects.IsTruncated, Equals, false)
 }
@@ -195,13 +195,9 @@ func (s *MySuite) TestNewObjectMetadata(c *C) {
 	err = donut.MakeBucket("foo", "private")
 	c.Assert(err, IsNil)
 
-	calculatedMd5Sum, err := donut.PutObject("foo", "obj", expectedMd5Sum, reader, metadata)
+	objectMetadata, err := donut.PutObject("foo", "obj", expectedMd5Sum, reader, metadata)
 	c.Assert(err, IsNil)
-	c.Assert(calculatedMd5Sum, Equals, expectedMd5Sum)
-
-	objectMetadata, err := donut.GetObjectMetadata("foo", "obj")
-	c.Assert(err, IsNil)
-
+	c.Assert(objectMetadata.MD5Sum, Equals, expectedMd5Sum)
 	c.Assert(objectMetadata.Metadata["contentType"], Equals, metadata["contentType"])
 	c.Assert(objectMetadata.Metadata["foo"], Equals, metadata["foo"])
 	c.Assert(objectMetadata.Metadata["hello"], Equals, metadata["hello"])
@@ -240,9 +236,9 @@ func (s *MySuite) TestNewObjectCanBeWritten(c *C) {
 	reader := ioutil.NopCloser(bytes.NewReader([]byte(data)))
 	metadata["contentLength"] = strconv.Itoa(len(data))
 
-	calculatedMd5Sum, err := donut.PutObject("foo", "obj", expectedMd5Sum, reader, metadata)
+	actualMetadata, err := donut.PutObject("foo", "obj", expectedMd5Sum, reader, metadata)
 	c.Assert(err, IsNil)
-	c.Assert(calculatedMd5Sum, Equals, expectedMd5Sum)
+	c.Assert(actualMetadata.MD5Sum, Equals, expectedMd5Sum)
 
 	reader, size, err := donut.GetObject("foo", "obj")
 	c.Assert(err, IsNil)
@@ -253,7 +249,7 @@ func (s *MySuite) TestNewObjectCanBeWritten(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(actualData.Bytes(), DeepEquals, []byte(data))
 
-	actualMetadata, err := donut.GetObjectMetadata("foo", "obj")
+	actualMetadata, err = donut.GetObjectMetadata("foo", "obj")
 	c.Assert(err, IsNil)
 	c.Assert(expectedMd5Sum, Equals, actualMetadata.MD5Sum)
 	c.Assert(int64(len(data)), Equals, actualMetadata.Size)
@@ -312,7 +308,8 @@ func (s *MySuite) TestMultipleNewObjects(c *C) {
 	// test list objects with only delimiter
 	listObjects, err = donut.ListObjects("foo", "", "", "1", 10)
 	c.Assert(err, IsNil)
-	c.Assert(listObjects.Objects[0], Equals, "obj2")
+	_, ok := listObjects.Objects["obj2"]
+	c.Assert(ok, Equals, true)
 	c.Assert(listObjects.IsTruncated, Equals, false)
 	c.Assert(listObjects.CommonPrefixes[0], Equals, "obj1")
 
@@ -320,7 +317,10 @@ func (s *MySuite) TestMultipleNewObjects(c *C) {
 	listObjects, err = donut.ListObjects("foo", "o", "", "", 10)
 	c.Assert(err, IsNil)
 	c.Assert(listObjects.IsTruncated, Equals, false)
-	c.Assert(listObjects.Objects, DeepEquals, []string{"obj1", "obj2"})
+	_, ok1 := listObjects.Objects["obj1"]
+	_, ok2 := listObjects.Objects["obj2"]
+	c.Assert(ok1, Equals, true)
+	c.Assert(ok2, Equals, true)
 
 	three := ioutil.NopCloser(bytes.NewReader([]byte("three")))
 	metadata["contentLength"] = strconv.Itoa(len("three"))

pkg/storage/donut/interfaces.go

@@ -40,7 +40,7 @@ type ObjectStorage interface {
 	// Object operations
 	GetObject(bucket, object string) (io.ReadCloser, int64, error)
 	GetObjectMetadata(bucket, object string) (ObjectMetadata, error)
-	PutObject(bucket, object, expectedMD5Sum string, reader io.Reader, metadata map[string]string) (string, error)
+	PutObject(bucket, object, expectedMD5Sum string, reader io.Reader, metadata map[string]string) (ObjectMetadata, error)
 }
 
 // Management is a donut management system interface

pkg/storage/drivers/donut/donut.go

@@ -448,6 +448,12 @@ func (d donutDriver) GetObjectMetadata(bucketName, objectName string) (drivers.O
 	return objectMetadata, nil
 }
 
+type byObjectName []drivers.ObjectMetadata
+
+func (b byObjectName) Len() int           { return len(b) }
+func (b byObjectName) Swap(i, j int)      { b[i], b[j] = b[j], b[i] }
+func (b byObjectName) Less(i, j int) bool { return b[i].Key < b[j].Key }
+
 // ListObjects - returns list of objects
 func (d donutDriver) ListObjects(bucketName string, resources drivers.BucketResourcesMetadata) ([]drivers.ObjectMetadata, drivers.BucketResourcesMetadata, error) {
 	d.lock.RLock()
@@ -470,23 +476,18 @@ func (d donutDriver) ListObjects(bucketName string, resources drivers.BucketReso
 	}
 	resources.CommonPrefixes = listObjects.CommonPrefixes
 	resources.IsTruncated = listObjects.IsTruncated
-	if resources.IsTruncated && resources.IsDelimiterSet() {
-		resources.NextMarker = listObjects.Objects[len(listObjects.Objects)-1]
-	}
-	// make sure to keep the lexical order same as returned by donut
-	// we do not have to sort here again
-	results := make([]drivers.ObjectMetadata, len(listObjects.Objects))
-	for i, objectName := range listObjects.Objects {
-		objectMetadata, err := d.donut.GetObjectMetadata(bucketName, objectName)
-		if err != nil {
-			return nil, drivers.BucketResourcesMetadata{}, iodine.New(err, errParams)
-		}
+	var results []drivers.ObjectMetadata
+	for _, objMetadata := range listObjects.Objects {
 		metadata := drivers.ObjectMetadata{
-			Key:     objectMetadata.Object,
-			Created: objectMetadata.Created,
-			Size:    objectMetadata.Size,
+			Key:     objMetadata.Object,
+			Created: objMetadata.Created,
+			Size:    objMetadata.Size,
 		}
-		results[i] = metadata
+		results = append(results, metadata)
+	}
+	sort.Sort(byObjectName(results))
+	if resources.IsTruncated && resources.IsDelimiterSet() {
+		resources.NextMarker = results[len(results)-1].Key
 	}
 	return results, resources, nil
 }
@@ -527,6 +528,10 @@ func (d donutDriver) CreateObject(bucketName, objectName, contentType, expectedM
 		"objectName":  objectName,
 		"contentType": contentType,
 	}
+	if d.donut == nil {
+		return "", iodine.New(drivers.InternalError{}, errParams)
+	}
+	// TODO - Should be able to write bigger than cache
 	if size > int64(d.maxSize) {
 		generic := drivers.GenericObjectError{Bucket: bucketName, Object: objectName}
 		return "", iodine.New(drivers.EntityTooLarge{
@@ -535,9 +540,6 @@ func (d donutDriver) CreateObject(bucketName, objectName, contentType, expectedM
 			MaxSize:            strconv.FormatUint(d.maxSize, 10),
 		}, nil)
 	}
-	if d.donut == nil {
-		return "", iodine.New(drivers.InternalError{}, errParams)
-	}
 	if !drivers.IsValidBucket(bucketName) {
 		return "", iodine.New(drivers.BucketNameInvalid{Bucket: bucketName}, nil)
 	}
@@ -564,7 +566,7 @@ func (d donutDriver) CreateObject(bucketName, objectName, contentType, expectedM
 		expectedMD5Sum = hex.EncodeToString(expectedMD5SumBytes)
 	}
 	newReader := newProxyReader(reader)
-	calculatedMD5Sum, err := d.donut.PutObject(bucketName, objectName, expectedMD5Sum, newReader, metadata)
+	objMetadata, err := d.donut.PutObject(bucketName, objectName, expectedMD5Sum, newReader, metadata)
 	if err != nil {
 		switch iodine.ToError(err).(type) {
 		case donut.BadDigest:
@@ -576,20 +578,16 @@ func (d donutDriver) CreateObject(bucketName, objectName, contentType, expectedM
 	// free up
 	newReader.readBytes = nil
 	go debug.FreeOSMemory()
-	objectMetadata, err := d.donut.GetObjectMetadata(bucketName, objectName)
-	if err != nil {
-		return "", iodine.New(err, nil)
-	}
 	newObject := drivers.ObjectMetadata{
 		Bucket: bucketName,
 		Key:    objectName,
 
-		ContentType: objectMetadata.Metadata["contentType"],
-		Created:     objectMetadata.Created,
-		Md5:         calculatedMD5Sum,
-		Size:        objectMetadata.Size,
+		ContentType: objMetadata.Metadata["contentType"],
+		Created:     objMetadata.Created,
+		Md5:         objMetadata.MD5Sum,
+		Size:        objMetadata.Size,
 	}
 	storedBucket.objectMetadata[objectKey] = newObject
 	d.storedBuckets[bucketName] = storedBucket
-	return calculatedMD5Sum, nil
+	return newObject.Md5, nil
 }