"regexp"
"strings"
"sync"
+ "sync/atomic"
"time"
"git.curoverse.com/arvados.git/sdk/go/arvados"
ReadOnly bool
UnsafeDelete bool
- bucket *s3.Bucket
+ bucket *s3bucket
startOnce sync.Once
}
client := s3.New(auth, region)
client.ConnectTimeout = time.Duration(v.ConnectTimeout)
client.ReadTimeout = time.Duration(v.ReadTimeout)
- v.bucket = &s3.Bucket{
- S3: client,
- Name: v.Bucket,
+ v.bucket = &s3bucket{
+ Bucket: &s3.Bucket{
+ S3: client,
+ Name: v.Bucket,
+ },
}
return nil
}
if err == nil || !os.IsNotExist(err) {
return
}
+
_, err = v.bucket.Head("recent/"+loc, nil)
err = v.translateError(err)
if err != nil {
err = os.ErrNotExist
return
}
+
rdr, err = v.bucket.GetReader(loc)
if err != nil {
log.Printf("warning: reading %s after successful fixRace: %s", loc, err)
func (v *S3Volume) IndexTo(prefix string, writer io.Writer) error {
// Use a merge sort to find matching sets of X and recent/X.
dataL := s3Lister{
- Bucket: v.bucket,
+ Bucket: v.bucket.Bucket,
Prefix: prefix,
PageSize: v.IndexPageSize,
}
recentL := s3Lister{
- Bucket: v.bucket,
+ Bucket: v.bucket.Bucket,
Prefix: "recent/" + prefix,
PageSize: v.IndexPageSize,
}
+ v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
+ v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
for data, recent := dataL.First(), recentL.First(); data != nil; data = dataL.Next() {
+ v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
if data.Key >= "g" {
// Conveniently, "recent/*" and "trash/*" are
// lexically greater than all hex-encoded data
for recent != nil {
if cmp := strings.Compare(recent.Key[7:], data.Key); cmp < 0 {
recent = recentL.Next()
+ v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
continue
} else if cmp == 0 {
stamp = recent
recent = recentL.Next()
+ v.bucket.stats.tick(&v.bucket.stats.Ops, &v.bucket.stats.ListOps)
break
} else {
// recent/X marker is missing: we'll
if !s3UnsafeDelete {
return ErrS3TrashDisabled
}
- return v.bucket.Del(loc)
+ return v.translateError(v.bucket.Del(loc))
}
err := v.checkRaceWindow(loc)
if err != nil {
}
}
+// InternalStats returns bucket I/O and API call counters.
+func (v *S3Volume) InternalStats() interface{} {
+ return &v.bucket.stats
+}
+
// String implements fmt.Stringer.
func (v *S3Volume) String() string {
- return fmt.Sprintf("s3-bucket:%+q", v.bucket.Name)
+ return fmt.Sprintf("s3-bucket:%+q", v.Bucket)
}
// Writable returns false if all future Put, Mtime, and Delete calls
// Use a merge sort to find matching sets of trash/X and recent/X.
trashL := s3Lister{
- Bucket: v.bucket,
+ Bucket: v.bucket.Bucket,
Prefix: "trash/",
PageSize: v.IndexPageSize,
}
v.fixRace(loc)
v.Touch(loc)
continue
- } else if _, err := v.bucket.Head(loc, nil); os.IsNotExist(err) {
+ }
+ _, err := v.bucket.Head(loc, nil)
+ if os.IsNotExist(err) {
log.Printf("notice: %s: EmptyTrash: detected recent race for %q, calling fixRace", v, loc)
v.fixRace(loc)
continue
}
return
}
+
+// s3bucket wraps s3.bucket and counts I/O and API usage stats.
+type s3bucket struct {
+ *s3.Bucket
+ stats s3bucketStats
+}
+
+func (b *s3bucket) GetReader(path string) (io.ReadCloser, error) {
+ rdr, err := b.Bucket.GetReader(path)
+ b.stats.tick(&b.stats.Ops, &b.stats.GetOps)
+ b.stats.tickErr(err)
+ return NewCountingReader(rdr, b.stats.tickInBytes), err
+}
+
+func (b *s3bucket) Head(path string, headers map[string][]string) (*http.Response, error) {
+ resp, err := b.Bucket.Head(path, headers)
+ b.stats.tick(&b.stats.Ops, &b.stats.HeadOps)
+ b.stats.tickErr(err)
+ return resp, err
+}
+
+func (b *s3bucket) PutReader(path string, r io.Reader, length int64, contType string, perm s3.ACL, options s3.Options) error {
+ err := b.Bucket.PutReader(path, NewCountingReader(r, b.stats.tickOutBytes), length, contType, perm, options)
+ b.stats.tick(&b.stats.Ops, &b.stats.PutOps)
+ b.stats.tickErr(err)
+ return err
+}
+
+func (b *s3bucket) Put(path string, data []byte, contType string, perm s3.ACL, options s3.Options) error {
+ err := b.Bucket.PutReader(path, NewCountingReader(bytes.NewBuffer(data), b.stats.tickOutBytes), int64(len(data)), contType, perm, options)
+ b.stats.tick(&b.stats.Ops, &b.stats.PutOps)
+ b.stats.tickErr(err)
+ return err
+}
+
+func (b *s3bucket) Del(path string) error {
+ err := b.Bucket.Del(path)
+ b.stats.tick(&b.stats.Ops, &b.stats.DelOps)
+ b.stats.tickErr(err)
+ return err
+}
+
+type s3bucketStats struct {
+ Errors uint64
+ Ops uint64
+ GetOps uint64
+ PutOps uint64
+ HeadOps uint64
+ DelOps uint64
+ ListOps uint64
+ InBytes uint64
+ OutBytes uint64
+
+ ErrorCodes map[string]uint64 `json:",omitempty"`
+
+ lock sync.Mutex
+}
+
+func (s *s3bucketStats) tickInBytes(n uint64) {
+ atomic.AddUint64(&s.InBytes, n)
+}
+
+func (s *s3bucketStats) tickOutBytes(n uint64) {
+ atomic.AddUint64(&s.OutBytes, n)
+}
+
+func (s *s3bucketStats) tick(counters ...*uint64) {
+ for _, counter := range counters {
+ atomic.AddUint64(counter, 1)
+ }
+}
+
+func (s *s3bucketStats) tickErr(err error) {
+ if err == nil {
+ return
+ }
+ atomic.AddUint64(&s.Errors, 1)
+ errStr := fmt.Sprintf("%T", err)
+ if err, ok := err.(*s3.Error); ok {
+ errStr = errStr + fmt.Sprintf(" %d %s", err.StatusCode, err.Code)
+ }
+ s.lock.Lock()
+ if s.ErrorCodes == nil {
+ s.ErrorCodes = make(map[string]uint64)
+ }
+ s.ErrorCodes[errStr]++
+ s.lock.Unlock()
+}
projects / arvados.git / blobdiff