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