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