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 NotFoundError if trash marker is found on the block
 138 func (v *AzureBlobVolume) checkTrashed(loc string) (bool, error) {
 139         metadata, err := v.bsClient.GetBlobMetadata(v.containerName, loc)
 140         if err != nil {
 141                 return false, v.translateError(err)
 142         }
 143         if metadata["expires_at"] != "" {
 144                 return true, v.translateError(NotFoundError)
 145         }
 146         return false, 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 == true {
 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 == true {
 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         extraHeaders := make(map[string]string)
 285         extraHeaders["x-ms-meta-last_write_at"] = fmt.Sprintf("%d", time.Now().Add(trashLifetime).Unix())
 286         return v.bsClient.CreateBlockBlobFromReader(v.containerName, loc, uint64(len(block)), bytes.NewReader(block), extraHeaders)
 287 }
 288
 289 func (v *AzureBlobVolume) addToMetadata(loc, name, value string) error {
 290         metadata, err := v.bsClient.GetBlobMetadata(v.containerName, loc)
 291         if err != nil {
 292                 return err
 293         }
 294         metadata[name] = value
 295         return v.bsClient.SetBlobMetadata(v.containerName, loc, metadata)
 296 }
 297
 298 func (v *AzureBlobVolume) removeFromMetadata(loc, name string) error {
 299         metadata, err := v.bsClient.GetBlobMetadata(v.containerName, loc)
 300         if err != nil {
 301                 return err
 302         }
 303         metadata[name] = ""
 304         return v.bsClient.SetBlobMetadata(v.containerName, loc, metadata)
 305 }
 306
 307 // Touch updates the last-modified property of a block blob.
 308 func (v *AzureBlobVolume) Touch(loc string) error {
 309         if v.readonly {
 310                 return MethodDisabledError
 311         }
 312         trashed, err := v.checkTrashed(loc)
 313         if err != nil {
 314                 return err
 315         }
 316         if trashed == true {
 317                 return os.ErrNotExist
 318         }
 319         return v.addToMetadata(loc, "last_write_at", fmt.Sprintf("%d", time.Now()))
 320 }
 321
 322 // Mtime returns the last-modified property of a block blob.
 323 func (v *AzureBlobVolume) Mtime(loc string) (time.Time, error) {
 324         trashed, err := v.checkTrashed(loc)
 325         if err != nil {
 326                 return time.Time{}, err
 327         }
 328         if trashed == true {
 329                 return time.Time{}, os.ErrNotExist
 330         }
 331         metadata, err := v.bsClient.GetBlobMetadata(v.containerName, loc)
 332         if err != nil {
 333                 return time.Time{}, v.translateError(err)
 334         }
 335
 336         lastWriteAt, err := strconv.ParseInt(metadata["last_write_at"], 10, 64)
 337         if err != nil {
 338                 return time.Time{}, v.translateError(err)
 339         }
 340         return time.Unix(lastWriteAt, 0), nil
 341 }
 342
 343 // IndexTo writes a list of Keep blocks that are stored in the
 344 // container.
 345 func (v *AzureBlobVolume) IndexTo(prefix string, writer io.Writer) error {
 346         params := storage.ListBlobsParameters{
 347                 Prefix: prefix,
 348         }
 349         for {
 350                 resp, err := v.bsClient.ListBlobs(v.containerName, params)
 351                 if err != nil {
 352                         return err
 353                 }
 354                 for _, b := range resp.Blobs {
 355                         t, err := time.Parse(time.RFC1123, b.Properties.LastModified)
 356                         if err != nil {
 357                                 return err
 358                         }
 359                         if !v.isKeepBlock(b.Name) {
 360                                 continue
 361                         }
 362                         if b.Properties.ContentLength == 0 && t.Add(azureWriteRaceInterval).After(time.Now()) {
 363                                 // A new zero-length blob is probably
 364                                 // just a new non-empty blob that
 365                                 // hasn't committed its data yet (see
 366                                 // Get()), and in any case has no
 367                                 // value.
 368                                 continue
 369                         }
 370                         if b.Metadata["expires_at"] != "" {
 371                                 // Trashed blob; exclude it from response
 372                                 continue
 373                         }
 374                         fmt.Fprintf(writer, "%s+%d %d\n", b.Name, b.Properties.ContentLength, t.Unix())
 375                 }
 376                 if resp.NextMarker == "" {
 377                         return nil
 378                 }
 379                 params.Marker = resp.NextMarker
 380         }
 381 }
 382
 383 // Trash a Keep block.
 384 func (v *AzureBlobVolume) Trash(loc string) error {
 385         if v.readonly {
 386                 return MethodDisabledError
 387         }
 388
 389         // Ideally we would use If-Unmodified-Since, but that
 390         // particular condition seems to be ignored by Azure. Instead,
 391         // we get the Etag before checking Mtime, and use If-Match to
 392         // ensure we don't delete data if Put() or Touch() happens
 393         // between our calls to Mtime() and DeleteBlob().
 394         props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 395         if err != nil {
 396                 return err
 397         }
 398         if t, err := v.Mtime(loc); err != nil {
 399                 return err
 400         } else if time.Since(t) < blobSignatureTTL {
 401                 return nil
 402         }
 403         if trashLifetime == 0 {
 404                 return v.bsClient.DeleteBlob(v.containerName, loc, map[string]string{
 405                         "If-Match": props.Etag,
 406                 })
 407         }
 408         // Mark as trash
 409         err = v.addToMetadata(loc, "expires_at", fmt.Sprintf("%d", time.Now().Add(trashLifetime).Unix()))
 410         if err != nil {
 411                 return err
 412         }
 413         return v.bsClient.CreateBlockBlobFromReader(v.containerName,
 414                 fmt.Sprintf("trash.%d.%v", time.Now().Add(trashLifetime).Unix(), loc), 0, nil, nil)
 415 }
 416
 417 // Untrash a Keep block.
 418 // Delete the expires_at metadata attribute and trash marker
 419 func (v *AzureBlobVolume) Untrash(loc string) error {
 420         // if expires_at does not exist, return NotFoundError
 421         metadata, err := v.bsClient.GetBlobMetadata(v.containerName, loc)
 422         if err != nil {
 423                 return v.translateError(err)
 424         }
 425         if metadata["expires_at"] == "" {
 426                 return v.translateError(NotFoundError)
 427         }
 428         // reset expires_at metadata attribute
 429         err = v.removeFromMetadata(loc, "expires_at")
 430         if err != nil {
 431                 return v.translateError(err)
 432         }
 433
 434         // delete trash marker if exists
 435         _, err = v.bsClient.DeleteBlobIfExists(v.containerName, fmt.Sprintf("trash.%v.%v", metadata["expires_at"], loc), map[string]string{})
 436         return v.translateError(err)
 437 }
 438
 439 // Status returns a VolumeStatus struct with placeholder data.
 440 func (v *AzureBlobVolume) Status() *VolumeStatus {
 441         return &VolumeStatus{
 442                 DeviceNum: 1,
 443                 BytesFree: BlockSize * 1000,
 444                 BytesUsed: 1,
 445         }
 446 }
 447
 448 // String returns a volume label, including the container name.
 449 func (v *AzureBlobVolume) String() string {
 450         return fmt.Sprintf("azure-storage-container:%+q", v.containerName)
 451 }
 452
 453 // Writable returns true, unless the -readonly flag was on when the
 454 // volume was added.
 455 func (v *AzureBlobVolume) Writable() bool {
 456         return !v.readonly
 457 }
 458
 459 // Replication returns the replication level of the container, as
 460 // specified by the -azure-storage-replication argument.
 461 func (v *AzureBlobVolume) Replication() int {
 462         return v.replication
 463 }
 464
 465 // If possible, translate an Azure SDK error to a recognizable error
 466 // like os.ErrNotExist.
 467 func (v *AzureBlobVolume) translateError(err error) error {
 468         switch {
 469         case err == nil:
 470                 return err
 471         case strings.Contains(err.Error(), "Not Found"):
 472                 // "storage: service returned without a response body (404 Not Found)"
 473                 return os.ErrNotExist
 474         default:
 475                 return err
 476         }
 477 }
 478
 479 var keepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
 480
 481 func (v *AzureBlobVolume) isKeepBlock(s string) bool {
 482         return keepBlockRegexp.MatchString(s)
 483 }
 484
 485 var azTrashLocRegexp = regexp.MustCompile(`trash\.(\d+)\.([0-9a-f]{32})$`)
 486
 487 // EmptyTrash looks for trashed blocks that exceeded trashLifetime
 488 // and deletes them from the volume.
 489 func (v *AzureBlobVolume) EmptyTrash() {
 490         var bytesDeleted, bytesInTrash int64
 491         var blocksDeleted, blocksInTrash int
 492         params := storage.ListBlobsParameters{
 493                 Prefix: "trash.",
 494         }
 495 blobListPage:
 496         for {
 497                 resp, err := v.bsClient.ListBlobs(v.containerName, params)
 498                 if err != nil {
 499                         log.Printf("EmptyTrash: ListBlobs: %v", err)
 500                         break
 501                 }
 502                 for _, b := range resp.Blobs {
 503                         matches := azTrashLocRegexp.FindStringSubmatch(b.Name)
 504                         if len(matches) != 3 {
 505                                 log.Printf("EmptyTrash: regexp mismatch for: %v", b.Name)
 506                                 continue
 507                         }
 508                         blocksInTrash++
 509                         deadline, err := strconv.ParseInt(matches[1], 10, 64)
 510                         if err != nil {
 511                                 log.Printf("EmptyTrash: %v: ParseInt(%v): %v", matches[1], err)
 512                                 continue
 513                         }
 514                         if deadline > time.Now().Unix() {
 515                                 break blobListPage
 516                         }
 517
 518                         metadata, err := v.bsClient.GetBlobMetadata(v.containerName, matches[2])
 519                         if err != nil {
 520                                 log.Printf("EmptyTrash: %v: GetBlobMetadata(%v): %v", matches[2], err)
 521                                 continue
 522                         }
 523
 524                         // Make sure the marker is for the current block, not an older one
 525                         if metadata["expires_at"] == matches[1] {
 526                                 err = v.bsClient.DeleteBlob(v.containerName, matches[2], map[string]string{})
 527                                 if err != nil {
 528                                         log.Printf("EmptyTrash: %v: DeleteBlob(%v): %v", matches[2], err)
 529                                 }
 530                                 blocksDeleted++
 531                         }
 532
 533                         err = v.bsClient.DeleteBlob(v.containerName, b.Name, map[string]string{})
 534                         if err != nil {
 535                                 log.Printf("EmptyTrash: %v: DeleteBlob(%v): %v", b.Name, err)
 536                         }
 537                 }
 538                 if resp.NextMarker == "" {
 539                         break
 540                 }
 541                 params.Marker = resp.NextMarker
 542         }
 543
 544         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)
 545 }