+ var bytesDeleted, bytesInTrash int64
+ var blocksDeleted, blocksInTrash int64
+
+ doBlob := func(b storage.Blob) {
+ // Check whether the block is flagged as trash
+ if b.Metadata["expires_at"] == "" {
+ return
+ }
+
+ atomic.AddInt64(&blocksInTrash, 1)
+ atomic.AddInt64(&bytesInTrash, b.Properties.ContentLength)
+
+ expiresAt, err := strconv.ParseInt(b.Metadata["expires_at"], 10, 64)
+ if err != nil {
+ log.Printf("EmptyTrash: ParseInt(%v): %v", b.Metadata["expires_at"], err)
+ return
+ }
+
+ if expiresAt > time.Now().Unix() {
+ return
+ }
+
+ err = v.container.DeleteBlob(b.Name, &storage.DeleteBlobOptions{
+ IfMatch: b.Properties.Etag,
+ })
+ if err != nil {
+ log.Printf("EmptyTrash: DeleteBlob(%v): %v", b.Name, err)
+ return
+ }
+ atomic.AddInt64(&blocksDeleted, 1)
+ atomic.AddInt64(&bytesDeleted, b.Properties.ContentLength)
+ }
+
+ var wg sync.WaitGroup
+ todo := make(chan storage.Blob, theConfig.EmptyTrashWorkers)
+ for i := 0; i < 1 || i < theConfig.EmptyTrashWorkers; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for b := range todo {
+ doBlob(b)
+ }
+ }()
+ }
+
+ params := storage.ListBlobsParameters{Include: &storage.IncludeBlobDataset{Metadata: true}}
+ for {
+ resp, err := v.container.ListBlobs(params)
+ if err != nil {
+ log.Printf("EmptyTrash: ListBlobs: %v", err)
+ break
+ }
+ for _, b := range resp.Blobs {
+ todo <- b
+ }
+ if resp.NextMarker == "" {
+ break
+ }
+ params.Marker = resp.NextMarker
+ }
+ close(todo)
+ wg.Wait()
+
+ 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)
+}
+
+// InternalStats returns bucket I/O and API call counters.
+func (v *AzureBlobVolume) InternalStats() interface{} {
+ return &v.container.stats
+}
+
+type azureBlobStats struct {
+ statsTicker
+ Ops uint64
+ GetOps uint64
+ GetRangeOps uint64
+ GetMetadataOps uint64
+ GetPropertiesOps uint64
+ CreateOps uint64
+ SetMetadataOps uint64
+ DelOps uint64
+ ListOps uint64
+}
+
+func (s *azureBlobStats) TickErr(err error) {
+ if err == nil {
+ return
+ }
+ errType := fmt.Sprintf("%T", err)
+ if err, ok := err.(storage.AzureStorageServiceError); ok {
+ errType = errType + fmt.Sprintf(" %d (%s)", err.StatusCode, err.Code)
+ }
+ log.Printf("errType %T, err %s", err, err)
+ s.statsTicker.TickErr(err, errType)
+}
+
+// azureContainer wraps storage.Container in order to count I/O and
+// API usage stats.
+type azureContainer struct {
+ ctr *storage.Container
+ stats azureBlobStats
+}
+
+func (c *azureContainer) Exists() (bool, error) {
+ c.stats.Tick(&c.stats.Ops)
+ ok, err := c.ctr.Exists()
+ c.stats.TickErr(err)
+ return ok, err
+}
+
+func (c *azureContainer) GetBlobMetadata(bname string) (storage.BlobMetadata, error) {
+ c.stats.Tick(&c.stats.Ops, &c.stats.GetMetadataOps)
+ b := c.ctr.GetBlobReference(bname)
+ err := b.GetMetadata(nil)
+ c.stats.TickErr(err)
+ return b.Metadata, err
+}
+
+func (c *azureContainer) GetBlobProperties(bname string) (*storage.BlobProperties, error) {
+ c.stats.Tick(&c.stats.Ops, &c.stats.GetPropertiesOps)
+ b := c.ctr.GetBlobReference(bname)
+ err := b.GetProperties(nil)
+ c.stats.TickErr(err)
+ return &b.Properties, err
+}
+
+func (c *azureContainer) GetBlob(bname string) (io.ReadCloser, error) {
+ c.stats.Tick(&c.stats.Ops, &c.stats.GetOps)
+ b := c.ctr.GetBlobReference(bname)
+ rdr, err := b.Get(nil)
+ c.stats.TickErr(err)
+ return NewCountingReader(rdr, c.stats.TickInBytes), err
+}
+
+func (c *azureContainer) GetBlobRange(bname string, start, end int, opts *storage.GetBlobOptions) (io.ReadCloser, error) {
+ c.stats.Tick(&c.stats.Ops, &c.stats.GetRangeOps)
+ b := c.ctr.GetBlobReference(bname)
+ rdr, err := b.GetRange(&storage.GetBlobRangeOptions{
+ Range: &storage.BlobRange{
+ Start: uint64(start),
+ End: uint64(end),
+ },
+ GetBlobOptions: opts,
+ })
+ c.stats.TickErr(err)
+ return NewCountingReader(rdr, c.stats.TickInBytes), err
+}
+
+// If we give it an io.Reader that doesn't also have a Len() int
+// method, the Azure SDK determines data size by copying the data into
+// a new buffer, which is not a good use of memory.
+type readerWithAzureLen struct {
+ io.Reader
+ len int
+}
+
+// Len satisfies the private lener interface in azure-sdk-for-go.
+func (r *readerWithAzureLen) Len() int {
+ return r.len
+}
+
+func (c *azureContainer) CreateBlockBlobFromReader(bname string, size int, rdr io.Reader, opts *storage.PutBlobOptions) error {
+ c.stats.Tick(&c.stats.Ops, &c.stats.CreateOps)
+ if size != 0 {
+ rdr = &readerWithAzureLen{
+ Reader: NewCountingReader(rdr, c.stats.TickOutBytes),
+ len: size,
+ }
+ }
+ b := c.ctr.GetBlobReference(bname)
+ err := b.CreateBlockBlobFromReader(rdr, opts)
+ c.stats.TickErr(err)
+ return err
+}
+
+func (c *azureContainer) SetBlobMetadata(bname string, m storage.BlobMetadata, opts *storage.SetBlobMetadataOptions) error {
+ c.stats.Tick(&c.stats.Ops, &c.stats.SetMetadataOps)
+ b := c.ctr.GetBlobReference(bname)
+ b.Metadata = m
+ err := b.SetMetadata(opts)
+ c.stats.TickErr(err)
+ return err
+}
+
+func (c *azureContainer) ListBlobs(params storage.ListBlobsParameters) (storage.BlobListResponse, error) {
+ c.stats.Tick(&c.stats.Ops, &c.stats.ListOps)
+ resp, err := c.ctr.ListBlobs(params)
+ c.stats.TickErr(err)
+ return resp, err
+}
+
+func (c *azureContainer) DeleteBlob(bname string, opts *storage.DeleteBlobOptions) error {
+ c.stats.Tick(&c.stats.Ops, &c.stats.DelOps)
+ b := c.ctr.GetBlobReference(bname)
+ err := b.Delete(opts)
+ c.stats.TickErr(err)
+ return err