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) ([]byte, error) {
 143         var deadline time.Time
 144         haveDeadline := false
 145         buf, err := v.get(loc)
 146         for err == nil && len(buf) == 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                 bufs.Put(buf)
 167                 time.Sleep(azureWriteRacePollTime)
 168                 buf, err = v.get(loc)
 169         }
 170         if haveDeadline {
 171                 log.Printf("Race ended with len(buf)==%d", len(buf))
 172         }
 173         return buf, err
 174 }
 175
 176 func (v *AzureBlobVolume) get(loc string) ([]byte, error) {
 177         expectSize := BlockSize
 178         if azureMaxGetBytes < BlockSize {
 179                 // Unfortunately the handler doesn't tell us how long the blob
 180                 // is expected to be, so we have to ask Azure.
 181                 props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 182                 if err != nil {
 183                         return nil, v.translateError(err)
 184                 }
 185                 if props.ContentLength > int64(BlockSize) || props.ContentLength < 0 {
 186                         return nil, fmt.Errorf("block %s invalid size %d (max %d)", loc, props.ContentLength, BlockSize)
 187                 }
 188                 expectSize = int(props.ContentLength)
 189         }
 190
 191         buf := bufs.Get(expectSize)
 192         if expectSize == 0 {
 193                 return buf, nil
 194         }
 195
 196         // We'll update this actualSize if/when we get the last piece.
 197         actualSize := -1
 198         pieces := (expectSize + azureMaxGetBytes - 1) / azureMaxGetBytes
 199         errors := make([]error, pieces)
 200         var wg sync.WaitGroup
 201         wg.Add(pieces)
 202         for p := 0; p < pieces; p++ {
 203                 go func(p int) {
 204                         defer wg.Done()
 205                         startPos := p * azureMaxGetBytes
 206                         endPos := startPos + azureMaxGetBytes
 207                         if endPos > expectSize {
 208                                 endPos = expectSize
 209                         }
 210                         var rdr io.ReadCloser
 211                         var err error
 212                         if startPos == 0 && endPos == expectSize {
 213                                 rdr, err = v.bsClient.GetBlob(v.containerName, loc)
 214                         } else {
 215                                 rdr, err = v.bsClient.GetBlobRange(v.containerName, loc, fmt.Sprintf("%d-%d", startPos, endPos-1))
 216                         }
 217                         if err != nil {
 218                                 errors[p] = err
 219                                 return
 220                         }
 221                         defer rdr.Close()
 222                         n, err := io.ReadFull(rdr, buf[startPos:endPos])
 223                         if pieces == 1 && (err == io.ErrUnexpectedEOF || err == io.EOF) {
 224                                 // If we don't know the actual size,
 225                                 // and just tried reading 64 MiB, it's
 226                                 // normal to encounter EOF.
 227                         } else if err != nil {
 228                                 errors[p] = err
 229                         }
 230                         if p == pieces-1 {
 231                                 actualSize = startPos + n
 232                         }
 233                 }(p)
 234         }
 235         wg.Wait()
 236         for _, err := range errors {
 237                 if err != nil {
 238                         bufs.Put(buf)
 239                         return nil, v.translateError(err)
 240                 }
 241         }
 242         return buf[:actualSize], nil
 243 }
 244
 245 // Compare the given data with existing stored data.
 246 func (v *AzureBlobVolume) Compare(loc string, expect []byte) error {
 247         rdr, err := v.bsClient.GetBlob(v.containerName, loc)
 248         if err != nil {
 249                 return v.translateError(err)
 250         }
 251         defer rdr.Close()
 252         return compareReaderWithBuf(rdr, expect, loc[:32])
 253 }
 254
 255 // Put stores a Keep block as a block blob in the container.
 256 func (v *AzureBlobVolume) Put(loc string, block []byte) error {
 257         if v.readonly {
 258                 return MethodDisabledError
 259         }
 260         return v.bsClient.CreateBlockBlobFromReader(v.containerName, loc, uint64(len(block)), bytes.NewReader(block))
 261 }
 262
 263 // Touch updates the last-modified property of a block blob.
 264 func (v *AzureBlobVolume) Touch(loc string) error {
 265         if v.readonly {
 266                 return MethodDisabledError
 267         }
 268         return v.bsClient.SetBlobMetadata(v.containerName, loc, map[string]string{
 269                 "touch": fmt.Sprintf("%d", time.Now()),
 270         })
 271 }
 272
 273 // Mtime returns the last-modified property of a block blob.
 274 func (v *AzureBlobVolume) Mtime(loc string) (time.Time, error) {
 275         props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 276         if err != nil {
 277                 return time.Time{}, err
 278         }
 279         return time.Parse(time.RFC1123, props.LastModified)
 280 }
 281
 282 // IndexTo writes a list of Keep blocks that are stored in the
 283 // container.
 284 func (v *AzureBlobVolume) IndexTo(prefix string, writer io.Writer) error {
 285         params := storage.ListBlobsParameters{
 286                 Prefix: prefix,
 287         }
 288         for {
 289                 resp, err := v.bsClient.ListBlobs(v.containerName, params)
 290                 if err != nil {
 291                         return err
 292                 }
 293                 for _, b := range resp.Blobs {
 294                         t, err := time.Parse(time.RFC1123, b.Properties.LastModified)
 295                         if err != nil {
 296                                 return err
 297                         }
 298                         if !v.isKeepBlock(b.Name) {
 299                                 continue
 300                         }
 301                         if b.Properties.ContentLength == 0 && t.Add(azureWriteRaceInterval).After(time.Now()) {
 302                                 // A new zero-length blob is probably
 303                                 // just a new non-empty blob that
 304                                 // hasn't committed its data yet (see
 305                                 // Get()), and in any case has no
 306                                 // value.
 307                                 continue
 308                         }
 309                         fmt.Fprintf(writer, "%s+%d %d\n", b.Name, b.Properties.ContentLength, t.Unix())
 310                 }
 311                 if resp.NextMarker == "" {
 312                         return nil
 313                 }
 314                 params.Marker = resp.NextMarker
 315         }
 316 }
 317
 318 // Trash a Keep block.
 319 func (v *AzureBlobVolume) Trash(loc string) error {
 320         if v.readonly {
 321                 return MethodDisabledError
 322         }
 323
 324         if trashLifetime != 0 {
 325                 return ErrNotImplemented
 326         }
 327
 328         // Ideally we would use If-Unmodified-Since, but that
 329         // particular condition seems to be ignored by Azure. Instead,
 330         // we get the Etag before checking Mtime, and use If-Match to
 331         // ensure we don't delete data if Put() or Touch() happens
 332         // between our calls to Mtime() and DeleteBlob().
 333         props, err := v.bsClient.GetBlobProperties(v.containerName, loc)
 334         if err != nil {
 335                 return err
 336         }
 337         if t, err := v.Mtime(loc); err != nil {
 338                 return err
 339         } else if time.Since(t) < blobSignatureTTL {
 340                 return nil
 341         }
 342         return v.bsClient.DeleteBlob(v.containerName, loc, map[string]string{
 343                 "If-Match": props.Etag,
 344         })
 345 }
 346
 347 // Untrash a Keep block.
 348 // TBD
 349 func (v *AzureBlobVolume) Untrash(loc string) error {
 350         return ErrNotImplemented
 351 }
 352
 353 // Status returns a VolumeStatus struct with placeholder data.
 354 func (v *AzureBlobVolume) Status() *VolumeStatus {
 355         return &VolumeStatus{
 356                 DeviceNum: 1,
 357                 BytesFree: BlockSize * 1000,
 358                 BytesUsed: 1,
 359         }
 360 }
 361
 362 // String returns a volume label, including the container name.
 363 func (v *AzureBlobVolume) String() string {
 364         return fmt.Sprintf("azure-storage-container:%+q", v.containerName)
 365 }
 366
 367 // Writable returns true, unless the -readonly flag was on when the
 368 // volume was added.
 369 func (v *AzureBlobVolume) Writable() bool {
 370         return !v.readonly
 371 }
 372
 373 // Replication returns the replication level of the container, as
 374 // specified by the -azure-storage-replication argument.
 375 func (v *AzureBlobVolume) Replication() int {
 376         return v.replication
 377 }
 378
 379 // If possible, translate an Azure SDK error to a recognizable error
 380 // like os.ErrNotExist.
 381 func (v *AzureBlobVolume) translateError(err error) error {
 382         switch {
 383         case err == nil:
 384                 return err
 385         case strings.Contains(err.Error(), "404 Not Found"):
 386                 // "storage: service returned without a response body (404 Not Found)"
 387                 return os.ErrNotExist
 388         default:
 389                 return err
 390         }
 391 }
 392
 393 var keepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
 394
 395 func (v *AzureBlobVolume) isKeepBlock(s string) bool {
 396         return keepBlockRegexp.MatchString(s)
 397 }