services/keepstore/azure_blob_volume.go

   1 package main
   2
   3 import (
   4         "bytes"
   5         "errors"
   6         "flag"
   7         "fmt"
   8         "io"
   9         "io/ioutil"
  10         "log"
  11         "os"
  12         "regexp"
  13         "strings"
  14         "sync"
  15         "time"
  16
  17         "github.com/curoverse/azure-sdk-for-go/storage"
  18 )
  19
  20 var (
  21         azureMaxGetBytes           int
  22         azureStorageAccountName    string
  23         azureStorageAccountKeyFile string
  24         azureStorageReplication    int
  25         azureWriteRaceInterval     = 15 * time.Second
  26         azureWriteRacePollTime     = time.Second
  27 )
  28
  29 func readKeyFromFile(file string) (string, error) {
  30         buf, err := ioutil.ReadFile(file)
  31         if err != nil {
  32                 return "", errors.New("reading key from " + file + ": " + err.Error())
  33         }
  34         accountKey := strings.TrimSpace(string(buf))
  35         if accountKey == "" {
  36                 return "", errors.New("empty account key in " + file)
  37         }
  38         return accountKey, nil
  39 }
  40
  41 type azureVolumeAdder struct {
  42         *volumeSet
  43 }
  44
  45 func (s *azureVolumeAdder) Set(containerName string) error {
  46         if trashLifetime != 0 {
  47                 return ErrNotImplemented
  48         }
  49
  50         if containerName == "" {
  51                 return errors.New("no container name given")
  52         }
  53         if azureStorageAccountName == "" || azureStorageAccountKeyFile == "" {
  54                 return errors.New("-azure-storage-account-name and -azure-storage-account-key-file arguments must given before -azure-storage-container-volume")
  55         }
  56         accountKey, err := readKeyFromFile(azureStorageAccountKeyFile)
  57         if err != nil {
  58                 return err
  59         }
  60         azClient, err := storage.NewBasicClient(azureStorageAccountName, accountKey)
  61         if err != nil {
  62                 return errors.New("creating Azure storage client: " + err.Error())
  63         }
  64         if flagSerializeIO {
  65                 log.Print("Notice: -serialize is not supported by azure-blob-container volumes.")
  66         }
  67         v := NewAzureBlobVolume(azClient, containerName, flagReadonly, azureStorageReplication)
  68         if err := v.Check(); err != nil {
  69                 return err
  70         }
  71         *s.volumeSet = append(*s.volumeSet, v)
  72         return nil
  73 }
  74
  75 func init() {
  76         flag.Var(&azureVolumeAdder{&volumes},
  77                 "azure-storage-container-volume",
  78                 "Use the given container as a storage volume. Can be given multiple times.")
  79         flag.StringVar(
  80                 &azureStorageAccountName,
  81                 "azure-storage-account-name",
  82                 "",
  83                 "Azure storage account name used for subsequent --azure-storage-container-volume arguments.")
  84         flag.StringVar(
  85                 &azureStorageAccountKeyFile,
  86                 "azure-storage-account-key-file",
  87                 "",
  88                 "File containing the account key used for subsequent --azure-storage-container-volume arguments.")
  89         flag.IntVar(
  90                 &azureStorageReplication,
  91                 "azure-storage-replication",
  92                 3,
  93                 "Replication level to report to clients when data is stored in an Azure container.")
  94         flag.IntVar(
  95                 &azureMaxGetBytes,
  96                 "azure-max-get-bytes",
  97                 BlockSize,
  98                 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))
  99 }
 100
 101 // An AzureBlobVolume stores and retrieves blocks in an Azure Blob
 102 // container.
 103 type AzureBlobVolume struct {
 104         azClient      storage.Client
 105         bsClient      storage.BlobStorageClient
 106         containerName string
 107         readonly      bool
 108         replication   int
 109 }
 110
 111 // NewAzureBlobVolume returns a new AzureBlobVolume using the given
 112 // client and container name. The replication argument specifies the
 113 // replication level to report when writing data.
 114 func NewAzureBlobVolume(client storage.Client, containerName string, readonly bool, replication int) *AzureBlobVolume {
 115         return &AzureBlobVolume{
 116                 azClient:      client,
 117                 bsClient:      client.GetBlobService(),
 118                 containerName: containerName,
 119                 readonly:      readonly,
 120                 replication:   replication,
 121         }
 122 }
 123
 124 // Check returns nil if the volume is usable.
 125 func (v *AzureBlobVolume) Check() error {
 126         ok, err := v.bsClient.ContainerExists(v.containerName)
 127         if err != nil {
 128                 return err
 129         }
 130         if !ok {
 131                 return errors.New("container does not exist")
 132         }
 133         return nil
 134 }
 135
 136 // Get reads a Keep block that has been stored as a block blob in the
 137 // container.
 138 //
 139 // If the block is younger than azureWriteRaceInterval and is
 140 // unexpectedly empty, assume a PutBlob operation is in progress, and
 141 // wait for it to finish writing.
 142 func (v *AzureBlobVolume) Get(loc string, buf []byte) (int, error) {
 143         var deadline time.Time
 144         haveDeadline := false
 145         size, err := v.get(loc, buf)
 146         for err == nil && size == 0 && loc != "d41d8cd98f00b204e9800998ecf8427e" {
 147                 // Seeing a brand new empty block probably means we're
 148                 // in a race with CreateBlob, which under the hood
 149                 // (apparently) does "CreateEmpty" and "CommitData"
 150                 // with no additional transaction locking.
 151                 if !haveDeadline {
 152                         t, err := v.Mtime(loc)
 153                         if err != nil {
 154                                 log.Print("Got empty block (possible race) but Mtime failed: ", err)
 155                                 break
 156                         }
 157                         deadline = t.Add(azureWriteRaceInterval)
 158                         if time.Now().After(deadline) {
 159                                 break
 160                         }
 161                         log.Printf("Race? Block %s is 0 bytes, %s old. Polling until %s", loc, time.Since(t), deadline)
 162                         haveDeadline = true
 163                 } else if time.Now().After(deadline) {
 164                         break
 165                 }
 166                 time.Sleep(azureWriteRacePollTime)
 167                 size, err = v.get(loc, buf)
 168         }
 169         if haveDeadline {
 170                 log.Printf("Race ended with size==%d", size)
 171         }
 172         return size, err
 173 }
 174
 175 func (v *AzureBlobVolume) get(loc string, buf []byte) (int, error) {
 176         expectSize := len(buf)
 177         if azureMaxGetBytes < BlockSize {
 178                 // Unfortunately the handler doesn't tell us how long the blob
 179                 // is expected to be, so we have to ask Azure.
 180                 props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 181                 if err != nil {
 182                         return 0, v.translateError(err)
 183                 }
 184                 if props.ContentLength > int64(BlockSize) || props.ContentLength < 0 {
 185                         return 0, fmt.Errorf("block %s invalid size %d (max %d)", loc, props.ContentLength, BlockSize)
 186                 }
 187                 expectSize = int(props.ContentLength)
 188         }
 189
 190         if expectSize == 0 {
 191                 return 0, nil
 192         }
 193
 194         // We'll update this actualSize if/when we get the last piece.
 195         actualSize := -1
 196         pieces := (expectSize + azureMaxGetBytes - 1) / azureMaxGetBytes
 197         errors := make([]error, pieces)
 198         var wg sync.WaitGroup
 199         wg.Add(pieces)
 200         for p := 0; p < pieces; p++ {
 201                 go func(p int) {
 202                         defer wg.Done()
 203                         startPos := p * azureMaxGetBytes
 204                         endPos := startPos + azureMaxGetBytes
 205                         if endPos > expectSize {
 206                                 endPos = expectSize
 207                         }
 208                         var rdr io.ReadCloser
 209                         var err error
 210                         if startPos == 0 && endPos == expectSize {
 211                                 rdr, err = v.bsClient.GetBlob(v.containerName, loc)
 212                         } else {
 213                                 rdr, err = v.bsClient.GetBlobRange(v.containerName, loc, fmt.Sprintf("%d-%d", startPos, endPos-1))
 214                         }
 215                         if err != nil {
 216                                 errors[p] = err
 217                                 return
 218                         }
 219                         defer rdr.Close()
 220                         n, err := io.ReadFull(rdr, buf[startPos:endPos])
 221                         if pieces == 1 && (err == io.ErrUnexpectedEOF || err == io.EOF) {
 222                                 // If we don't know the actual size,
 223                                 // and just tried reading 64 MiB, it's
 224                                 // normal to encounter EOF.
 225                         } else if err != nil {
 226                                 errors[p] = err
 227                         }
 228                         if p == pieces-1 {
 229                                 actualSize = startPos + n
 230                         }
 231                 }(p)
 232         }
 233         wg.Wait()
 234         for _, err := range errors {
 235                 if err != nil {
 236                         return 0, v.translateError(err)
 237                 }
 238         }
 239         return actualSize, nil
 240 }
 241
 242 // Compare the given data with existing stored data.
 243 func (v *AzureBlobVolume) Compare(loc string, expect []byte) error {
 244         rdr, err := v.bsClient.GetBlob(v.containerName, loc)
 245         if err != nil {
 246                 return v.translateError(err)
 247         }
 248         defer rdr.Close()
 249         return compareReaderWithBuf(rdr, expect, loc[:32])
 250 }
 251
 252 // Put stores a Keep block as a block blob in the container.
 253 func (v *AzureBlobVolume) Put(loc string, block []byte) error {
 254         if v.readonly {
 255                 return MethodDisabledError
 256         }
 257         return v.bsClient.CreateBlockBlobFromReader(v.containerName, loc, uint64(len(block)), bytes.NewReader(block), nil)
 258 }
 259
 260 // Touch updates the last-modified property of a block blob.
 261 func (v *AzureBlobVolume) Touch(loc string) error {
 262         if v.readonly {
 263                 return MethodDisabledError
 264         }
 265         return v.bsClient.SetBlobMetadata(v.containerName, loc, map[string]string{
 266                 "touch": fmt.Sprintf("%d", time.Now()),
 267         })
 268 }
 269
 270 // Mtime returns the last-modified property of a block blob.
 271 func (v *AzureBlobVolume) Mtime(loc string) (time.Time, error) {
 272         props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 273         if err != nil {
 274                 return time.Time{}, err
 275         }
 276         return time.Parse(time.RFC1123, props.LastModified)
 277 }
 278
 279 // IndexTo writes a list of Keep blocks that are stored in the
 280 // container.
 281 func (v *AzureBlobVolume) IndexTo(prefix string, writer io.Writer) error {
 282         params := storage.ListBlobsParameters{
 283                 Prefix: prefix,
 284         }
 285         for {
 286                 resp, err := v.bsClient.ListBlobs(v.containerName, params)
 287                 if err != nil {
 288                         return err
 289                 }
 290                 for _, b := range resp.Blobs {
 291                         t, err := time.Parse(time.RFC1123, b.Properties.LastModified)
 292                         if err != nil {
 293                                 return err
 294                         }
 295                         if !v.isKeepBlock(b.Name) {
 296                                 continue
 297                         }
 298                         if b.Properties.ContentLength == 0 && t.Add(azureWriteRaceInterval).After(time.Now()) {
 299                                 // A new zero-length blob is probably
 300                                 // just a new non-empty blob that
 301                                 // hasn't committed its data yet (see
 302                                 // Get()), and in any case has no
 303                                 // value.
 304                                 continue
 305                         }
 306                         fmt.Fprintf(writer, "%s+%d %d\n", b.Name, b.Properties.ContentLength, t.Unix())
 307                 }
 308                 if resp.NextMarker == "" {
 309                         return nil
 310                 }
 311                 params.Marker = resp.NextMarker
 312         }
 313 }
 314
 315 // Trash a Keep block.
 316 func (v *AzureBlobVolume) Trash(loc string) error {
 317         if v.readonly {
 318                 return MethodDisabledError
 319         }
 320
 321         if trashLifetime != 0 {
 322                 return ErrNotImplemented
 323         }
 324
 325         // Ideally we would use If-Unmodified-Since, but that
 326         // particular condition seems to be ignored by Azure. Instead,
 327         // we get the Etag before checking Mtime, and use If-Match to
 328         // ensure we don't delete data if Put() or Touch() happens
 329         // between our calls to Mtime() and DeleteBlob().
 330         props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 331         if err != nil {
 332                 return err
 333         }
 334         if t, err := v.Mtime(loc); err != nil {
 335                 return err
 336         } else if time.Since(t) < blobSignatureTTL {
 337                 return nil
 338         }
 339         return v.bsClient.DeleteBlob(v.containerName, loc, map[string]string{
 340                 "If-Match": props.Etag,
 341         })
 342 }
 343
 344 // Untrash a Keep block.
 345 // TBD
 346 func (v *AzureBlobVolume) Untrash(loc string) error {
 347         return ErrNotImplemented
 348 }
 349
 350 // Status returns a VolumeStatus struct with placeholder data.
 351 func (v *AzureBlobVolume) Status() *VolumeStatus {
 352         return &VolumeStatus{
 353                 DeviceNum: 1,
 354                 BytesFree: BlockSize * 1000,
 355                 BytesUsed: 1,
 356         }
 357 }
 358
 359 // String returns a volume label, including the container name.
 360 func (v *AzureBlobVolume) String() string {
 361         return fmt.Sprintf("azure-storage-container:%+q", v.containerName)
 362 }
 363
 364 // Writable returns true, unless the -readonly flag was on when the
 365 // volume was added.
 366 func (v *AzureBlobVolume) Writable() bool {
 367         return !v.readonly
 368 }
 369
 370 // Replication returns the replication level of the container, as
 371 // specified by the -azure-storage-replication argument.
 372 func (v *AzureBlobVolume) Replication() int {
 373         return v.replication
 374 }
 375
 376 // If possible, translate an Azure SDK error to a recognizable error
 377 // like os.ErrNotExist.
 378 func (v *AzureBlobVolume) translateError(err error) error {
 379         switch {
 380         case err == nil:
 381                 return err
 382         case strings.Contains(err.Error(), "404 Not Found"):
 383                 // "storage: service returned without a response body (404 Not Found)"
 384                 return os.ErrNotExist
 385         default:
 386                 return err
 387         }
 388 }
 389
 390 var keepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
 391
 392 func (v *AzureBlobVolume) isKeepBlock(s string) bool {
 393         return keepBlockRegexp.MatchString(s)
 394 }
 395
 396 // EmptyTrash looks for trashed blocks that exceeded trashLifetime
 397 // and deletes them from the volume.
 398 // TBD
 399 func (v *AzureBlobVolume) EmptyTrash() {
 400 }