Merge branch 'master' into 7167-keep-rsync

diff --git a/sdk/go/streamer/streamer_test.go b/sdk/go/streamer/streamer_test.go
index 853d7d30354daddb86e602c4b580141fc18e1bdf..80aeb268975d8acbc7f7d1c2e771c17e7adfa719 100644 (file)
--- a/sdk/go/streamer/streamer_test.go
+++ b/sdk/go/streamer/streamer_test.go
@@ -251,6 +251,7 @@ func (s *StandaloneSuite) TestTransferShortBuffer(c *C) {
 
        n, err := sr.Read(out)
        c.Check(n, Equals, 100)
+       c.Check(err, IsNil)
 
        n, err = sr.Read(out)
        c.Check(n, Equals, 0)

diff --git a/sdk/go/streamer/transfer.go b/sdk/go/streamer/transfer.go
index a4a194f69bcc8fbdbf43650caa881325bde5dc24..3f5f9344c521f3021e2f6fd35f025e4b9c7fb95e 100644 (file)
--- a/sdk/go/streamer/transfer.go
+++ b/sdk/go/streamer/transfer.go
@@ -249,9 +249,7 @@ func (this *AsyncStream) transfer(source_reader io.Reader) {
                                        }
                                }
                        } else {
-                               if reader_status == io.EOF {
-                                       // no more reads expected, so this is ok
-                               } else {
+                               if reader_status == nil {
                                        // slices channel closed without signaling EOF
                                        reader_status = io.ErrUnexpectedEOF
                                }

diff --git a/services/keepstore/azure_blob_volume.go b/services/keepstore/azure_blob_volume.go
index 657c3151caade8aeca6cbedcd96ffef70d3d7ac4..e9fda2ab76dd41c6cdde36cb139e3f3f030c273b 100644 (file)
--- 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
@@ -169,6 +179,7 @@ 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 {
@@ -178,6 +189,7 @@ func (v *AzureBlobVolume) Compare(loc string, expect []byte) error {
        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
@@ -185,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
@@ -194,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 {
@@ -202,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,
@@ -216,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
@@ -233,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
@@ -256,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,
@@ -264,14 +285,19 @@ 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
 }
@@ -289,3 +315,8 @@ func (v *AzureBlobVolume) translateError(err error) error {
                return err
        }
 }
+
+var keepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
+func (v *AzureBlobVolume) isKeepBlock(s string) bool {
+       return keepBlockRegexp.MatchString(s)
+}

diff --git a/services/keepstore/azure_blob_volume_test.go b/services/keepstore/azure_blob_volume_test.go
index a4c6e62a7d5f6f02c938ae27872cb96bb6a8e3e6..a240c23e1622b525f62a6957a23c025651f94190 100644 (file)
--- a/services/keepstore/azure_blob_volume_test.go
+++ b/services/keepstore/azure_blob_volume_test.go
@@ -60,11 +60,11 @@ func newAzStubHandler() *azStubHandler {
 }
 
 func (h *azStubHandler) TouchWithDate(container, hash string, t time.Time) {
-       if blob, ok := h.blobs[container+"|"+hash]; !ok {
+       blob, ok := h.blobs[container+"|"+hash]
+       if !ok {
                return
-       } else {
-               blob.Mtime = t
        }
+       blob.Mtime = t
 }
 
 func (h *azStubHandler) PutRaw(container, hash string, data []byte) {
@@ -427,7 +427,7 @@ func TestAzureBlobVolumeCreateBlobRaceDeadline(t *testing.T) {
        azureWriteRaceInterval = 2 * time.Second
        azureWriteRacePollTime = 5 * time.Millisecond
 
-       v.PutRaw(TestHash, []byte{})
+       v.PutRaw(TestHash, nil)
 
        buf := new(bytes.Buffer)
        v.IndexTo("", buf)

diff --git a/services/keepstore/volume_generic_test.go b/services/keepstore/volume_generic_test.go
index 7ef079f1f7ac5b588c87e31918739db77a22ab82..61088f10fa2d4ef30e969a77e107b824073684e6 100644 (file)
--- a/services/keepstore/volume_generic_test.go
+++ b/services/keepstore/volume_generic_test.go
@@ -350,6 +350,13 @@ func testIndexTo(t *testing.T, factory TestableVolumeFactory) {
        v.PutRaw(TestHash2, TestBlock2)
        v.PutRaw(TestHash3, TestBlock3)
 
+       // Blocks whose names aren't Keep hashes should be omitted from
+       // index
+       v.PutRaw("fffffffffnotreallyahashfffffffff", nil)
+       v.PutRaw("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF", nil)
+       v.PutRaw("f0000000000000000000000000000000f", nil)
+       v.PutRaw("f00", nil)
+
        buf := new(bytes.Buffer)
        v.IndexTo("", buf)
        indexRows := strings.Split(string(buf.Bytes()), "\n")

diff --git a/services/keepstore/volume_unix.go b/services/keepstore/volume_unix.go
index e745eb269100c5b5b0ba5ff7e1df6c62ab5cb089..910cc25d613cb7690f944b418aebf5c205c7aced 100644 (file)
--- a/services/keepstore/volume_unix.go
+++ b/services/keepstore/volume_unix.go
@@ -292,6 +292,7 @@ func (v *UnixVolume) Status() *VolumeStatus {
 }
 
 var blockDirRe = regexp.MustCompile(`^[0-9a-f]+$`)
+var blockFileRe = regexp.MustCompile(`^[0-9a-f]{32}$`)
 
 // IndexTo writes (to the given Writer) a list of blocks found on this
 // volume which begin with the specified prefix. If the prefix is an
@@ -348,6 +349,9 @@ func (v *UnixVolume) IndexTo(prefix string, w io.Writer) error {
                        if !strings.HasPrefix(name, prefix) {
                                continue
                        }
+                       if !blockFileRe.MatchString(name) {
+                               continue
+                       }
                        _, err = fmt.Fprint(w,
                                name,
                                "+", fileInfo[0].Size(),