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