X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/40c9b26a39c773e806e0a1430774f1787820376f..94788532e822ef26b8c9eac7818f03e3a94fb124:/services/keepstore/azure_blob_volume.go

diff --git a/services/keepstore/azure_blob_volume.go b/services/keepstore/azure_blob_volume.go
index d545aad39c..e9fda2ab76 100644
--- a/services/keepstore/azure_blob_volume.go
+++ b/services/keepstore/azure_blob_volume.go
@@ -9,6 +9,7 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
+	"regexp"
 	"strings"
 	"time"
 
@@ -19,8 +20,8 @@ var (
 	azureStorageAccountName    string
 	azureStorageAccountKeyFile string
 	azureStorageReplication    int
-	azureWriteRaceInterval     time.Duration = 15 * time.Second
-	azureWriteRacePollTime     time.Duration = time.Second
+	azureWriteRaceInterval     = 15 * time.Second
+	azureWriteRacePollTime     = time.Second
 )
 
 func readKeyFromFile(file string) (string, error) {
@@ -96,6 +97,9 @@ type AzureBlobVolume struct {
 	replication   int
 }
 
+// NewAzureBlobVolume returns a new AzureBlobVolume using the given
+// client and container name. The replication argument specifies the
+// replication level to report when writing data.
 func NewAzureBlobVolume(client storage.Client, containerName string, readonly bool, replication int) *AzureBlobVolume {
 	return &AzureBlobVolume{
 		azClient:      client,
@@ -118,6 +122,12 @@ func (v *AzureBlobVolume) Check() error {
 	return nil
 }
 
+// Get reads a Keep block that has been stored as a block blob in the
+// container.
+//
+// If the block is younger than azureWriteRaceInterval and is
+// unexpectedly empty, assume a PutBlob operation is in progress, and
+// wait for it to finish writing.
 func (v *AzureBlobVolume) Get(loc string) ([]byte, error) {
 	var deadline time.Time
 	haveDeadline := false
@@ -155,11 +165,7 @@ func (v *AzureBlobVolume) Get(loc string) ([]byte, error) {
 func (v *AzureBlobVolume) get(loc string) ([]byte, error) {
 	rdr, err := v.bsClient.GetBlob(v.containerName, loc)
 	if err != nil {
-		if strings.Contains(err.Error(), "404 Not Found") {
-			// "storage: service returned without a response body (404 Not Found)"
-			return nil, os.ErrNotExist
-		}
-		return nil, err
+		return nil, v.translateError(err)
 	}
 	defer rdr.Close()
 	buf := bufs.Get(BlockSize)
@@ -173,15 +179,17 @@ func (v *AzureBlobVolume) get(loc string) ([]byte, error) {
 	}
 }
 
+// Compare the given data with existing stored data.
 func (v *AzureBlobVolume) Compare(loc string, expect []byte) error {
 	rdr, err := v.bsClient.GetBlob(v.containerName, loc)
 	if err != nil {
-		return err
+		return v.translateError(err)
 	}
 	defer rdr.Close()
 	return compareReaderWithBuf(rdr, expect, loc[:32])
 }
 
+// Put sotres a Keep block as a block blob in the container.
 func (v *AzureBlobVolume) Put(loc string, block []byte) error {
 	if v.readonly {
 		return MethodDisabledError
@@ -189,6 +197,7 @@ func (v *AzureBlobVolume) Put(loc string, block []byte) error {
 	return v.bsClient.CreateBlockBlobFromReader(v.containerName, loc, uint64(len(block)), bytes.NewReader(block))
 }
 
+// Touch updates the last-modified property of a block blob.
 func (v *AzureBlobVolume) Touch(loc string) error {
 	if v.readonly {
 		return MethodDisabledError
@@ -198,6 +207,7 @@ func (v *AzureBlobVolume) Touch(loc string) error {
 	})
 }
 
+// Mtime returns the last-modified property of a block blob.
 func (v *AzureBlobVolume) Mtime(loc string) (time.Time, error) {
 	props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 	if err != nil {
@@ -206,6 +216,8 @@ func (v *AzureBlobVolume) Mtime(loc string) (time.Time, error) {
 	return time.Parse(time.RFC1123, props.LastModified)
 }
 
+// IndexTo writes a list of Keep blocks that are stored in the
+// container.
 func (v *AzureBlobVolume) IndexTo(prefix string, writer io.Writer) error {
 	params := storage.ListBlobsParameters{
 		Prefix: prefix,
@@ -220,6 +232,9 @@ func (v *AzureBlobVolume) IndexTo(prefix string, writer io.Writer) error {
 			if err != nil {
 				return err
 			}
+			if !v.isKeepBlock(b.Name) {
+				continue
+			}
 			if b.Properties.ContentLength == 0 && t.Add(azureWriteRaceInterval).After(time.Now()) {
 				// A new zero-length blob is probably
 				// just a new non-empty blob that
@@ -237,6 +252,7 @@ func (v *AzureBlobVolume) IndexTo(prefix string, writer io.Writer) error {
 	}
 }
 
+// Delete a Keep block.
 func (v *AzureBlobVolume) Delete(loc string) error {
 	if v.readonly {
 		return MethodDisabledError
@@ -260,6 +276,7 @@ func (v *AzureBlobVolume) Delete(loc string) error {
 	})
 }
 
+// Status returns a VolumeStatus struct with placeholder data.
 func (v *AzureBlobVolume) Status() *VolumeStatus {
 	return &VolumeStatus{
 		DeviceNum: 1,
@@ -268,14 +285,38 @@ func (v *AzureBlobVolume) Status() *VolumeStatus {
 	}
 }
 
+// String returns a volume label, including the container name.
 func (v *AzureBlobVolume) String() string {
 	return fmt.Sprintf("azure-storage-container:%+q", v.containerName)
 }
 
+// Writable returns true, unless the -readonly flag was on when the
+// volume was added.
 func (v *AzureBlobVolume) Writable() bool {
 	return !v.readonly
 }
 
+// Replication returns the replication level of the container, as
+// specified by the -azure-storage-replication argument.
 func (v *AzureBlobVolume) Replication() int {
 	return v.replication
 }
+
+// If possible, translate an Azure SDK error to a recognizable error
+// like os.ErrNotExist.
+func (v *AzureBlobVolume) translateError(err error) error {
+	switch {
+	case err == nil:
+		return err
+	case strings.Contains(err.Error(), "404 Not Found"):
+		// "storage: service returned without a response body (404 Not Found)"
+		return os.ErrNotExist
+	default:
+		return err
+	}
+}
+
+var keepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
+func (v *AzureBlobVolume) isKeepBlock(s string) bool {
+	return keepBlockRegexp.MatchString(s)
+}