20690: Remove obsolete docs/comments about wb1.
[arvados.git] / services / keepstore / azure_blob_volume_test.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package keepstore
6
7 import (
8         "bytes"
9         "context"
10         "crypto/md5"
11         "encoding/base64"
12         "encoding/json"
13         "encoding/xml"
14         "flag"
15         "fmt"
16         "io/ioutil"
17         "math/rand"
18         "net"
19         "net/http"
20         "net/http/httptest"
21         "os"
22         "regexp"
23         "sort"
24         "strconv"
25         "strings"
26         "sync"
27         "time"
28
29         "git.arvados.org/arvados.git/sdk/go/arvados"
30         "git.arvados.org/arvados.git/sdk/go/ctxlog"
31         "github.com/Azure/azure-sdk-for-go/storage"
32         "github.com/prometheus/client_golang/prometheus"
33         "github.com/sirupsen/logrus"
34         check "gopkg.in/check.v1"
35 )
36
37 const (
38         // This cannot be the fake account name "devstoreaccount1"
39         // used by Microsoft's Azure emulator: the Azure SDK
40         // recognizes that magic string and changes its behavior to
41         // cater to the Azure SDK's own test suite.
42         fakeAccountName = "fakeaccountname"
43         fakeAccountKey  = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
44 )
45
46 var (
47         azureTestContainer string
48         azureTestDebug     = os.Getenv("ARVADOS_DEBUG") != ""
49 )
50
51 func init() {
52         flag.StringVar(
53                 &azureTestContainer,
54                 "test.azure-storage-container-volume",
55                 "",
56                 "Name of Azure container to use for testing. Do not use a container with real data! Use -azure-storage-account-name and -azure-storage-key-file arguments to supply credentials.")
57 }
58
59 type azBlob struct {
60         Data        []byte
61         Etag        string
62         Metadata    map[string]string
63         Mtime       time.Time
64         Uncommitted map[string][]byte
65 }
66
67 type azStubHandler struct {
68         sync.Mutex
69         logger     logrus.FieldLogger
70         blobs      map[string]*azBlob
71         race       chan chan struct{}
72         didlist503 bool
73 }
74
75 func newAzStubHandler(c *check.C) *azStubHandler {
76         return &azStubHandler{
77                 blobs:  make(map[string]*azBlob),
78                 logger: ctxlog.TestLogger(c),
79         }
80 }
81
82 func (h *azStubHandler) TouchWithDate(container, hash string, t time.Time) {
83         blob, ok := h.blobs[container+"|"+hash]
84         if !ok {
85                 return
86         }
87         blob.Mtime = t
88 }
89
90 func (h *azStubHandler) PutRaw(container, hash string, data []byte) {
91         h.Lock()
92         defer h.Unlock()
93         h.blobs[container+"|"+hash] = &azBlob{
94                 Data:        data,
95                 Mtime:       time.Now(),
96                 Metadata:    make(map[string]string),
97                 Uncommitted: make(map[string][]byte),
98         }
99 }
100
101 func (h *azStubHandler) unlockAndRace() {
102         if h.race == nil {
103                 return
104         }
105         h.Unlock()
106         // Signal caller that race is starting by reading from
107         // h.race. If we get a channel, block until that channel is
108         // ready to receive. If we get nil (or h.race is closed) just
109         // proceed.
110         if c := <-h.race; c != nil {
111                 c <- struct{}{}
112         }
113         h.Lock()
114 }
115
116 var rangeRegexp = regexp.MustCompile(`^bytes=(\d+)-(\d+)$`)
117
118 func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
119         h.Lock()
120         defer h.Unlock()
121         if azureTestDebug {
122                 defer h.logger.Printf("azStubHandler: %+v", r)
123         }
124
125         path := strings.Split(r.URL.Path, "/")
126         container := path[1]
127         hash := ""
128         if len(path) > 2 {
129                 hash = path[2]
130         }
131
132         if err := r.ParseForm(); err != nil {
133                 h.logger.Printf("azStubHandler(%+v): %s", r, err)
134                 rw.WriteHeader(http.StatusBadRequest)
135                 return
136         }
137
138         if (r.Method == "PUT" || r.Method == "POST") && r.Header.Get("Content-Length") == "" {
139                 rw.WriteHeader(http.StatusLengthRequired)
140                 return
141         }
142
143         body, err := ioutil.ReadAll(r.Body)
144         if err != nil {
145                 return
146         }
147
148         type blockListRequestBody struct {
149                 XMLName     xml.Name `xml:"BlockList"`
150                 Uncommitted []string
151         }
152
153         blob, blobExists := h.blobs[container+"|"+hash]
154
155         switch {
156         case r.Method == "PUT" && r.Form.Get("comp") == "":
157                 // "Put Blob" API
158                 if _, ok := h.blobs[container+"|"+hash]; !ok {
159                         // Like the real Azure service, we offer a
160                         // race window during which other clients can
161                         // list/get the new blob before any data is
162                         // committed.
163                         h.blobs[container+"|"+hash] = &azBlob{
164                                 Mtime:       time.Now(),
165                                 Uncommitted: make(map[string][]byte),
166                                 Metadata:    make(map[string]string),
167                                 Etag:        makeEtag(),
168                         }
169                         h.unlockAndRace()
170                 }
171                 metadata := make(map[string]string)
172                 for k, v := range r.Header {
173                         if strings.HasPrefix(strings.ToLower(k), "x-ms-meta-") {
174                                 name := k[len("x-ms-meta-"):]
175                                 metadata[strings.ToLower(name)] = v[0]
176                         }
177                 }
178                 h.blobs[container+"|"+hash] = &azBlob{
179                         Data:        body,
180                         Mtime:       time.Now(),
181                         Uncommitted: make(map[string][]byte),
182                         Metadata:    metadata,
183                         Etag:        makeEtag(),
184                 }
185                 rw.WriteHeader(http.StatusCreated)
186         case r.Method == "PUT" && r.Form.Get("comp") == "block":
187                 // "Put Block" API
188                 if !blobExists {
189                         h.logger.Printf("Got block for nonexistent blob: %+v", r)
190                         rw.WriteHeader(http.StatusBadRequest)
191                         return
192                 }
193                 blockID, err := base64.StdEncoding.DecodeString(r.Form.Get("blockid"))
194                 if err != nil || len(blockID) == 0 {
195                         h.logger.Printf("Invalid blockid: %+q", r.Form.Get("blockid"))
196                         rw.WriteHeader(http.StatusBadRequest)
197                         return
198                 }
199                 blob.Uncommitted[string(blockID)] = body
200                 rw.WriteHeader(http.StatusCreated)
201         case r.Method == "PUT" && r.Form.Get("comp") == "blocklist":
202                 // "Put Block List" API
203                 bl := &blockListRequestBody{}
204                 if err := xml.Unmarshal(body, bl); err != nil {
205                         h.logger.Printf("xml Unmarshal: %s", err)
206                         rw.WriteHeader(http.StatusBadRequest)
207                         return
208                 }
209                 for _, encBlockID := range bl.Uncommitted {
210                         blockID, err := base64.StdEncoding.DecodeString(encBlockID)
211                         if err != nil || len(blockID) == 0 || blob.Uncommitted[string(blockID)] == nil {
212                                 h.logger.Printf("Invalid blockid: %+q", encBlockID)
213                                 rw.WriteHeader(http.StatusBadRequest)
214                                 return
215                         }
216                         blob.Data = blob.Uncommitted[string(blockID)]
217                         blob.Etag = makeEtag()
218                         blob.Mtime = time.Now()
219                         delete(blob.Uncommitted, string(blockID))
220                 }
221                 rw.WriteHeader(http.StatusCreated)
222         case r.Method == "PUT" && r.Form.Get("comp") == "metadata":
223                 // "Set Metadata Headers" API. We don't bother
224                 // stubbing "Get Metadata Headers": AzureBlobVolume
225                 // sets metadata headers only as a way to bump Etag
226                 // and Last-Modified.
227                 if !blobExists {
228                         h.logger.Printf("Got metadata for nonexistent blob: %+v", r)
229                         rw.WriteHeader(http.StatusBadRequest)
230                         return
231                 }
232                 blob.Metadata = make(map[string]string)
233                 for k, v := range r.Header {
234                         if strings.HasPrefix(strings.ToLower(k), "x-ms-meta-") {
235                                 name := k[len("x-ms-meta-"):]
236                                 blob.Metadata[strings.ToLower(name)] = v[0]
237                         }
238                 }
239                 blob.Mtime = time.Now()
240                 blob.Etag = makeEtag()
241         case (r.Method == "GET" || r.Method == "HEAD") && r.Form.Get("comp") == "metadata" && hash != "":
242                 // "Get Blob Metadata" API
243                 if !blobExists {
244                         rw.WriteHeader(http.StatusNotFound)
245                         return
246                 }
247                 for k, v := range blob.Metadata {
248                         rw.Header().Set(fmt.Sprintf("x-ms-meta-%s", k), v)
249                 }
250                 return
251         case (r.Method == "GET" || r.Method == "HEAD") && hash != "":
252                 // "Get Blob" API
253                 if !blobExists {
254                         rw.WriteHeader(http.StatusNotFound)
255                         return
256                 }
257                 data := blob.Data
258                 if rangeSpec := rangeRegexp.FindStringSubmatch(r.Header.Get("Range")); rangeSpec != nil {
259                         b0, err0 := strconv.Atoi(rangeSpec[1])
260                         b1, err1 := strconv.Atoi(rangeSpec[2])
261                         if err0 != nil || err1 != nil || b0 >= len(data) || b1 >= len(data) || b0 > b1 {
262                                 rw.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", len(data)))
263                                 rw.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
264                                 return
265                         }
266                         rw.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", b0, b1, len(data)))
267                         rw.WriteHeader(http.StatusPartialContent)
268                         data = data[b0 : b1+1]
269                 }
270                 rw.Header().Set("Last-Modified", blob.Mtime.Format(time.RFC1123))
271                 rw.Header().Set("Content-Length", strconv.Itoa(len(data)))
272                 if r.Method == "GET" {
273                         if _, err := rw.Write(data); err != nil {
274                                 h.logger.Printf("write %+q: %s", data, err)
275                         }
276                 }
277                 h.unlockAndRace()
278         case r.Method == "DELETE" && hash != "":
279                 // "Delete Blob" API
280                 if !blobExists {
281                         rw.WriteHeader(http.StatusNotFound)
282                         return
283                 }
284                 delete(h.blobs, container+"|"+hash)
285                 rw.WriteHeader(http.StatusAccepted)
286         case r.Method == "GET" && r.Form.Get("comp") == "list" && r.Form.Get("restype") == "container":
287                 // "List Blobs" API
288                 if !h.didlist503 {
289                         h.didlist503 = true
290                         rw.WriteHeader(http.StatusServiceUnavailable)
291                         return
292                 }
293                 prefix := container + "|" + r.Form.Get("prefix")
294                 marker := r.Form.Get("marker")
295
296                 maxResults := 2
297                 if n, err := strconv.Atoi(r.Form.Get("maxresults")); err == nil && n >= 1 && n <= 5000 {
298                         maxResults = n
299                 }
300
301                 resp := storage.BlobListResponse{
302                         Marker:     marker,
303                         NextMarker: "",
304                         MaxResults: int64(maxResults),
305                 }
306                 var hashes sort.StringSlice
307                 for k := range h.blobs {
308                         if strings.HasPrefix(k, prefix) {
309                                 hashes = append(hashes, k[len(container)+1:])
310                         }
311                 }
312                 hashes.Sort()
313                 for _, hash := range hashes {
314                         if len(resp.Blobs) == maxResults {
315                                 resp.NextMarker = hash
316                                 break
317                         }
318                         if len(resp.Blobs) > 0 || marker == "" || marker == hash {
319                                 blob := h.blobs[container+"|"+hash]
320                                 bmeta := map[string]string(nil)
321                                 if r.Form.Get("include") == "metadata" {
322                                         bmeta = blob.Metadata
323                                 }
324                                 b := storage.Blob{
325                                         Name: hash,
326                                         Properties: storage.BlobProperties{
327                                                 LastModified:  storage.TimeRFC1123(blob.Mtime),
328                                                 ContentLength: int64(len(blob.Data)),
329                                                 Etag:          blob.Etag,
330                                         },
331                                         Metadata: bmeta,
332                                 }
333                                 resp.Blobs = append(resp.Blobs, b)
334                         }
335                 }
336                 buf, err := xml.Marshal(resp)
337                 if err != nil {
338                         h.logger.Error(err)
339                         rw.WriteHeader(http.StatusInternalServerError)
340                 }
341                 rw.Write(buf)
342         default:
343                 h.logger.Printf("azStubHandler: not implemented: %+v Body:%+q", r, body)
344                 rw.WriteHeader(http.StatusNotImplemented)
345         }
346 }
347
348 // azStubDialer is a net.Dialer that notices when the Azure driver
349 // tries to connect to "devstoreaccount1.blob.127.0.0.1:46067", and
350 // in such cases transparently dials "127.0.0.1:46067" instead.
351 type azStubDialer struct {
352         logger logrus.FieldLogger
353         net.Dialer
354 }
355
356 var localHostPortRe = regexp.MustCompile(`(127\.0\.0\.1|localhost|\[::1\]):\d+`)
357
358 func (d *azStubDialer) Dial(network, address string) (net.Conn, error) {
359         if hp := localHostPortRe.FindString(address); hp != "" {
360                 if azureTestDebug {
361                         d.logger.Debug("azStubDialer: dial", hp, "instead of", address)
362                 }
363                 address = hp
364         }
365         return d.Dialer.Dial(network, address)
366 }
367
368 type TestableAzureBlobVolume struct {
369         *AzureBlobVolume
370         azHandler *azStubHandler
371         azStub    *httptest.Server
372         t         TB
373 }
374
375 func (s *StubbedAzureBlobSuite) newTestableAzureBlobVolume(t TB, cluster *arvados.Cluster, volume arvados.Volume, metrics *volumeMetricsVecs) *TestableAzureBlobVolume {
376         azHandler := newAzStubHandler(t.(*check.C))
377         azStub := httptest.NewServer(azHandler)
378
379         var azClient storage.Client
380         var err error
381
382         container := azureTestContainer
383         if container == "" {
384                 // Connect to stub instead of real Azure storage service
385                 stubURLBase := strings.Split(azStub.URL, "://")[1]
386                 if azClient, err = storage.NewClient(fakeAccountName, fakeAccountKey, stubURLBase, storage.DefaultAPIVersion, false); err != nil {
387                         t.Fatal(err)
388                 }
389                 container = "fakecontainername"
390         } else {
391                 // Connect to real Azure storage service
392                 if azClient, err = storage.NewBasicClient(os.Getenv("ARVADOS_TEST_AZURE_ACCOUNT_NAME"), os.Getenv("ARVADOS_TEST_AZURE_ACCOUNT_KEY")); err != nil {
393                         t.Fatal(err)
394                 }
395         }
396         azClient.Sender = &singleSender{}
397
398         bs := azClient.GetBlobService()
399         v := &AzureBlobVolume{
400                 ContainerName:        container,
401                 WriteRaceInterval:    arvados.Duration(time.Millisecond),
402                 WriteRacePollTime:    arvados.Duration(time.Nanosecond),
403                 ListBlobsMaxAttempts: 2,
404                 ListBlobsRetryDelay:  arvados.Duration(time.Millisecond),
405                 azClient:             azClient,
406                 container:            &azureContainer{ctr: bs.GetContainerReference(container)},
407                 cluster:              cluster,
408                 volume:               volume,
409                 logger:               ctxlog.TestLogger(t),
410                 metrics:              metrics,
411         }
412         if err = v.check(); err != nil {
413                 t.Fatal(err)
414         }
415
416         return &TestableAzureBlobVolume{
417                 AzureBlobVolume: v,
418                 azHandler:       azHandler,
419                 azStub:          azStub,
420                 t:               t,
421         }
422 }
423
424 var _ = check.Suite(&StubbedAzureBlobSuite{})
425
426 type StubbedAzureBlobSuite struct {
427         origHTTPTransport http.RoundTripper
428 }
429
430 func (s *StubbedAzureBlobSuite) SetUpTest(c *check.C) {
431         s.origHTTPTransport = http.DefaultTransport
432         http.DefaultTransport = &http.Transport{
433                 Dial: (&azStubDialer{logger: ctxlog.TestLogger(c)}).Dial,
434         }
435 }
436
437 func (s *StubbedAzureBlobSuite) TearDownTest(c *check.C) {
438         http.DefaultTransport = s.origHTTPTransport
439 }
440
441 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeWithGeneric(c *check.C) {
442         DoGenericVolumeTests(c, false, func(t TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume {
443                 return s.newTestableAzureBlobVolume(t, cluster, volume, metrics)
444         })
445 }
446
447 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeConcurrentRanges(c *check.C) {
448         // Test (BlockSize mod azureMaxGetBytes)==0 and !=0 cases
449         for _, b := range []int{2 << 22, 2<<22 - 1} {
450                 DoGenericVolumeTests(c, false, func(t TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume {
451                         v := s.newTestableAzureBlobVolume(t, cluster, volume, metrics)
452                         v.MaxGetBytes = b
453                         return v
454                 })
455         }
456 }
457
458 func (s *StubbedAzureBlobSuite) TestReadonlyAzureBlobVolumeWithGeneric(c *check.C) {
459         DoGenericVolumeTests(c, false, func(c TB, cluster *arvados.Cluster, volume arvados.Volume, logger logrus.FieldLogger, metrics *volumeMetricsVecs) TestableVolume {
460                 return s.newTestableAzureBlobVolume(c, cluster, volume, metrics)
461         })
462 }
463
464 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeRangeFenceposts(c *check.C) {
465         v := s.newTestableAzureBlobVolume(c, testCluster(c), arvados.Volume{Replication: 3}, newVolumeMetricsVecs(prometheus.NewRegistry()))
466         defer v.Teardown()
467
468         for _, size := range []int{
469                 2<<22 - 1, // one <max read
470                 2 << 22,   // one =max read
471                 2<<22 + 1, // one =max read, one <max
472                 2 << 23,   // two =max reads
473                 BlockSize - 1,
474                 BlockSize,
475         } {
476                 data := make([]byte, size)
477                 for i := range data {
478                         data[i] = byte((i + 7) & 0xff)
479                 }
480                 hash := fmt.Sprintf("%x", md5.Sum(data))
481                 err := v.Put(context.Background(), hash, data)
482                 if err != nil {
483                         c.Error(err)
484                 }
485                 gotData := make([]byte, len(data))
486                 gotLen, err := v.Get(context.Background(), hash, gotData)
487                 if err != nil {
488                         c.Error(err)
489                 }
490                 gotHash := fmt.Sprintf("%x", md5.Sum(gotData))
491                 if gotLen != size {
492                         c.Errorf("length mismatch: got %d != %d", gotLen, size)
493                 }
494                 if gotHash != hash {
495                         c.Errorf("hash mismatch: got %s != %s", gotHash, hash)
496                 }
497         }
498 }
499
500 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeCreateBlobRace(c *check.C) {
501         v := s.newTestableAzureBlobVolume(c, testCluster(c), arvados.Volume{Replication: 3}, newVolumeMetricsVecs(prometheus.NewRegistry()))
502         defer v.Teardown()
503
504         var wg sync.WaitGroup
505
506         v.azHandler.race = make(chan chan struct{})
507
508         wg.Add(1)
509         go func() {
510                 defer wg.Done()
511                 err := v.Put(context.Background(), TestHash, TestBlock)
512                 if err != nil {
513                         c.Error(err)
514                 }
515         }()
516         continuePut := make(chan struct{})
517         // Wait for the stub's Put to create the empty blob
518         v.azHandler.race <- continuePut
519         wg.Add(1)
520         go func() {
521                 defer wg.Done()
522                 buf := make([]byte, len(TestBlock))
523                 _, err := v.Get(context.Background(), TestHash, buf)
524                 if err != nil {
525                         c.Error(err)
526                 }
527         }()
528         // Wait for the stub's Get to get the empty blob
529         close(v.azHandler.race)
530         // Allow stub's Put to continue, so the real data is ready
531         // when the volume's Get retries
532         <-continuePut
533         // Wait for Get() and Put() to finish
534         wg.Wait()
535 }
536
537 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeCreateBlobRaceDeadline(c *check.C) {
538         v := s.newTestableAzureBlobVolume(c, testCluster(c), arvados.Volume{Replication: 3}, newVolumeMetricsVecs(prometheus.NewRegistry()))
539         v.AzureBlobVolume.WriteRaceInterval.Set("2s")
540         v.AzureBlobVolume.WriteRacePollTime.Set("5ms")
541         defer v.Teardown()
542
543         v.PutRaw(TestHash, nil)
544
545         buf := new(bytes.Buffer)
546         v.IndexTo("", buf)
547         if buf.Len() != 0 {
548                 c.Errorf("Index %+q should be empty", buf.Bytes())
549         }
550
551         v.TouchWithDate(TestHash, time.Now().Add(-1982*time.Millisecond))
552
553         allDone := make(chan struct{})
554         go func() {
555                 defer close(allDone)
556                 buf := make([]byte, BlockSize)
557                 n, err := v.Get(context.Background(), TestHash, buf)
558                 if err != nil {
559                         c.Error(err)
560                         return
561                 }
562                 if n != 0 {
563                         c.Errorf("Got %+q, expected empty buf", buf[:n])
564                 }
565         }()
566         select {
567         case <-allDone:
568         case <-time.After(time.Second):
569                 c.Error("Get should have stopped waiting for race when block was 2s old")
570         }
571
572         buf.Reset()
573         v.IndexTo("", buf)
574         if !bytes.HasPrefix(buf.Bytes(), []byte(TestHash+"+0")) {
575                 c.Errorf("Index %+q should have %+q", buf.Bytes(), TestHash+"+0")
576         }
577 }
578
579 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeContextCancelGet(c *check.C) {
580         s.testAzureBlobVolumeContextCancel(c, func(ctx context.Context, v *TestableAzureBlobVolume) error {
581                 v.PutRaw(TestHash, TestBlock)
582                 _, err := v.Get(ctx, TestHash, make([]byte, BlockSize))
583                 return err
584         })
585 }
586
587 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeContextCancelPut(c *check.C) {
588         s.testAzureBlobVolumeContextCancel(c, func(ctx context.Context, v *TestableAzureBlobVolume) error {
589                 return v.Put(ctx, TestHash, make([]byte, BlockSize))
590         })
591 }
592
593 func (s *StubbedAzureBlobSuite) TestAzureBlobVolumeContextCancelCompare(c *check.C) {
594         s.testAzureBlobVolumeContextCancel(c, func(ctx context.Context, v *TestableAzureBlobVolume) error {
595                 v.PutRaw(TestHash, TestBlock)
596                 return v.Compare(ctx, TestHash, TestBlock2)
597         })
598 }
599
600 func (s *StubbedAzureBlobSuite) testAzureBlobVolumeContextCancel(c *check.C, testFunc func(context.Context, *TestableAzureBlobVolume) error) {
601         v := s.newTestableAzureBlobVolume(c, testCluster(c), arvados.Volume{Replication: 3}, newVolumeMetricsVecs(prometheus.NewRegistry()))
602         defer v.Teardown()
603         v.azHandler.race = make(chan chan struct{})
604
605         ctx, cancel := context.WithCancel(context.Background())
606         allDone := make(chan struct{})
607         go func() {
608                 defer close(allDone)
609                 err := testFunc(ctx, v)
610                 if err != context.Canceled {
611                         c.Errorf("got %T %q, expected %q", err, err, context.Canceled)
612                 }
613         }()
614         releaseHandler := make(chan struct{})
615         select {
616         case <-allDone:
617                 c.Error("testFunc finished without waiting for v.azHandler.race")
618         case <-time.After(10 * time.Second):
619                 c.Error("timed out waiting to enter handler")
620         case v.azHandler.race <- releaseHandler:
621         }
622
623         cancel()
624
625         select {
626         case <-time.After(10 * time.Second):
627                 c.Error("timed out waiting to cancel")
628         case <-allDone:
629         }
630
631         go func() {
632                 <-releaseHandler
633         }()
634 }
635
636 func (s *StubbedAzureBlobSuite) TestStats(c *check.C) {
637         volume := s.newTestableAzureBlobVolume(c, testCluster(c), arvados.Volume{Replication: 3}, newVolumeMetricsVecs(prometheus.NewRegistry()))
638         defer volume.Teardown()
639
640         stats := func() string {
641                 buf, err := json.Marshal(volume.InternalStats())
642                 c.Check(err, check.IsNil)
643                 return string(buf)
644         }
645
646         c.Check(stats(), check.Matches, `.*"Ops":0,.*`)
647         c.Check(stats(), check.Matches, `.*"Errors":0,.*`)
648
649         loc := "acbd18db4cc2f85cedef654fccc4a4d8"
650         _, err := volume.Get(context.Background(), loc, make([]byte, 3))
651         c.Check(err, check.NotNil)
652         c.Check(stats(), check.Matches, `.*"Ops":[^0],.*`)
653         c.Check(stats(), check.Matches, `.*"Errors":[^0],.*`)
654         c.Check(stats(), check.Matches, `.*"storage\.AzureStorageServiceError 404 \(404 Not Found\)":[^0].*`)
655         c.Check(stats(), check.Matches, `.*"InBytes":0,.*`)
656
657         err = volume.Put(context.Background(), loc, []byte("foo"))
658         c.Check(err, check.IsNil)
659         c.Check(stats(), check.Matches, `.*"OutBytes":3,.*`)
660         c.Check(stats(), check.Matches, `.*"CreateOps":1,.*`)
661
662         _, err = volume.Get(context.Background(), loc, make([]byte, 3))
663         c.Check(err, check.IsNil)
664         _, err = volume.Get(context.Background(), loc, make([]byte, 3))
665         c.Check(err, check.IsNil)
666         c.Check(stats(), check.Matches, `.*"InBytes":6,.*`)
667 }
668
669 func (v *TestableAzureBlobVolume) PutRaw(locator string, data []byte) {
670         v.azHandler.PutRaw(v.ContainerName, locator, data)
671 }
672
673 func (v *TestableAzureBlobVolume) TouchWithDate(locator string, lastPut time.Time) {
674         v.azHandler.TouchWithDate(v.ContainerName, locator, lastPut)
675 }
676
677 func (v *TestableAzureBlobVolume) Teardown() {
678         v.azStub.Close()
679 }
680
681 func (v *TestableAzureBlobVolume) ReadWriteOperationLabelValues() (r, w string) {
682         return "get", "create"
683 }
684
685 func makeEtag() string {
686         return fmt.Sprintf("0x%x", rand.Int63())
687 }