X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/94788532e822ef26b8c9eac7818f03e3a94fb124..380a54a7d97b34119cbaa3bee05d6b6cd241eee5:/services/keepstore/azure_blob_volume_test.go diff --git a/services/keepstore/azure_blob_volume_test.go b/services/keepstore/azure_blob_volume_test.go index a240c23e16..8d02def144 100644 --- a/services/keepstore/azure_blob_volume_test.go +++ b/services/keepstore/azure_blob_volume_test.go @@ -1,17 +1,24 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( "bytes" + "context" + "crypto/md5" "encoding/base64" + "encoding/json" "encoding/xml" "flag" "fmt" "io/ioutil" - "log" "math/rand" "net" "net/http" "net/http/httptest" + "os" "regexp" "sort" "strconv" @@ -20,16 +27,26 @@ import ( "testing" "time" - "github.com/curoverse/azure-sdk-for-go/storage" + "git.curoverse.com/arvados.git/sdk/go/arvados" + "github.com/Azure/azure-sdk-for-go/storage" + "github.com/ghodss/yaml" + "github.com/prometheus/client_golang/prometheus" + check "gopkg.in/check.v1" ) const ( - // The same fake credentials used by Microsoft's Azure emulator - emulatorAccountName = "devstoreaccount1" - emulatorAccountKey = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==" + // This cannot be the fake account name "devstoreaccount1" + // used by Microsoft's Azure emulator: the Azure SDK + // recognizes that magic string and changes its behavior to + // cater to the Azure SDK's own test suite. + fakeAccountName = "fakeaccountname" + fakeAccountKey = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==" ) -var azureTestContainer string +var ( + azureTestContainer string + azureTestDebug = os.Getenv("ARVADOS_DEBUG") != "" +) func init() { flag.StringVar( @@ -49,8 +66,9 @@ type azBlob struct { type azStubHandler struct { sync.Mutex - blobs map[string]*azBlob - race chan chan struct{} + blobs map[string]*azBlob + race chan chan struct{} + didlist503 bool } func newAzStubHandler() *azStubHandler { @@ -73,6 +91,7 @@ func (h *azStubHandler) PutRaw(container, hash string, data []byte) { h.blobs[container+"|"+hash] = &azBlob{ Data: data, Mtime: time.Now(), + Metadata: make(map[string]string), Uncommitted: make(map[string][]byte), } } @@ -92,10 +111,14 @@ func (h *azStubHandler) unlockAndRace() { h.Lock() } +var rangeRegexp = regexp.MustCompile(`^bytes=(\d+)-(\d+)$`) + func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { h.Lock() defer h.Unlock() - // defer log.Printf("azStubHandler: %+v", r) + if azureTestDebug { + defer log.Printf("azStubHandler: %+v", r) + } path := strings.Split(r.URL.Path, "/") container := path[1] @@ -110,6 +133,11 @@ func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { return } + if (r.Method == "PUT" || r.Method == "POST") && r.Header.Get("Content-Length") == "" { + rw.WriteHeader(http.StatusLengthRequired) + return + } + body, err := ioutil.ReadAll(r.Body) if err != nil { return @@ -133,14 +161,23 @@ func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { h.blobs[container+"|"+hash] = &azBlob{ Mtime: time.Now(), Uncommitted: make(map[string][]byte), + Metadata: make(map[string]string), Etag: makeEtag(), } h.unlockAndRace() } + metadata := make(map[string]string) + for k, v := range r.Header { + if strings.HasPrefix(strings.ToLower(k), "x-ms-meta-") { + name := k[len("x-ms-meta-"):] + metadata[strings.ToLower(name)] = v[0] + } + } h.blobs[container+"|"+hash] = &azBlob{ Data: body, Mtime: time.Now(), Uncommitted: make(map[string][]byte), + Metadata: metadata, Etag: makeEtag(), } rw.WriteHeader(http.StatusCreated) @@ -193,22 +230,46 @@ func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { blob.Metadata = make(map[string]string) for k, v := range r.Header { if strings.HasPrefix(strings.ToLower(k), "x-ms-meta-") { - blob.Metadata[k] = v[0] + name := k[len("x-ms-meta-"):] + blob.Metadata[strings.ToLower(name)] = v[0] } } blob.Mtime = time.Now() blob.Etag = makeEtag() + case (r.Method == "GET" || r.Method == "HEAD") && r.Form.Get("comp") == "metadata" && hash != "": + // "Get Blob Metadata" API + if !blobExists { + rw.WriteHeader(http.StatusNotFound) + return + } + for k, v := range blob.Metadata { + rw.Header().Set(fmt.Sprintf("x-ms-meta-%s", k), v) + } + return case (r.Method == "GET" || r.Method == "HEAD") && hash != "": // "Get Blob" API if !blobExists { rw.WriteHeader(http.StatusNotFound) return } + data := blob.Data + if rangeSpec := rangeRegexp.FindStringSubmatch(r.Header.Get("Range")); rangeSpec != nil { + b0, err0 := strconv.Atoi(rangeSpec[1]) + b1, err1 := strconv.Atoi(rangeSpec[2]) + if err0 != nil || err1 != nil || b0 >= len(data) || b1 >= len(data) || b0 > b1 { + rw.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", len(data))) + rw.WriteHeader(http.StatusRequestedRangeNotSatisfiable) + return + } + rw.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", b0, b1, len(data))) + rw.WriteHeader(http.StatusPartialContent) + data = data[b0 : b1+1] + } rw.Header().Set("Last-Modified", blob.Mtime.Format(time.RFC1123)) - rw.Header().Set("Content-Length", strconv.Itoa(len(blob.Data))) + rw.Header().Set("Content-Length", strconv.Itoa(len(data))) if r.Method == "GET" { - if _, err := rw.Write(blob.Data); err != nil { - log.Printf("write %+q: %s", blob.Data, err) + if _, err := rw.Write(data); err != nil { + log.Printf("write %+q: %s", data, err) } } h.unlockAndRace() @@ -222,6 +283,11 @@ func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { rw.WriteHeader(http.StatusAccepted) case r.Method == "GET" && r.Form.Get("comp") == "list" && r.Form.Get("restype") == "container": // "List Blobs" API + if !h.didlist503 { + h.didlist503 = true + rw.WriteHeader(http.StatusServiceUnavailable) + return + } prefix := container + "|" + r.Form.Get("prefix") marker := r.Form.Get("marker") @@ -249,14 +315,20 @@ func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { } if len(resp.Blobs) > 0 || marker == "" || marker == hash { blob := h.blobs[container+"|"+hash] - resp.Blobs = append(resp.Blobs, storage.Blob{ + bmeta := map[string]string(nil) + if r.Form.Get("include") == "metadata" { + bmeta = blob.Metadata + } + b := storage.Blob{ Name: hash, Properties: storage.BlobProperties{ - LastModified: blob.Mtime.Format(time.RFC1123), + LastModified: storage.TimeRFC1123(blob.Mtime), ContentLength: int64(len(blob.Data)), Etag: blob.Etag, }, - }) + Metadata: bmeta, + } + resp.Blobs = append(resp.Blobs, b) } } buf, err := xml.Marshal(resp) @@ -282,7 +354,9 @@ var localHostPortRe = regexp.MustCompile(`(127\.0\.0\.1|localhost|\[::1\]):\d+`) func (d *azStubDialer) Dial(network, address string) (net.Conn, error) { if hp := localHostPortRe.FindString(address); hp != "" { - log.Println("azStubDialer: dial", hp, "instead of", address) + if azureTestDebug { + log.Println("azStubDialer: dial", hp, "instead of", address) + } address = hp } return d.Dialer.Dial(network, address) @@ -292,10 +366,10 @@ type TestableAzureBlobVolume struct { *AzureBlobVolume azHandler *azStubHandler azStub *httptest.Server - t *testing.T + t TB } -func NewTestableAzureBlobVolume(t *testing.T, readonly bool, replication int) *TestableAzureBlobVolume { +func NewTestableAzureBlobVolume(t TB, readonly bool, replication int) *TestableAzureBlobVolume { azHandler := newAzStubHandler() azStub := httptest.NewServer(azHandler) @@ -306,7 +380,7 @@ func NewTestableAzureBlobVolume(t *testing.T, readonly bool, replication int) *T // Connect to stub instead of real Azure storage service stubURLBase := strings.Split(azStub.URL, "://")[1] var err error - if azClient, err = storage.NewClient(emulatorAccountName, emulatorAccountKey, stubURLBase, storage.DefaultAPIVersion, false); err != nil { + if azClient, err = storage.NewClient(fakeAccountName, fakeAccountKey, stubURLBase, storage.DefaultAPIVersion, false); err != nil { t.Fatal(err) } container = "fakecontainername" @@ -321,8 +395,18 @@ func NewTestableAzureBlobVolume(t *testing.T, readonly bool, replication int) *T t.Fatal(err) } } - - v := NewAzureBlobVolume(azClient, container, readonly, replication) + azClient.Sender = &singleSender{} + + bs := azClient.GetBlobService() + v := &AzureBlobVolume{ + ContainerName: container, + ReadOnly: readonly, + AzureReplication: replication, + ListBlobsMaxAttempts: 2, + ListBlobsRetryDelay: arvados.Duration(time.Millisecond), + azClient: azClient, + container: &azureContainer{ctr: bs.GetContainerReference(container)}, + } return &TestableAzureBlobVolume{ AzureBlobVolume: v, @@ -332,6 +416,29 @@ func NewTestableAzureBlobVolume(t *testing.T, readonly bool, replication int) *T } } +var _ = check.Suite(&StubbedAzureBlobSuite{}) + +type StubbedAzureBlobSuite struct { + volume *TestableAzureBlobVolume + origHTTPTransport http.RoundTripper +} + +func (s *StubbedAzureBlobSuite) SetUpTest(c *check.C) { + s.origHTTPTransport = http.DefaultTransport + http.DefaultTransport = &http.Transport{ + Dial: (&azStubDialer{}).Dial, + } + azureWriteRaceInterval = time.Millisecond + azureWriteRacePollTime = time.Nanosecond + + s.volume = NewTestableAzureBlobVolume(c, false, 3) +} + +func (s *StubbedAzureBlobSuite) TearDownTest(c *check.C) { + s.volume.Teardown() + http.DefaultTransport = s.origHTTPTransport +} + func TestAzureBlobVolumeWithGeneric(t *testing.T) { defer func(t http.RoundTripper) { http.DefaultTransport = t @@ -341,11 +448,32 @@ func TestAzureBlobVolumeWithGeneric(t *testing.T) { } azureWriteRaceInterval = time.Millisecond azureWriteRacePollTime = time.Nanosecond - DoGenericVolumeTests(t, func(t *testing.T) TestableVolume { + DoGenericVolumeTests(t, func(t TB) TestableVolume { return NewTestableAzureBlobVolume(t, false, azureStorageReplication) }) } +func TestAzureBlobVolumeConcurrentRanges(t *testing.T) { + defer func(b int) { + azureMaxGetBytes = b + }(azureMaxGetBytes) + + defer func(t http.RoundTripper) { + http.DefaultTransport = t + }(http.DefaultTransport) + http.DefaultTransport = &http.Transport{ + Dial: (&azStubDialer{}).Dial, + } + azureWriteRaceInterval = time.Millisecond + azureWriteRacePollTime = time.Nanosecond + // Test (BlockSize mod azureMaxGetBytes)==0 and !=0 cases + for _, azureMaxGetBytes = range []int{2 << 22, 2<<22 - 1} { + DoGenericVolumeTests(t, func(t TB) TestableVolume { + return NewTestableAzureBlobVolume(t, false, azureStorageReplication) + }) + } +} + func TestReadonlyAzureBlobVolumeWithGeneric(t *testing.T) { defer func(t http.RoundTripper) { http.DefaultTransport = t @@ -355,11 +483,54 @@ func TestReadonlyAzureBlobVolumeWithGeneric(t *testing.T) { } azureWriteRaceInterval = time.Millisecond azureWriteRacePollTime = time.Nanosecond - DoGenericVolumeTests(t, func(t *testing.T) TestableVolume { + DoGenericVolumeTests(t, func(t TB) TestableVolume { return NewTestableAzureBlobVolume(t, true, azureStorageReplication) }) } +func TestAzureBlobVolumeRangeFenceposts(t *testing.T) { + defer func(t http.RoundTripper) { + http.DefaultTransport = t + }(http.DefaultTransport) + http.DefaultTransport = &http.Transport{ + Dial: (&azStubDialer{}).Dial, + } + + v := NewTestableAzureBlobVolume(t, false, 3) + defer v.Teardown() + + for _, size := range []int{ + 2<<22 - 1, // one