services/keepstore/s3_volume.go

   1 package main
   2
   3 import (
   4         "bytes"
   5         "context"
   6         "encoding/base64"
   7         "encoding/hex"
   8         "flag"
   9         "fmt"
  10         "io"
  11         "io/ioutil"
  12         "net/http"
  13         "os"
  14         "regexp"
  15         "strings"
  16         "sync"
  17         "sync/atomic"
  18         "time"
  19
  20         "git.curoverse.com/arvados.git/sdk/go/arvados"
  21         "github.com/AdRoll/goamz/aws"
  22         "github.com/AdRoll/goamz/s3"
  23         log "github.com/Sirupsen/logrus"
  24 )
  25
  26 const (
  27         s3DefaultReadTimeout    = arvados.Duration(10 * time.Minute)
  28         s3DefaultConnectTimeout = arvados.Duration(time.Minute)
  29 )
  30
  31 var (
  32         // ErrS3TrashDisabled is returned by Trash if that operation
  33         // is impossible with the current config.
  34         ErrS3TrashDisabled = fmt.Errorf("trash function is disabled because -trash-lifetime=0 and -s3-unsafe-delete=false")
  35
  36         s3AccessKeyFile string
  37         s3SecretKeyFile string
  38         s3RegionName    string
  39         s3Endpoint      string
  40         s3Replication   int
  41         s3UnsafeDelete  bool
  42         s3RaceWindow    time.Duration
  43
  44         s3ACL = s3.Private
  45 )
  46
  47 const (
  48         maxClockSkew  = 600 * time.Second
  49         nearlyRFC1123 = "Mon, 2 Jan 2006 15:04:05 GMT"
  50 )
  51
  52 type s3VolumeAdder struct {
  53         *Config
  54 }
  55
  56 // String implements flag.Value
  57 func (s *s3VolumeAdder) String() string {
  58         return "-"
  59 }
  60
  61 func (s *s3VolumeAdder) Set(bucketName string) error {
  62         if bucketName == "" {
  63                 return fmt.Errorf("no container name given")
  64         }
  65         if s3AccessKeyFile == "" || s3SecretKeyFile == "" {
  66                 return fmt.Errorf("-s3-access-key-file and -s3-secret-key-file arguments must given before -s3-bucket-volume")
  67         }
  68         if deprecated.flagSerializeIO {
  69                 log.Print("Notice: -serialize is not supported by s3-bucket volumes.")
  70         }
  71         s.Config.Volumes = append(s.Config.Volumes, &S3Volume{
  72                 Bucket:        bucketName,
  73                 AccessKeyFile: s3AccessKeyFile,
  74                 SecretKeyFile: s3SecretKeyFile,
  75                 Endpoint:      s3Endpoint,
  76                 Region:        s3RegionName,
  77                 RaceWindow:    arvados.Duration(s3RaceWindow),
  78                 S3Replication: s3Replication,
  79                 UnsafeDelete:  s3UnsafeDelete,
  80                 ReadOnly:      deprecated.flagReadonly,
  81                 IndexPageSize: 1000,
  82         })
  83         return nil
  84 }
  85
  86 func s3regions() (okList []string) {
  87         for r := range aws.Regions {
  88                 okList = append(okList, r)
  89         }
  90         return
  91 }
  92
  93 func init() {
  94         VolumeTypes = append(VolumeTypes, func() VolumeWithExamples { return &S3Volume{} })
  95
  96         flag.Var(&s3VolumeAdder{theConfig},
  97                 "s3-bucket-volume",
  98                 "Use the given bucket as a storage volume. Can be given multiple times.")
  99         flag.StringVar(
 100                 &s3RegionName,
 101                 "s3-region",
 102                 "",
 103                 fmt.Sprintf("AWS region used for subsequent -s3-bucket-volume arguments. Allowed values are %+q.", s3regions()))
 104         flag.StringVar(
 105                 &s3Endpoint,
 106                 "s3-endpoint",
 107                 "",
 108                 "Endpoint URL used for subsequent -s3-bucket-volume arguments. If blank, use the AWS endpoint corresponding to the -s3-region argument. For Google Storage, use \"https://storage.googleapis.com\".")
 109         flag.StringVar(
 110                 &s3AccessKeyFile,
 111                 "s3-access-key-file",
 112                 "",
 113                 "`File` containing the access key used for subsequent -s3-bucket-volume arguments.")
 114         flag.StringVar(
 115                 &s3SecretKeyFile,
 116                 "s3-secret-key-file",
 117                 "",
 118                 "`File` containing the secret key used for subsequent -s3-bucket-volume arguments.")
 119         flag.DurationVar(
 120                 &s3RaceWindow,
 121                 "s3-race-window",
 122                 24*time.Hour,
 123                 "Maximum eventual consistency latency for subsequent -s3-bucket-volume arguments.")
 124         flag.IntVar(
 125                 &s3Replication,
 126                 "s3-replication",
 127                 2,
 128                 "Replication level reported to clients for subsequent -s3-bucket-volume arguments.")
 129         flag.BoolVar(
 130                 &s3UnsafeDelete,
 131                 "s3-unsafe-delete",
 132                 false,
 133                 "EXPERIMENTAL. Enable deletion (garbage collection), even though there are known race conditions that can cause data loss.")
 134 }
 135
 136 // S3Volume implements Volume using an S3 bucket.
 137 type S3Volume struct {
 138         AccessKeyFile      string
 139         SecretKeyFile      string
 140         Endpoint           string
 141         Region             string
 142         Bucket             string
 143         LocationConstraint bool
 144         IndexPageSize      int
 145         S3Replication      int
 146         ConnectTimeout     arvados.Duration
 147         ReadTimeout        arvados.Duration
 148         RaceWindow         arvados.Duration
 149         ReadOnly           bool
 150         UnsafeDelete       bool
 151
 152         bucket *s3bucket
 153
 154         startOnce sync.Once
 155 }
 156
 157 // Examples implements VolumeWithExamples.
 158 func (*S3Volume) Examples() []Volume {
 159         return []Volume{
 160                 &S3Volume{
 161                         AccessKeyFile:  "/etc/aws_s3_access_key.txt",
 162                         SecretKeyFile:  "/etc/aws_s3_secret_key.txt",
 163                         Endpoint:       "",
 164                         Region:         "us-east-1",
 165                         Bucket:         "example-bucket-name",
 166                         IndexPageSize:  1000,
 167                         S3Replication:  2,
 168                         RaceWindow:     arvados.Duration(24 * time.Hour),
 169                         ConnectTimeout: arvados.Duration(time.Minute),
 170                         ReadTimeout:    arvados.Duration(5 * time.Minute),
 171                 },
 172                 &S3Volume{
 173                         AccessKeyFile:  "/etc/gce_s3_access_key.txt",
 174                         SecretKeyFile:  "/etc/gce_s3_secret_key.txt",
 175                         Endpoint:       "https://storage.googleapis.com",
 176                         Region:         "",
 177                         Bucket:         "example-bucket-name",
 178                         IndexPageSize:  1000,
 179                         S3Replication:  2,
 180                         RaceWindow:     arvados.Duration(24 * time.Hour),
 181                         ConnectTimeout: arvados.Duration(time.Minute),
 182                         ReadTimeout:    arvados.Duration(5 * time.Minute),
 183                 },
 184         }
 185 }
 186
 187 // Type implements Volume.
 188 func (*S3Volume) Type() string {
 189         return "S3"
 190 }
 191
 192 // Start populates private fields and verifies the configuration is
 193 // valid.
 194 func (v *S3Volume) Start() error {
 195         region, ok := aws.Regions[v.Region]
 196         if v.Endpoint == "" {
 197                 if !ok {
 198                         return fmt.Errorf("unrecognized region %+q; try specifying -s3-endpoint instead", v.Region)
 199                 }
 200         } else if ok {
 201                 return fmt.Errorf("refusing to use AWS region name %+q with endpoint %+q; "+
 202                         "specify empty endpoint (\"-s3-endpoint=\") or use a different region name", v.Region, v.Endpoint)
 203         } else {
 204                 region = aws.Region{
 205                         Name:                 v.Region,
 206                         S3Endpoint:           v.Endpoint,
 207                         S3LocationConstraint: v.LocationConstraint,
 208                 }
 209         }
 210
 211         var err error
 212         var auth aws.Auth
 213         auth.AccessKey, err = readKeyFromFile(v.AccessKeyFile)
 214         if err != nil {
 215                 return err
 216         }
 217         auth.SecretKey, err = readKeyFromFile(v.SecretKeyFile)
 218         if err != nil {
 219                 return err
 220         }
 221
 222         // Zero timeouts mean "wait forever", which is a bad
 223         // default. Default to long timeouts instead.
 224         if v.ConnectTimeout == 0 {
 225                 v.ConnectTimeout = s3DefaultConnectTimeout
 226         }
 227         if v.ReadTimeout == 0 {
 228                 v.ReadTimeout = s3DefaultReadTimeout
 229         }
 230
 231         client := s3.New(auth, region)
 232         client.ConnectTimeout = time.Duration(v.ConnectTimeout)
 233         client.ReadTimeout = time.Duration(v.ReadTimeout)
 234         v.bucket = &s3bucket{
 235                 Bucket: &s3.Bucket{
 236                         S3:   client,
 237                         Name: v.Bucket,
 238                 },
 239         }
 240         return nil
 241 }
 242
 243 func (v *S3Volume) getReaderWithContext(ctx context.Context, loc string) (rdr io.ReadCloser, err error) {
 244         ready := make(chan bool)
 245         go func() {
 246                 rdr, err = v.getReader(loc)
 247                 close(ready)
 248         }()
 249         select {
 250         case <-ready:
 251                 return
 252         case <-ctx.Done():
 253                 theConfig.debugLogf("s3: abandoning getReader(): %s", ctx.Err())
 254                 go func() {
 255                         <-ready
 256                         if err == nil {
 257                                 rdr.Close()
 258                         }
 259                 }()
 260                 return nil, ctx.Err()
 261         }
 262 }
 263
 264 // getReader wraps (Bucket)GetReader.
 265 //
 266 // In situations where (Bucket)GetReader would fail because the block
 267 // disappeared in a Trash race, getReader calls fixRace to recover the
 268 // data, and tries again.
 269 func (v *S3Volume) getReader(loc string) (rdr io.ReadCloser, err error) {
 270         rdr, err = v.bucket.GetReader(loc)
 271         err = v.translateError(err)
 272         if err == nil || !os.IsNotExist(err) {
 273                 return
 274         }
 275
 276         _, err = v.bucket.Head("recent/"+loc, nil)
 277         err = v.translateError(err)
 278         if err != nil {
 279                 // If we can't read recent/X, there's no point in
 280                 // trying fixRace. Give up.
 281                 return
 282         }
 283         if !v.fixRace(loc) {
 284                 err = os.ErrNotExist
 285                 return
 286         }
 287
 288         rdr, err = v.bucket.GetReader(loc)
 289         if err != nil {
 290                 log.Printf("warning: reading %s after successful fixRace: %s", loc, err)
 291                 err = v.translateError(err)
 292         }
 293         return
 294 }
 295
 296 // Get a block: copy the block data into buf, and return the number of
 297 // bytes copied.
 298 func (v *S3Volume) Get(ctx context.Context, loc string, buf []byte) (int, error) {
 299         rdr, err := v.getReaderWithContext(ctx, loc)
 300         if err != nil {
 301                 return 0, err
 302         }
 303
 304         var n int
 305         ready := make(chan bool)
 306         go func() {
 307                 defer close(ready)
 308
 309                 defer rdr.Close()
 310                 n, err = io.ReadFull(rdr, buf)
 311
 312                 switch err {
 313                 case nil, io.EOF, io.ErrUnexpectedEOF:
 314                         err = nil
 315                 default:
 316                         err = v.translateError(err)
 317                 }
 318         }()
 319         select {
 320         case <-ctx.Done():
 321                 theConfig.debugLogf("s3: interrupting ReadFull() with Close() because %s", ctx.Err())
 322                 rdr.Close()
 323                 // Must wait for ReadFull to return, to ensure it
 324                 // doesn't write to buf after we return.
 325                 theConfig.debugLogf("s3: waiting for ReadFull() to fail")
 326                 <-ready
 327                 return 0, ctx.Err()
 328         case <-ready:
 329                 return n, err
 330         }
 331 }
 332
 333 // Compare the given data with the stored data.
 334 func (v *S3Volume) Compare(ctx context.Context, loc string, expect []byte) error {
 335         rdr, err := v.getReaderWithContext(ctx, loc)
 336         if err != nil {
 337                 return err
 338         }
 339         defer rdr.Close()
 340         return v.translateError(compareReaderWithBuf(ctx, rdr, expect, loc[:32]))
 341 }
 342
 343 // Put writes a block.
 344 func (v *S3Volume) Put(ctx context.Context, loc string, block []byte) error {
 345         if v.ReadOnly {
 346                 return MethodDisabledError
 347         }
 348         var opts s3.Options
 349         size := len(block)
 350         if size > 0 {
 351                 md5, err := hex.DecodeString(loc)
 352                 if err != nil {
 353                         return err
 354                 }
 355                 opts.ContentMD5 = base64.StdEncoding.EncodeToString(md5)
 356         }
 357
 358         // Send the block data through a pipe, so that (if we need to)
 359         // we can close the pipe early and abandon our PutReader()
 360         // goroutine, without worrying about PutReader() accessing our
 361         // block buffer after we release it.
 362         bufr, bufw := io.Pipe()
 363         go func() {
 364                 io.Copy(bufw, bytes.NewReader(block))
 365                 bufw.Close()
 366         }()
 367
 368         var err error
 369         ready := make(chan bool)
 370         go func() {
 371                 defer func() {
 372                         if ctx.Err() != nil {
 373                                 theConfig.debugLogf("%s: abandoned PutReader goroutine finished with err: %s", v, err)
 374                         }
 375                 }()
 376                 defer close(ready)
 377                 err = v.bucket.PutReader(loc, bufr, int64(size), "application/octet-stream", s3ACL, opts)
 378                 if err != nil {
 379                         return
 380                 }
 381                 err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
 382         }()
 383         select {
 384         case <-ctx.Done():
 385                 theConfig.debugLogf("%s: taking PutReader's input away: %s", v, ctx.Err())
 386                 // Our pipe might be stuck in Write(), waiting for
 387                 // io.Copy() to read. If so, un-stick it. This means
 388                 // PutReader will get corrupt data, but that's OK: the
 389                 // size and MD5 won't match, so the write will fail.
 390                 go io.Copy(ioutil.Discard, bufr)
 391                 // CloseWithError() will return once pending I/O is done.
 392                 bufw.CloseWithError(ctx.Err())
 393                 theConfig.debugLogf("%s: abandoning PutReader goroutine", v)
 394                 return ctx.Err()
 395         case <-ready:
 396                 return v.translateError(err)
 397         }
 398 }
 399
 400 // Touch sets the timestamp for the given locator to the current time.
 401 func (v *S3Volume) Touch(loc string) error {
 402         if v.ReadOnly {
 403                 return MethodDisabledError
 404         }
 405         _, err := v.bucket.Head(loc, nil)
 406         err = v.translateError(err)
 407         if os.IsNotExist(err) && v.fixRace(loc) {
 408                 // The data object got trashed in a race, but fixRace
 409                 // rescued it.
 410         } else if err != nil {
 411                 return err
 412         }
 413         err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
 414         return v.translateError(err)
 415 }
 416
 417 // Mtime returns the stored timestamp for the given locator.
 418 func (v *S3Volume) Mtime(loc string) (time.Time, error) {
 419         _, err := v.bucket.Head(loc, nil)
 420         if err != nil {
 421                 return zeroTime, v.translateError(err)
 422         }
 423         resp, err := v.bucket.Head("recent/"+loc, nil)
 424         err = v.translateError(err)
 425         if os.IsNotExist(err) {
 426                 // The data object X exists, but recent/X is missing.
 427                 err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
 428                 if err != nil {
 429                         log.Printf("error: creating %q: %s", "recent/"+loc, err)
 430                         return zeroTime, v.translateError(err)
 431                 }
 432                 log.Printf("info: created %q to migrate existing block to new storage scheme", "recent/"+loc)
 433                 resp, err = v.bucket.Head("recent/"+loc, nil)
 434                 if err != nil {
 435                         log.Printf("error: created %q but HEAD failed: %s", "recent/"+loc, err)
 436                         return zeroTime, v.translateError(err)
 437                 }
 438         } else if err != nil {
 439                 // HEAD recent/X failed for some other reason.
 440                 return zeroTime, err
 441         }
 442         return v.lastModified(resp)
 443 }
 444
 445 // IndexTo writes a complete list of locators with the given prefix
 446 // for which Get() can retrieve data.
 447 func (v *S3Volume) IndexTo(prefix string, writer io.Writer) error {
 448         // Use a merge sort to find matching sets of X and recent/X.
 449         dataL := s3Lister{
 450                 Bucket:   v.bucket.Bucket,
 451                 Prefix:   prefix,
 452                 PageSize: v.IndexPageSize,
 453         }
 454         recentL := s3Lister{
 455                 Bucket:   v.bucket.Bucket,
 456                 Prefix:   "recent/" + prefix,
 457                 PageSize: v.IndexPageSize,
 458         }
 459         v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
 460         v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
 461         for data, recent := dataL.First(), recentL.First(); data != nil; data = dataL.Next() {
 462                 v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
 463                 if data.Key >= "g" {
 464                         // Conveniently, "recent/*" and "trash/*" are
 465                         // lexically greater than all hex-encoded data
 466                         // hashes, so stopping here avoids iterating
 467                         // over all of them needlessly with dataL.
 468                         break
 469                 }
 470                 if !v.isKeepBlock(data.Key) {
 471                         continue
 472                 }
 473
 474                 // stamp is the list entry we should use to report the
 475                 // last-modified time for this data block: it will be
 476                 // the recent/X entry if one exists, otherwise the
 477                 // entry for the data block itself.
 478                 stamp := data
 479
 480                 // Advance to the corresponding recent/X marker, if any
 481                 for recent != nil {
 482                         if cmp := strings.Compare(recent.Key[7:], data.Key); cmp < 0 {
 483                                 recent = recentL.Next()
 484                                 v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
 485                                 continue
 486                         } else if cmp == 0 {
 487                                 stamp = recent
 488                                 recent = recentL.Next()
 489                                 v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
 490                                 break
 491                         } else {
 492                                 // recent/X marker is missing: we'll
 493                                 // use the timestamp on the data
 494                                 // object.
 495                                 break
 496                         }
 497                 }
 498                 t, err := time.Parse(time.RFC3339, stamp.LastModified)
 499                 if err != nil {
 500                         return err
 501                 }
 502                 fmt.Fprintf(writer, "%s+%d %d\n", data.Key, data.Size, t.UnixNano())
 503         }
 504         return nil
 505 }
 506
 507 // Trash a Keep block.
 508 func (v *S3Volume) Trash(loc string) error {
 509         if v.ReadOnly {
 510                 return MethodDisabledError
 511         }
 512         if t, err := v.Mtime(loc); err != nil {
 513                 return err
 514         } else if time.Since(t) < theConfig.BlobSignatureTTL.Duration() {
 515                 return nil
 516         }
 517         if theConfig.TrashLifetime == 0 {
 518                 if !s3UnsafeDelete {
 519                         return ErrS3TrashDisabled
 520                 }
 521                 return v.translateError(v.bucket.Del(loc))
 522         }
 523         err := v.checkRaceWindow(loc)
 524         if err != nil {
 525                 return err
 526         }
 527         err = v.safeCopy("trash/"+loc, loc)
 528         if err != nil {
 529                 return err
 530         }
 531         return v.translateError(v.bucket.Del(loc))
 532 }
 533
 534 // checkRaceWindow returns a non-nil error if trash/loc is, or might
 535 // be, in the race window (i.e., it's not safe to trash loc).
 536 func (v *S3Volume) checkRaceWindow(loc string) error {
 537         resp, err := v.bucket.Head("trash/"+loc, nil)
 538         err = v.translateError(err)
 539         if os.IsNotExist(err) {
 540                 // OK, trash/X doesn't exist so we're not in the race
 541                 // window
 542                 return nil
 543         } else if err != nil {
 544                 // Error looking up trash/X. We don't know whether
 545                 // we're in the race window
 546                 return err
 547         }
 548         t, err := v.lastModified(resp)
 549         if err != nil {
 550                 // Can't parse timestamp
 551                 return err
 552         }
 553         safeWindow := t.Add(theConfig.TrashLifetime.Duration()).Sub(time.Now().Add(time.Duration(v.RaceWindow)))
 554         if safeWindow <= 0 {
 555                 // We can't count on "touch trash/X" to prolong
 556                 // trash/X's lifetime. The new timestamp might not
 557                 // become visible until now+raceWindow, and EmptyTrash
 558                 // is allowed to delete trash/X before then.
 559                 return fmt.Errorf("same block is already in trash, and safe window ended %s ago", -safeWindow)
 560         }
 561         // trash/X exists, but it won't be eligible for deletion until
 562         // after now+raceWindow, so it's safe to overwrite it.
 563         return nil
 564 }
 565
 566 // safeCopy calls PutCopy, and checks the response to make sure the
 567 // copy succeeded and updated the timestamp on the destination object
 568 // (PutCopy returns 200 OK if the request was received, even if the
 569 // copy failed).
 570 func (v *S3Volume) safeCopy(dst, src string) error {
 571         resp, err := v.bucket.PutCopy(dst, s3ACL, s3.CopyOptions{
 572                 ContentType:       "application/octet-stream",
 573                 MetadataDirective: "REPLACE",
 574         }, v.bucket.Name+"/"+src)
 575         err = v.translateError(err)
 576         if err != nil {
 577                 return err
 578         }
 579         if t, err := time.Parse(time.RFC3339Nano, resp.LastModified); err != nil {
 580                 return fmt.Errorf("PutCopy succeeded but did not return a timestamp: %q: %s", resp.LastModified, err)
 581         } else if time.Now().Sub(t) > maxClockSkew {
 582                 return fmt.Errorf("PutCopy succeeded but returned an old timestamp: %q: %s", resp.LastModified, t)
 583         }
 584         return nil
 585 }
 586
 587 // Get the LastModified header from resp, and parse it as RFC1123 or
 588 // -- if it isn't valid RFC1123 -- as Amazon's variant of RFC1123.
 589 func (v *S3Volume) lastModified(resp *http.Response) (t time.Time, err error) {
 590         s := resp.Header.Get("Last-Modified")
 591         t, err = time.Parse(time.RFC1123, s)
 592         if err != nil && s != "" {
 593                 // AWS example is "Sun, 1 Jan 2006 12:00:00 GMT",
 594                 // which isn't quite "Sun, 01 Jan 2006 12:00:00 GMT"
 595                 // as required by HTTP spec. If it's not a valid HTTP
 596                 // header value, it's probably AWS (or s3test) giving
 597                 // us a nearly-RFC1123 timestamp.
 598                 t, err = time.Parse(nearlyRFC1123, s)
 599         }
 600         return
 601 }
 602
 603 // Untrash moves block from trash back into store
 604 func (v *S3Volume) Untrash(loc string) error {
 605         err := v.safeCopy(loc, "trash/"+loc)
 606         if err != nil {
 607                 return err
 608         }
 609         err = v.bucket.Put("recent/"+loc, nil, "application/octet-stream", s3ACL, s3.Options{})
 610         return v.translateError(err)
 611 }
 612
 613 // Status returns a *VolumeStatus representing the current in-use
 614 // storage capacity and a fake available capacity that doesn't make
 615 // the volume seem full or nearly-full.
 616 func (v *S3Volume) Status() *VolumeStatus {
 617         return &VolumeStatus{
 618                 DeviceNum: 1,
 619                 BytesFree: BlockSize * 1000,
 620                 BytesUsed: 1,
 621         }
 622 }
 623
 624 // InternalStats returns bucket I/O and API call counters.
 625 func (v *S3Volume) InternalStats() interface{} {
 626         return &v.bucket.stats
 627 }
 628
 629 // String implements fmt.Stringer.
 630 func (v *S3Volume) String() string {
 631         return fmt.Sprintf("s3-bucket:%+q", v.Bucket)
 632 }
 633
 634 // Writable returns false if all future Put, Mtime, and Delete calls
 635 // are expected to fail.
 636 func (v *S3Volume) Writable() bool {
 637         return !v.ReadOnly
 638 }
 639
 640 // Replication returns the storage redundancy of the underlying
 641 // device. Configured via command line flag.
 642 func (v *S3Volume) Replication() int {
 643         return v.S3Replication
 644 }
 645
 646 var s3KeepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
 647
 648 func (v *S3Volume) isKeepBlock(s string) bool {
 649         return s3KeepBlockRegexp.MatchString(s)
 650 }
 651
 652 // fixRace(X) is called when "recent/X" exists but "X" doesn't
 653 // exist. If the timestamps on "recent/"+loc and "trash/"+loc indicate
 654 // there was a race between Put and Trash, fixRace recovers from the
 655 // race by Untrashing the block.
 656 func (v *S3Volume) fixRace(loc string) bool {
 657         trash, err := v.bucket.Head("trash/"+loc, nil)
 658         if err != nil {
 659                 if !os.IsNotExist(v.translateError(err)) {
 660                         log.Printf("error: fixRace: HEAD %q: %s", "trash/"+loc, err)
 661                 }
 662                 return false
 663         }
 664         trashTime, err := v.lastModified(trash)
 665         if err != nil {
 666                 log.Printf("error: fixRace: parse %q: %s", trash.Header.Get("Last-Modified"), err)
 667                 return false
 668         }
 669
 670         recent, err := v.bucket.Head("recent/"+loc, nil)
 671         if err != nil {
 672                 log.Printf("error: fixRace: HEAD %q: %s", "recent/"+loc, err)
 673                 return false
 674         }
 675         recentTime, err := v.lastModified(recent)
 676         if err != nil {
 677                 log.Printf("error: fixRace: parse %q: %s", recent.Header.Get("Last-Modified"), err)
 678                 return false
 679         }
 680
 681         ageWhenTrashed := trashTime.Sub(recentTime)
 682         if ageWhenTrashed >= theConfig.BlobSignatureTTL.Duration() {
 683                 // No evidence of a race: block hasn't been written
 684                 // since it became eligible for Trash. No fix needed.
 685                 return false
 686         }
 687
 688         log.Printf("notice: fixRace: %q: trashed at %s but touched at %s (age when trashed = %s < %s)", loc, trashTime, recentTime, ageWhenTrashed, theConfig.BlobSignatureTTL)
 689         log.Printf("notice: fixRace: copying %q to %q to recover from race between Put/Touch and Trash", "recent/"+loc, loc)
 690         err = v.safeCopy(loc, "trash/"+loc)
 691         if err != nil {
 692                 log.Printf("error: fixRace: %s", err)
 693                 return false
 694         }
 695         return true
 696 }
 697
 698 func (v *S3Volume) translateError(err error) error {
 699         switch err := err.(type) {
 700         case *s3.Error:
 701                 if (err.StatusCode == http.StatusNotFound && err.Code == "NoSuchKey") ||
 702                         strings.Contains(err.Error(), "Not Found") {
 703                         return os.ErrNotExist
 704                 }
 705                 // Other 404 errors like NoSuchVersion and
 706                 // NoSuchBucket are different problems which should
 707                 // get called out downstream, so we don't convert them
 708                 // to os.ErrNotExist.
 709         }
 710         return err
 711 }
 712
 713 // EmptyTrash looks for trashed blocks that exceeded TrashLifetime
 714 // and deletes them from the volume.
 715 func (v *S3Volume) EmptyTrash() {
 716         var bytesInTrash, blocksInTrash, bytesDeleted, blocksDeleted int64
 717
 718         // Use a merge sort to find matching sets of trash/X and recent/X.
 719         trashL := s3Lister{
 720                 Bucket:   v.bucket.Bucket,
 721                 Prefix:   "trash/",
 722                 PageSize: v.IndexPageSize,
 723         }
 724         // Define "ready to delete" as "...when EmptyTrash started".
 725         startT := time.Now()
 726         for trash := trashL.First(); trash != nil; trash = trashL.Next() {
 727                 loc := trash.Key[6:]
 728                 if !v.isKeepBlock(loc) {
 729                         continue
 730                 }
 731                 bytesInTrash += trash.Size
 732                 blocksInTrash++
 733
 734                 trashT, err := time.Parse(time.RFC3339, trash.LastModified)
 735                 if err != nil {
 736                         log.Printf("warning: %s: EmptyTrash: %q: parse %q: %s", v, trash.Key, trash.LastModified, err)
 737                         continue
 738                 }
 739                 recent, err := v.bucket.Head("recent/"+loc, nil)
 740                 if err != nil && os.IsNotExist(v.translateError(err)) {
 741                         log.Printf("warning: %s: EmptyTrash: found trash marker %q but no %q (%s); calling Untrash", v, trash.Key, "recent/"+loc, err)
 742                         err = v.Untrash(loc)
 743                         if err != nil {
 744                                 log.Printf("error: %s: EmptyTrash: Untrash(%q): %s", v, loc, err)
 745                         }
 746                         continue
 747                 } else if err != nil {
 748                         log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, "recent/"+loc, err)
 749                         continue
 750                 }
 751                 recentT, err := v.lastModified(recent)
 752                 if err != nil {
 753                         log.Printf("warning: %s: EmptyTrash: %q: parse %q: %s", v, "recent/"+loc, recent.Header.Get("Last-Modified"), err)
 754                         continue
 755                 }
 756                 if trashT.Sub(recentT) < theConfig.BlobSignatureTTL.Duration() {
 757                         if age := startT.Sub(recentT); age >= theConfig.BlobSignatureTTL.Duration()-time.Duration(v.RaceWindow) {
 758                                 // recent/loc is too old to protect
 759                                 // loc from being Trashed again during
 760                                 // the raceWindow that starts if we
 761                                 // delete trash/X now.
 762                                 //
 763                                 // Note this means (TrashCheckInterval
 764                                 // < BlobSignatureTTL - raceWindow) is
 765                                 // necessary to avoid starvation.
 766                                 log.Printf("notice: %s: EmptyTrash: detected old race for %q, calling fixRace + Touch", v, loc)
 767                                 v.fixRace(loc)
 768                                 v.Touch(loc)
 769                                 continue
 770                         }
 771                         _, err := v.bucket.Head(loc, nil)
 772                         if os.IsNotExist(err) {
 773                                 log.Printf("notice: %s: EmptyTrash: detected recent race for %q, calling fixRace", v, loc)
 774                                 v.fixRace(loc)
 775                                 continue
 776                         } else if err != nil {
 777                                 log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, loc, err)
 778                                 continue
 779                         }
 780                 }
 781                 if startT.Sub(trashT) < theConfig.TrashLifetime.Duration() {
 782                         continue
 783                 }
 784                 err = v.bucket.Del(trash.Key)
 785                 if err != nil {
 786                         log.Printf("warning: %s: EmptyTrash: deleting %q: %s", v, trash.Key, err)
 787                         continue
 788                 }
 789                 bytesDeleted += trash.Size
 790                 blocksDeleted++
 791
 792                 _, err = v.bucket.Head(loc, nil)
 793                 if os.IsNotExist(err) {
 794                         err = v.bucket.Del("recent/" + loc)
 795                         if err != nil {
 796                                 log.Printf("warning: %s: EmptyTrash: deleting %q: %s", v, "recent/"+loc, err)
 797                         }
 798                 } else if err != nil {
 799                         log.Printf("warning: %s: EmptyTrash: HEAD %q: %s", v, "recent/"+loc, err)
 800                 }
 801         }
 802         if err := trashL.Error(); err != nil {
 803                 log.Printf("error: %s: EmptyTrash: lister: %s", v, err)
 804         }
 805         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)
 806 }
 807
 808 type s3Lister struct {
 809         Bucket     *s3.Bucket
 810         Prefix     string
 811         PageSize   int
 812         nextMarker string
 813         buf        []s3.Key
 814         err        error
 815 }
 816
 817 // First fetches the first page and returns the first item. It returns
 818 // nil if the response is the empty set or an error occurs.
 819 func (lister *s3Lister) First() *s3.Key {
 820         lister.getPage()
 821         return lister.pop()
 822 }
 823
 824 // Next returns the next item, fetching the next page if necessary. It
 825 // returns nil if the last available item has already been fetched, or
 826 // an error occurs.
 827 func (lister *s3Lister) Next() *s3.Key {
 828         if len(lister.buf) == 0 && lister.nextMarker != "" {
 829                 lister.getPage()
 830         }
 831         return lister.pop()
 832 }
 833
 834 // Return the most recent error encountered by First or Next.
 835 func (lister *s3Lister) Error() error {
 836         return lister.err
 837 }
 838
 839 func (lister *s3Lister) getPage() {
 840         resp, err := lister.Bucket.List(lister.Prefix, "", lister.nextMarker, lister.PageSize)
 841         lister.nextMarker = ""
 842         if err != nil {
 843                 lister.err = err
 844                 return
 845         }
 846         if resp.IsTruncated {
 847                 lister.nextMarker = resp.NextMarker
 848         }
 849         lister.buf = make([]s3.Key, 0, len(resp.Contents))
 850         for _, key := range resp.Contents {
 851                 if !strings.HasPrefix(key.Key, lister.Prefix) {
 852                         log.Printf("warning: s3Lister: S3 Bucket.List(prefix=%q) returned key %q", lister.Prefix, key.Key)
 853                         continue
 854                 }
 855                 lister.buf = append(lister.buf, key)
 856         }
 857 }
 858
 859 func (lister *s3Lister) pop() (k *s3.Key) {
 860         if len(lister.buf) > 0 {
 861                 k = &lister.buf[0]
 862                 lister.buf = lister.buf[1:]
 863         }
 864         return
 865 }
 866
 867 // s3bucket wraps s3.bucket and counts I/O and API usage stats.
 868 type s3bucket struct {
 869         *s3.Bucket
 870         stats s3bucketStats
 871 }
 872
 873 func (b *s3bucket) GetReader(path string) (io.ReadCloser, error) {
 874         rdr, err := b.Bucket.GetReader(path)
 875         b.stats.tick(&b.stats.Ops, &b.stats.GetOps)
 876         b.stats.tickErr(err)
 877         return NewCountingReader(rdr, b.stats.tickInBytes), err
 878 }
 879
 880 func (b *s3bucket) Head(path string, headers map[string][]string) (*http.Response, error) {
 881         resp, err := b.Bucket.Head(path, headers)
 882         b.stats.tick(&b.stats.Ops, &b.stats.HeadOps)
 883         b.stats.tickErr(err)
 884         return resp, err
 885 }
 886
 887 func (b *s3bucket) PutReader(path string, r io.Reader, length int64, contType string, perm s3.ACL, options s3.Options) error {
 888         err := b.Bucket.PutReader(path, NewCountingReader(r, b.stats.tickOutBytes), length, contType, perm, options)
 889         b.stats.tick(&b.stats.Ops, &b.stats.PutOps)
 890         b.stats.tickErr(err)
 891         return err
 892 }
 893
 894 func (b *s3bucket) Put(path string, data []byte, contType string, perm s3.ACL, options s3.Options) error {
 895         err := b.Bucket.PutReader(path, NewCountingReader(bytes.NewBuffer(data), b.stats.tickOutBytes), int64(len(data)), contType, perm, options)
 896         b.stats.tick(&b.stats.Ops, &b.stats.PutOps)
 897         b.stats.tickErr(err)
 898         return err
 899 }
 900
 901 func (b *s3bucket) Del(path string) error {
 902         err := b.Bucket.Del(path)
 903         b.stats.tick(&b.stats.Ops, &b.stats.DelOps)
 904         b.stats.tickErr(err)
 905         return err
 906 }
 907
 908 type s3bucketStats struct {
 909         Errors   uint64
 910         Ops      uint64
 911         GetOps   uint64
 912         PutOps   uint64
 913         HeadOps  uint64
 914         DelOps   uint64
 915         ListOps  uint64
 916         InBytes  uint64
 917         OutBytes uint64
 918
 919         ErrorCodes map[string]uint64 `json:",omitempty"`
 920
 921         lock sync.Mutex
 922 }
 923
 924 func (s *s3bucketStats) tickInBytes(n uint64) {
 925         atomic.AddUint64(&s.InBytes, n)
 926 }
 927
 928 func (s *s3bucketStats) tickOutBytes(n uint64) {
 929         atomic.AddUint64(&s.OutBytes, n)
 930 }
 931
 932 func (s *s3bucketStats) tick(counters ...*uint64) {
 933         for _, counter := range counters {
 934                 atomic.AddUint64(counter, 1)
 935         }
 936 }
 937
 938 func (s *s3bucketStats) tickErr(err error) {
 939         if err == nil {
 940                 return
 941         }
 942         atomic.AddUint64(&s.Errors, 1)
 943         errStr := fmt.Sprintf("%T", err)
 944         if err, ok := err.(*s3.Error); ok {
 945                 errStr = errStr + fmt.Sprintf(" %d %s", err.StatusCode, err.Code)
 946         }
 947         s.lock.Lock()
 948         if s.ErrorCodes == nil {
 949                 s.ErrorCodes = make(map[string]uint64)
 950         }
 951         s.ErrorCodes[errStr]++
 952         s.lock.Unlock()
 953 }