X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d1e55b0176e20451df3ad11f7a0c24c0fe666738..80f547e339c452e5b03be5beee00e845d56d8e18:/services/keepstore/azure_blob_volume.go diff --git a/services/keepstore/azure_blob_volume.go b/services/keepstore/azure_blob_volume.go index b3b7613240..c0033d9186 100644 --- a/services/keepstore/azure_blob_volume.go +++ b/services/keepstore/azure_blob_volume.go @@ -11,17 +11,19 @@ import ( "os" "regexp" "strings" + "sync" "time" "github.com/curoverse/azure-sdk-for-go/storage" ) var ( + azureMaxGetBytes int 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) { @@ -85,6 +87,11 @@ func init() { "azure-storage-replication", 3, "Replication level to report to clients when data is stored in an Azure container.") + flag.IntVar( + &azureMaxGetBytes, + "azure-max-get-bytes", + BlockSize, + fmt.Sprintf("Maximum bytes to request in a single GET request. If smaller than %d, use multiple concurrent range requests to retrieve a block.", BlockSize)) } // An AzureBlobVolume stores and retrieves blocks in an Azure Blob @@ -97,6 +104,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, @@ -119,6 +129,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 @@ -154,22 +170,75 @@ 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 { - return nil, v.translateError(err) + expectSize := BlockSize + if azureMaxGetBytes < BlockSize { + // Unfortunately the handler doesn't tell us how long the blob + // is expected to be, so we have to ask Azure. + props, err := v.bsClient.GetBlobProperties(v.containerName, loc) + if err != nil { + return nil, v.translateError(err) + } + if props.ContentLength > int64(BlockSize) || props.ContentLength < 0 { + return nil, fmt.Errorf("block %s invalid size %d (max %d)", loc, props.ContentLength, BlockSize) + } + expectSize = int(props.ContentLength) } - defer rdr.Close() - buf := bufs.Get(BlockSize) - n, err := io.ReadFull(rdr, buf) - switch err { - case nil, io.EOF, io.ErrUnexpectedEOF: - return buf[:n], nil - default: - bufs.Put(buf) - return nil, err + + buf := bufs.Get(expectSize) + if expectSize == 0 { + return buf, nil + } + + // We'll update this actualSize if/when we get the last piece. + actualSize := -1 + pieces := (expectSize + azureMaxGetBytes - 1) / azureMaxGetBytes + errors := make([]error, pieces) + var wg sync.WaitGroup + wg.Add(pieces) + for p := 0; p < pieces; p++ { + go func(p int) { + defer wg.Done() + startPos := p * azureMaxGetBytes + endPos := startPos + azureMaxGetBytes + if endPos > expectSize { + endPos = expectSize + } + var rdr io.ReadCloser + var err error + if startPos == 0 && endPos == expectSize { + rdr, err = v.bsClient.GetBlob(v.containerName, loc) + } else { + rdr, err = v.bsClient.GetBlobRange(v.containerName, loc, fmt.Sprintf("%d-%d", startPos, endPos-1)) + } + if err != nil { + errors[p] = err + return + } + defer rdr.Close() + n, err := io.ReadFull(rdr, buf[startPos:endPos]) + if pieces == 1 && (err == io.ErrUnexpectedEOF || err == io.EOF) { + // If we don't know the actual size, + // and just tried reading 64 MiB, it's + // normal to encounter EOF. + } else if err != nil { + errors[p] = err + } + if p == pieces-1 { + actualSize = startPos + n + } + }(p) + } + wg.Wait() + for _, err := range errors { + if err != nil { + bufs.Put(buf) + return nil, v.translateError(err) + } } + return buf[:actualSize], nil } +// 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 { @@ -179,6 +248,7 @@ func (v *AzureBlobVolume) Compare(loc string, expect []byte) error { return compareReaderWithBuf(rdr, expect, loc[:32]) } +// Put stores a Keep block as a block blob in the container. func (v *AzureBlobVolume) Put(loc string, block []byte) error { if v.readonly { return MethodDisabledError @@ -186,6 +256,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 @@ -195,6 +266,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 { @@ -203,6 +275,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, @@ -237,6 +311,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 +335,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 +344,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 } @@ -295,6 +376,7 @@ func (v *AzureBlobVolume) translateError(err error) error { } var keepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`) + func (v *AzureBlobVolume) isKeepBlock(s string) bool { return keepBlockRegexp.MatchString(s) }