[storage] Adds support for Append Blobs

storage/blob.go

@@ -167,8 +167,9 @@ type BlobType string
 
 // Types of page blobs
 const (
-	BlobTypeBlock BlobType = "BlockBlob"
-	BlobTypePage  BlobType = "PageBlob"
+	BlobTypeBlock  BlobType = "BlockBlob"
+	BlobTypePage   BlobType = "PageBlob"
+	BlobTypeAppend BlobType = "AppendBlob"
 )
 
 // PageWriteType defines the type updates that are going to be
@@ -330,7 +331,6 @@ func (b BlobStorageClient) createContainer(name string, access ContainerAccessTy
 	uri := b.client.getEndpoint(blobServiceName, pathForContainer(name), url.Values{"restype": {"container"}})
 
 	headers := b.client.getStandardHeaders()
-	headers["Content-Length"] = "0"
 	if access != "" {
 		headers["x-ms-blob-public-access"] = string(access)
 	}
@@ -556,7 +556,6 @@ func (b BlobStorageClient) SetBlobMetadata(container, name string, metadata map[
 	for k, v := range metadata {
 		headers[userDefinedMetadataHeaderPrefix+k] = v
 	}
-	headers["Content-Length"] = "0"
 
 	resp, err := b.client.exec("PUT", uri, headers, nil)
 	if err != nil {
@@ -731,7 +730,6 @@ func (b BlobStorageClient) PutPageBlob(container, name string, size int64, extra
 	headers := b.client.getStandardHeaders()
 	headers["x-ms-blob-type"] = string(BlobTypePage)
 	headers["x-ms-blob-content-length"] = fmt.Sprintf("%v", size)
-	headers["Content-Length"] = fmt.Sprintf("%v", 0)
 
 	for k, v := range extraHeaders {
 		headers[k] = v
@@ -801,6 +799,48 @@ func (b BlobStorageClient) GetPageRanges(container, name string) (GetPageRangesR
 	return out, err
 }
 
+// PutAppendBlob initializes an empty append blob with specified name. An
+// append blob must be created using this method before appending blocks.
+//
+// See https://msdn.microsoft.com/en-us/library/azure/dd179451.aspx
+func (b BlobStorageClient) PutAppendBlob(container, name string, extraHeaders map[string]string) error {
+	path := fmt.Sprintf("%s/%s", container, name)
+	uri := b.client.getEndpoint(blobServiceName, path, url.Values{})
+	headers := b.client.getStandardHeaders()
+	headers["x-ms-blob-type"] = string(BlobTypeAppend)
+
+	for k, v := range extraHeaders {
+		headers[k] = v
+	}
+
+	resp, err := b.client.exec("PUT", uri, headers, nil)
+	if err != nil {
+		return err
+	}
+	defer resp.body.Close()
+
+	return checkRespCode(resp.statusCode, []int{http.StatusCreated})
+}
+
+// AppendBlock appends a block to an append blob.
+//
+// See https://msdn.microsoft.com/en-us/library/azure/mt427365.aspx
+func (b BlobStorageClient) AppendBlock(container, name string, chunk []byte) error {
+	path := fmt.Sprintf("%s/%s", container, name)
+	uri := b.client.getEndpoint(blobServiceName, path, url.Values{"comp": {"appendblock"}})
+	headers := b.client.getStandardHeaders()
+	headers["x-ms-blob-type"] = string(BlobTypeAppend)
+	headers["Content-Length"] = fmt.Sprintf("%v", len(chunk))
+
+	resp, err := b.client.exec("PUT", uri, headers, bytes.NewReader(chunk))
+	if err != nil {
+		return err
+	}
+	defer resp.body.Close()
+
+	return checkRespCode(resp.statusCode, []int{http.StatusCreated})
+}
+
 // CopyBlob starts a blob copy operation and waits for the operation to
 // complete. sourceBlob parameter must be a canonical URL to the blob (can be
 // obtained using GetBlobURL method.) There is no SLA on blob copy and therefore
@@ -820,7 +860,6 @@ func (b BlobStorageClient) startBlobCopy(container, name, sourceBlob string) (st
 	uri := b.client.getEndpoint(blobServiceName, pathForBlob(container, name), url.Values{})
 
 	headers := b.client.getStandardHeaders()
-	headers["Content-Length"] = "0"
 	headers["x-ms-copy-source"] = sourceBlob
 
 	resp, err := b.client.exec("PUT", uri, headers, nil)
@@ -951,6 +990,10 @@ func (b BlobStorageClient) GetBlobSASURI(container, name string, expiry time.Tim
 func blobSASStringToSign(signedVersion, canonicalizedResource, signedExpiry, signedPermissions string) (string, error) {
 	var signedStart, signedIdentifier, rscc, rscd, rsce, rscl, rsct string
 
+	if signedVersion >= "2015-02-21" {
+		canonicalizedResource = "/blob" + canonicalizedResource
+	}
+
 	// reference: http://msdn.microsoft.com/en-us/library/azure/dn140255.aspx
 	if signedVersion >= "2013-08-15" {
 		return fmt.Sprintf("%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s", signedPermissions, signedStart, signedExpiry, canonicalizedResource, signedIdentifier, signedVersion, rscc, rscd, rsce, rscl, rsct), nil

storage/blob_test.go

@@ -653,6 +653,59 @@ func (s *StorageBlobSuite) TestGetPageRanges(c *chk.C) {
 	c.Assert(len(out.PageList), chk.Equals, 2)
 }
 
+func (s *StorageBlobSuite) TestPutAppendBlob(c *chk.C) {
+	cli := getBlobClient(c)
+	cnt := randContainer()
+	c.Assert(cli.CreateContainer(cnt, ContainerAccessTypePrivate), chk.IsNil)
+	defer cli.deleteContainer(cnt)
+
+	blob := randString(20)
+	c.Assert(cli.PutAppendBlob(cnt, blob, nil), chk.IsNil)
+
+	// Verify
+	props, err := cli.GetBlobProperties(cnt, blob)
+	c.Assert(err, chk.IsNil)
+	c.Assert(props.ContentLength, chk.Equals, int64(0))
+	c.Assert(props.BlobType, chk.Equals, BlobTypeAppend)
+}
+
+func (s *StorageBlobSuite) TestPutAppendBlobAppendBlocks(c *chk.C) {
+	cli := getBlobClient(c)
+	cnt := randContainer()
+	c.Assert(cli.CreateContainer(cnt, ContainerAccessTypePrivate), chk.IsNil)
+	defer cli.deleteContainer(cnt)
+
+	blob := randString(20)
+	c.Assert(cli.PutAppendBlob(cnt, blob, nil), chk.IsNil)
+
+	chunk1 := []byte(randString(1024))
+	chunk2 := []byte(randString(512))
+
+	// Append first block
+	c.Assert(cli.AppendBlock(cnt, blob, chunk1), chk.IsNil)
+
+	// Verify contents
+	out, err := cli.GetBlobRange(cnt, blob, fmt.Sprintf("%v-%v", 0, len(chunk1)-1))
+	c.Assert(err, chk.IsNil)
+	defer out.Close()
+	blobContents, err := ioutil.ReadAll(out)
+	c.Assert(err, chk.IsNil)
+	c.Assert(blobContents, chk.DeepEquals, chunk1)
+	out.Close()
+
+	// Append second block
+	c.Assert(cli.AppendBlock(cnt, blob, chunk2), chk.IsNil)
+
+	// Verify contents
+	out, err = cli.GetBlobRange(cnt, blob, fmt.Sprintf("%v-%v", 0, len(chunk1)+len(chunk2)-1))
+	c.Assert(err, chk.IsNil)
+	defer out.Close()
+	blobContents, err = ioutil.ReadAll(out)
+	c.Assert(err, chk.IsNil)
+	c.Assert(blobContents, chk.DeepEquals, append(chunk1, chunk2...))
+	out.Close()
+}
+
 func deleteTestContainers(cli BlobStorageClient) error {
 	for {
 		resp, err := cli.ListContainers(ListContainersParameters{Prefix: testContainerPrefix})

storage/client.go

@@ -24,7 +24,7 @@ const (
 
 	// DefaultAPIVersion is the  Azure Storage API version string used when a
 	// basic client is created.
-	DefaultAPIVersion = "2014-02-14"
+	DefaultAPIVersion = "2015-02-21"
 
 	defaultUseHTTPS = true
 
@@ -266,11 +266,15 @@ func (c Client) buildCanonicalizedResource(uri string) (string, error) {
 }
 
 func (c Client) buildCanonicalizedString(verb string, headers map[string]string, canonicalizedResource string) string {
+	contentLength := headers["Content-Length"]
+	if contentLength == "0" {
+		contentLength = ""
+	}
 	canonicalizedString := fmt.Sprintf("%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s\n%s",
 		verb,
 		headers["Content-Encoding"],
 		headers["Content-Language"],
-		headers["Content-Length"],
+		contentLength,
 		headers["Content-MD5"],
 		headers["Content-Type"],
 		headers["Date"],

storage/file.go

@@ -49,7 +49,6 @@ func (f FileServiceClient) CreateShareIfNotExists(name string) (bool, error) {
 func (f FileServiceClient) createShare(name string) (*storageResponse, error) {
 	uri := f.client.getEndpoint(fileServiceName, pathForFileShare(name), url.Values{"restype": {"share"}})
 	headers := f.client.getStandardHeaders()
-	headers["Content-Length"] = "0"
 	return f.client.exec("PUT", uri, headers, nil)
 }
 

storage/queue.go

@@ -134,7 +134,6 @@ type QueueMetadataResponse struct {
 func (c QueueServiceClient) SetMetadata(name string, metadata map[string]string) error {
 	uri := c.client.getEndpoint(queueServiceName, pathForQueue(name), url.Values{"comp": []string{"metadata"}})
 	headers := c.client.getStandardHeaders()
-	headers["Content-Length"] = "0"
 	for k, v := range metadata {
 		headers[userDefinedMetadataHeaderPrefix+k] = v
 	}
@@ -195,7 +194,6 @@ func (c QueueServiceClient) GetMetadata(name string) (QueueMetadataResponse, err
 func (c QueueServiceClient) CreateQueue(name string) error {
 	uri := c.client.getEndpoint(queueServiceName, pathForQueue(name), url.Values{})
 	headers := c.client.getStandardHeaders()
-	headers["Content-Length"] = "0"
 	resp, err := c.client.exec("PUT", uri, headers, nil)
 	if err != nil {
 		return err